import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
public Promise<Void> restart() {

    Deferred<Void> future = new Deferred<>();
    Thread restartThread = new Thread("Restart-Server-Thread") {
        @Override
        public void run() {
            Server.inMaintenance = true;
            try {
                log.info("Restarting server.");
                boolean successStarted = startNewServer();
                if (!successStarted) {
                    throw new Exception("Fail to start new server process.");
                }
                log.info("Restart server (pid:" + ServerUtil.getCurrentPid() + ")");

                future.resolve(null);
            } catch (Exception ex) {
                future.fail(ex);
            }
        }
    };
    restartThread.start();
    return future.getPromise();
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
/**
 * Put a resolve job onto the queue. The resolve queue is a little
 * different: because there is no need to resolve the same file twice, we
 * check whether there's already a pending resolve job for the given file.
 * If so, just return the same promise.
 */
private Promise<InstallableUnitImpl> putResolveJob(File file) {
    Promise<InstallableUnitImpl> promise;

    Job<InstallableUnitImpl> newJob = new Job<>(() -> {
        InstallableUnitImpl unit = performResolve(file);
        replaceUnit(file, null, unit);
        return unit;
    });
    synchronized (this.processorThread) {
        Job<InstallableUnitImpl> existing = this.pendingResolves.putIfAbsent(file, newJob);
        if (existing != null) {
            // no need to unblock the processor thread because we haven't added anything
            debug("Not adding resolve job for %s as it is already on the queue", file);
            promise = existing.getPromise();
        } else {
            debug("Added resolve job for %s", file);
            promise = newJob.getPromise();
            this.processorThread.notifyAll();
        }
    }
    return promise;
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
/**
 * Unsponsor the installable unit. Returns a promise which is resolved when the uninstall is completed.
 */
private Promise<List<Bundle>> unsponsorUnit(File file) {
    List<InstallableUnitEvent> events = new LinkedList<>();

    Promise<List<Bundle>> bundles = withLock(this.unitsLock.writeLock(), () -> {
        Promise<List<Bundle>> promise = Promises.resolved(Collections.emptyList());
        InstallableUnitImpl existing = this.units.remove(file);
        if (existing != null) {
            State origState = existing.getState();
            if (origState.equals(State.INSTALLED)) {
                promise = putUninstallJob(existing);
            }

            if (existing.setState(State.REMOVED)) {
                events.add(new InstallableUnitEvent(origState, State.REMOVED, existing));
            }
        }
        return promise;
    });

    notifyListeners(events);
    return bundles;
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@SuppressWarnings("unchecked")
protected void doService(SerializationStrategy serializationStrategy, ClassLoader loader, Method method, Object target, DataByteArrayInputStream requestStream, final DataByteArrayOutputStream responseStream, final Runnable onComplete) {

    final AsyncServiceResponse helper = new AsyncServiceResponse(loader, method, responseStream, onComplete, serializationStrategy);
    try {
        Class<?>[] types = method.getParameterTypes();
        final Object[] args = new Object[types.length];
        serializationStrategy.decodeRequest(loader, types, requestStream, args);
        final Promise<Object> promise = (Promise<Object>)method.invoke(target, args);
        promise.onResolve(() -> {
            try{
                helper.send(promise.getFailure(), promise.getFailure()==null ? promise.getValue() : null);
            }
            catch (Exception e){
                helper.send(e, null);
            }
        });

    } catch (Throwable t) {
        helper.send(t, null);
    }
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
public Promise<String> callAsyncPromise(final int delay) {
    final Deferred<String> deferred = new Deferred<String>();
    new Thread(new Runnable() {
        
        @Override
        public void run() {
            if (delay == -1) {
                deferred.fail(new ExpectedTestException());
                return;
            }
            sleep(delay);
            deferred.resolve("Finished");
        }
    }).start();
    
    return deferred.getPromise();
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
public <T, E extends Throwable> T onCall(CallContext callContext, InterceptionHandler<T> handler) throws E {
    Stopwatch started = Stopwatch.createStarted();
    String methodName = callContext.target.getName() + "::" + callContext.method.getName();

    boolean async = (callContext.method.getReturnType() == Promise.class);

    try {
        T result = handler.invoke();
        if (async) {
            Promise<?> p = (Promise<?>) result;
            p.onResolve(() -> System.out
                    .println("Async call to " + methodName + " took " + started.elapsed(TimeUnit.MICROSECONDS)
                            + " µs"));
        }
        return result;
    } finally {
        System.out.println(
                "Sync call to " + methodName + " took " + started.elapsed(TimeUnit.MICROSECONDS) + " µs");
    }
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
public <T, E extends Throwable> T onCall(Timed annotation, CallContext callContext,
        InterceptionHandler<T> handler) {

    Stopwatch started = Stopwatch.createStarted();
    String methodName = callContext.target.getName() + "::" + callContext.method.getName();

    boolean async = (callContext.method.getReturnType() == Promise.class);

    try {
        T result = handler.invoke();
        if (async) {
            Promise<?> p = (Promise<?>) result;
            p.onResolve(() -> System.out
                    .println("Async call to " + methodName + " took " + started.elapsed(TimeUnit.MICROSECONDS)
                            + " µs"));
        }
        return result;
    } finally {
        System.out.println(
                "Sync call to " + methodName + " took " + started.elapsed(TimeUnit.MICROSECONDS) + " µs");
    }
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
public Promise<Long> getLongResult() {
    Deferred<Long> d = new Deferred<>();

    Promise<Long> promise = superSlowService.compute();
    promise.onResolve(() -> {
        new Thread(new Runnable() {

            @Override
            public void run() {
                d.resolveWith(promise);
            }
        }).start();
    });
    Promise<Long> promise2 = d.getPromise();

    return promise2;
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
public Promise<NeuralNetworkSequenceResult> forward(UUID[] inputIds, UUID[] outputIds, List<Tensor>[] inputs, String... tags){
	if(!valid)
		throw new RuntimeException("This neural network object is no longer valid");
	
	// store inputs
	if(inputIds == null){
		sequenceInputs.put(null, inputs[0]);
	} else {
		for(int i=0;i<inputIds.length;i++){
			sequenceInputs.put(inputIds[i], inputs[i]);
		}
	}

	// TODO what if other sequence already executing?!
	return forward(0, inputIds, outputIds, inputs, tags);
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
public Promise<NeuralNetwork> getNeuralNetwork(NeuralNetworkInstanceDTO nni) {
	Deferred<NeuralNetwork> result = new Deferred<>();
	
	NeuralNetwork nn = nnServices.get(nni.id);
	if(nn!=null){
		result.resolve(nn);
		return result.getPromise();
	}
	
	ToWatch w = new ToWatch();
	w.nni = nni;
	w.deferred = result; 
	toWatchFor.put(nni.id, w);
	
	createNeuralNetwork(nni.id);
	
	return result.getPromise();
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
public Promise<EvaluationResult> eval(String dataset, Map<String, String> config, NeuralNetworkDTO... nns) {
	if(nns != null){
		try {
			for(NeuralNetworkDTO nn : nns)
				repository.storeNeuralNetwork(nn);
		} catch(Exception e){
			// NN could be locked but still evaluation should be possible
		}
	}
	
	EvaluationJob job = new EvaluationJob(this, dataset, config, nns);
	queueEval.add(job);
	
	sendNotification(job.jobId, Level.INFO, "Evaluation job \""+job.name+"\" submitted.");
	
	schedule(Type.EVALUATE);
	
	return job.getPromise();
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
public Promise<AgentResult> act(String dataset, Map<String, String> config, NeuralNetworkDTO... nns) {
	if(nns !=null){
		try {
			for(NeuralNetworkDTO nn : nns)
				repository.storeNeuralNetwork(nn);
		} catch(Exception e){
			// NN could be locked but still evaluation should be possible
		}
	}
	
	ActJob job = new ActJob(this, dataset, config, nns);
	queueAct.add(job);
	
	sendNotification(job.jobId, Level.INFO, "Act job \""+job.name+"\" submitted.");
	
	schedule(Type.ACT);
	
	return job.getPromise();
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
public Promise<List<Bundle>> install() {
    if (!this.state.equals(State.RESOLVED)) {
        throw new IllegalStateException(String.format("Cannot install unit %s: not in the %s state", this.symbolicName, State.RESOLVED));
    }
    return this.installer.putInstallJob(this);
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
/**
 * Put an install job onto the queue
 */
Promise<List<Bundle>> putInstallJob(InstallableUnitImpl unit) {
    Job<List<Bundle>> job = new Job<>(() -> installArtifacts(unit));
    synchronized (this.processorThread) {
        this.pendingInstalls.add(job);
        this.processorThread.notifyAll();
    }
    return job.getPromise();
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
/**
 * Put an uninstall job onto the queue
 */
Promise<List<Bundle>> putUninstallJob(InstallableUnitImpl unit) {
    Job<List<Bundle>> job = new Job<>(() -> uninstallArtifacts(unit));
    synchronized (this.processorThread) {
        this.pendingInstalls.add(job);
        this.processorThread.notifyAll();
    }
    return job.getPromise();
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Test
public void testInvokeFutureExceptionally() throws Exception {

    Promise<String> promise = testService.exceptionPromise();
    try{
        promise.getValue();
        fail("Must throw an exception");
    } catch (InvocationTargetException e) {
        assertTrue(e.getTargetException() instanceof IOException);
        assertEquals("test", e.getCause().getMessage());
        assertTrue(promise.getFailure() instanceof IOException);
    }
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
public Promise<String> exceptionPromise() throws IOException {
    final Deferred<String> deferred = new Deferred<String>();
    new Thread(() -> {
        sleep(500);
        deferred.fail(new IOException("test"));
    }).start();
     return deferred.getPromise();
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    if (Future.class.isAssignableFrom(method.getReturnType()) ||
        CompletionStage.class.isAssignableFrom(method.getReturnType())) {
        return createFutureResult(method, args);
    } else if (Promise.class.isAssignableFrom(method.getReturnType())) {
        return createPromiseResult(method, args);
    } else {
        return handleSyncCall(method, args);
    }
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Test(expected = ExpectedTestException.class)
public void testAsyncPromiseException() throws Throwable {
    Promise<String> result = myServiceProxy.callAsyncPromise(-1);
    try {
        result.getValue();
    } catch (InvocationTargetException e) {
        throw e.getCause();
    }
}
 

import org.osgi.util.promise.Promise; //导入依赖的package包/类
@Override
@Timed
public Promise<Long> compute() {
    Deferred<Long> deferred = new Deferred<>();
    executor.submit(() -> {
        Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
        deferred.resolve(42L);
    });
    return deferred.getPromise();
}