Java 类java.util.concurrent.Future 实例源码

@Test
public void testInvokeMainWithMainLauncherWithTwoArgs() throws Exception {
  Future<Boolean> future = this.futures.submit(new Callable<Boolean>() {
    @Override
    public Boolean call() throws Exception {
      Class<?> clazz = MainLauncher.class;
      Method mainMethod = clazz.getMethod("main", String[].class);
      String[] args = new String[] {launchedClass, "arg0", "arg1"};
      mainMethod.invoke(null, new Object[] {args});
      return true;
    }
  });
  systemInMock.provideText("\n");
  assertTrue(future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS));
  assertTrue(flag);
}

@SuppressWarnings("unchecked")
@Override
public <P extends EasyTransRequest<R,E>,E extends EasyTransExecutor,R extends Serializable> Future<R> execute(final Integer callSeq, final P params) {

    MessageRecordContent content = new MessageRecordContent();
    content.setParams(params);
    content.setCallSeq(callSeq);
    transSynchronizer.registerLog(content);
    final PublishResult result = new PublishResult();
    result.setMessageContentId(content.getcId());

    FutureTask<PublishResult> future = new FutureTask<PublishResult>(new Callable<PublishResult>() {
        @Override
        public PublishResult call() throws Exception {
            return result;
        }
    });
    future.run();

    return (Future<R>) future;
}

public void shutdown() {
    running = false;
    interrupt();

    List<Future> cancelFutures = new ArrayList<Future>();
    for (Map.Entry<Long, Future> entry : pendingFuture.entrySet()) {
        if (!entry.getValue().isDone()) {
            logger.warn("WARN ## Task future processId[{}] canceled!", entry.getKey());
            cancelFutures.add(entry.getValue());
        }
    }

    for (Future future : cancelFutures) {
        future.cancel(true);
    }
    pendingFuture.clear();
}

/**
 * Sends a ConBee request to the NCP and waits for the response. The response is correlated with the request and the
 * returned {@link ConBeeFrame} contains the request and response data.
 *
 * @param transaction
 *            Request {@link ConBeeTransaction}
 * @return response {@link ConBeeFrame}
 */
public ConBeeTransaction sendTransaction(ConBeeTransaction transaction) {
    Future<ConBeeFrame> futureResponse = sendRequestAsync(transaction);
    if (futureResponse == null) {
        logger.debug("Error sending ConBee transaction: Future is null");
        return null;
    }

    try {
        futureResponse.get();
        return transaction;
    } catch (InterruptedException | ExecutionException e) {
        logger.debug("Error sending ConBee transaction to listeners: ", e);
    }

    return null;
}

public void run() {
    try {
        Future<Void> future = model.runReadActionWhenReady(new MetadataModelAction<T, Void>() {
            public Void run(T metadata) throws Exception {
                state = State.READING_MODEL;
                fireChange();
                result = action.run(metadata);
                state = State.FINISHED;
                fireChange();
                return null;
            }
        });
        // get any exceptions thrown if the action was run asynchronously
        future.get();
    } catch (Exception e) {
        state = State.FINISHED;
        executionException = new ExecutionException(e.getMessage(), e);
        fireChange();
    }
}

public void sendMessage(MetaVersion ver, MetaWrapper newMeta, MetaCompareResult result) {
    ControlMessage message = new ControlMessage(System.currentTimeMillis(), ControlType.G_META_SYNC_WARNING.toString(), "dbus-appender");

    message.addPayload("datasource", GlobalCache.getDatasource().getDsName());
    message.addPayload("schema", ver.getSchema());
    message.addPayload("tableId", ver.getTableId());
    message.addPayload("table", ver.getTable());
    message.addPayload("before", ver.getMeta());
    message.addPayload("after", newMeta);
    message.addPayload("compare-result", JSON.toJSON(result));
    message.addPayload("version", ver.getVersion());

    String topic = PropertiesHolder.getProperties(Constants.Properties.CONFIGURE, Constants.ConfigureKey.GLOBAL_EVENT_TOPIC);
    ProducerRecord<String, String> record = new ProducerRecord<>(topic, message.getType(), message.toJSONString());
    Future<RecordMetadata> future = producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            logger.error("Send global event error.{}", exception.getMessage());
        }
    });
    try {
        future.get(10000, TimeUnit.MILLISECONDS);
    } catch (Exception e) {
        logger.error(e.getMessage(), e);
    }
}

public void testSimpleRun() throws InterruptedException {
    TestProcess process = new TestProcess(0);
    TestCallable callable = new TestCallable();
    callable.addProcess(process);

    ExecutionDescriptor descriptor = new ExecutionDescriptor();
    ExecutionService service = ExecutionService.newService(
            callable, descriptor, "Test");

    Future<Integer> task = service.run();
    assertNotNull(task);

    process.waitStarted();

    process.destroy();
    process.waitFor();
    assertTrue(process.isFinished());
    assertEquals(0, process.exitValue());
}

@NonNull
public static Future<String> download(
        @NonNull final URL url,
        @NonNull final Callable<Boolean> cancel) {
    return RP.submit(()-> {
        if (cancel.call()) {
            return "";  //NOI18N
        }
        final ProgressHandle handle = ProgressHandle.createHandle(NbBundle.getMessage(DocDownloader.class, "LBL_DownloadingDoc"));
        handle.start();
        try {
            final ByteArrayOutputStream out = new ByteArrayOutputStream();
            try(BufferedInputStream in = new BufferedInputStream(url.openStream())) {
                FileUtil.copy(in, out);
            }
            return cancel.call() ?
                    ""  //NOI18N
                    : new String(out.toByteArray(),"UTF-8");  //NOI18N
        } finally {
            handle.finish();
        }
    });
}

/**
 * {@inheritDoc}
 * <p/>
 * Executes the given tasks, returning the result of one which has
 * completed and cancelling any incomplete tasks.
 * @since org.openide.util 8.2
 */
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
    Parameters.notNull("tasks", tasks); //NOI18N
    CountDownLatch wait = new CountDownLatch(1);
    List<Future<T>> result = new ArrayList<Future<T>>(tasks.size());
    AtomicReference<T> ref = new AtomicReference<T>();
    try {
        for (Callable<T> c : tasks) {
            if (c == null) {
                throw new NullPointerException ("Contains null tasks: " +  //NOI18N
                        tasks);
            }
            Callable<T> delegate = new WaitableCallable<T>(c, ref, wait);
            result.add (submit(delegate));
        }
        wait.await();
    } finally {
        for (Future<T> f : result) {
            RPFutureTask<?> ft = (RPFutureTask<?>) f;
            ft.cancel(true);
        }
    }
    return ref.get();
}

@Test
public void testFutures() {
    final Callable<Integer> callable1 = successCallable(999, 100);
    final Callable<String> callable2 = successCallable("HI", 1000);

    ExecutorService es = deferredManager.getExecutorService();
    Future<Integer> future1 = es.submit(callable1);
    Future<String> future2 = es.submit(callable2);
    final AtomicInteger doneCount = new AtomicInteger();
    deferredManager.when(future1, future2).done(new DoneCallback<MultipleResults>() {
        @Override
        public void onDone(MultipleResults result) {
            Assert.assertEquals(2, result.size());
            Assert.assertEquals(999, result.get(0).getResult());
            Assert.assertEquals("HI", result.get(1).getResult());
            doneCount.incrementAndGet();
        }
    });

    waitForCompletion();
    Assert.assertEquals(1, doneCount.get());
}

public static ExecutorService blocksOnFirstCallFromCallableOfType(final ExecutorService delegate, final Class<? extends Callable> type) {
    return new ForwardingExecutorService() {
        private boolean firstCall = true;

        @Override
        protected ExecutorService delegate() {
            return delegate;
        }

        @Override
        public <T> Future<T> submit(Callable<T> task) {
            if (task.getClass().equals(type) && firstCall) {
                firstCall = false;
                try {
                    return Futures.immediateFuture(task.call());
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
            return super.submit(task);
        }
    };
}

@Transactional
@Async
@Override
public Future<List<TaskDTO>> getTasksOfUser(final Long userId) {
  final CompletableFuture<List<TaskDTO>> future = new CompletableFuture<>();

  final TasksOfUserMessage.Request request = new TasksOfUserMessage.Request(userId);

  PatternsCS.ask(userSupervisorActor, request, Global.TIMEOUT).toCompletableFuture()
      .whenComplete((msg, exc) -> {
        if (exc == null) {
          future.complete(((TasksOfUserMessage.Response) msg).getTasks());
        } else {
          future.completeExceptionally(exc);
        }
      });

  return future;
}

/**
 * Send a request to the server and generate a future for the 
 * eventual reply.  Note that this call can block if there is no active
 * connection while a new connection is re-established or if the maximum
 * number of requests is already pending
 * @param xid the transaction ID for the request
 * @param request the actual request to send
 * @return A {@link Future} for the reply message
 * @throws InterruptedException 
 */
public Future<SyncReply> sendRequest(int xid,
                                        SyncMessage request) 
                                     throws RemoteStoreException {
    ensureConnected();
    RemoteSyncFuture future = new RemoteSyncFuture(xid, 
                                                   connectionGeneration);
    futureMap.put(Integer.valueOf(xid), future);

    if (futureMap.size() > MAX_PENDING_REQUESTS) {
        synchronized (futureNotify) {
            while (futureMap.size() > MAX_PENDING_REQUESTS) {
                try {
                    futureNotify.wait();
                } catch (InterruptedException e) {
                    throw new RemoteStoreException("Could not send request",
                                                   e);
                }
            }
        }
    }
    channel.write(request); 
    return future;
}

@Test
public void verifyConcurrentServiceTicketGeneration() throws Exception {
    final TicketGrantingTicket newTgt = newTGT();
    addTicketInTransaction(newTgt);
    final ExecutorService executor = Executors.newFixedThreadPool(CONCURRENT_SIZE);
    try {
        final List<ServiceTicketGenerator> generators = new ArrayList<>(CONCURRENT_SIZE);
        for (int i = 0; i < CONCURRENT_SIZE; i++) {
            generators.add(new ServiceTicketGenerator(newTgt.getId(), this.jpaTicketRegistry, this.txManager));
        }
        final List<Future<String>> results = executor.invokeAll(generators);
        for (final Future<String> result : results) {
            assertNotNull(result.get());
        }
    } catch (final Exception e) {
        logger.error("testConcurrentServiceTicketGeneration produced an error", e);
        fail("testConcurrentServiceTicketGeneration failed.");
    } finally {
        executor.shutdownNow();
    }
}

private <T> void addListener(final RemotePromise<T> result) {
    result.getAddFuture().addListener(new FutureListener<Boolean>() {

        @Override
        public void operationComplete(io.netty.util.concurrent.Future<Boolean> future) throws Exception {
            if (!future.isSuccess()) {
                result.tryFailure(future.cause());
                return;
            }

            if (!future.getNow()) {
                result.tryFailure(new RejectedExecutionException("Task rejected. ExecutorService is in shutdown state"));
            }

        }
    });
}

private Future<Response> executeAsync(ListenableFuture<Response> future, final HTTPCallback<Response> callback) {
    Futures.addCallback(future, new FutureCallback<Response>() {
        @Override
        public void onSuccess(Response response) {
            callback.success(response);
        }

        @Override
        public void onFailure(Throwable throwable) {
            callback.failure(throwable);
        }
    });

    return callback.getFuture();
}

@Test
public void test() throws Exception {
    List<Message<String>> requests = new CopyOnWriteArrayList<>();

    createChannel(c -> useBackPressure(c)
        .withReceiver(requests::add)
    ).join();

    MessagingChannel<String> sender = createChannel(this::useBackPressure).join().get().forRemotes();

    // Enforce back pressure on sender.
    List<ResponseFuture<String>> futureResponses = requestUpToHighWatermark(sender);

    busyWait("requests received", () -> requests.size() == futureResponses.size());

    assertBackPressureEnabled(sender);

    // Go down to low watermark.
    requests.stream().limit(getLowWatermarkBounds()).forEach(r -> r.reply("ok"));

    busyWait("responses received", () ->
        futureResponses.stream().filter(CompletableFuture::isDone).count() == getLowWatermarkBounds()
    );

    // Check that new request can be processed.
    get(sender.send("last"));

    requests.stream().filter(Message::mustReply).forEach(r -> r.reply("ok"));

    for (Future<?> future : futureResponses) {
        get(future);
    }
}

@Override
public Pair<byte[][], int[]> getChunk(List<Integer> redisIds, String cacheKey, long chunkPos, int chunkSize) throws
                                                       ECFileCacheException {

  List<DecoratedJedisPool> jedisPools = getJedisPools(redisIds);

  byte[][] redisDataArray = new byte[jedisPools.size()][];

  CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(pool);
  List<Future<Integer>> futures = new ArrayList<Future<Integer>>();

  String field = chunkPos + SEP + chunkSize;

  int failCount = 0;
  for (int i = 0; i < jedisPools.size(); ++i) {
    DecoratedJedisPool jedis = jedisPools.get(i);
    if (jedis != null) {
      String key = cacheKey + SEP + i;
      RedisGetChunk redisGetChunk = new RedisGetChunk(jedis, key, field, redisDataArray, i);

      if (!pool.isShutdown()) {
        Future<Integer> future = completionService.submit(redisGetChunk);
        futures.add(future);
      }
    } else {
      failCount++;
    }
  }
  checkRedisResult(completionService, futures, failCount);

  return convertChunk(redisDataArray, chunkSize);
}

final public Future<Void> process() {
    if (!processing.compareAndSet(false, true)) {
        throw new IllegalStateException("Can not re-run the processing");
    }
    Thread t = new Thread(new Runnable() {
        @Override
        public void run() {
            processingTask.run();
        }
    });
    t.setDaemon(true);
    t.start();

    return processingTask;
}

@Test
public void testSynchronous() throws Exception {

    //同步调用
    System.out.println(new CommandHelloWorld("World").execute());
    //异步调用
    Future<String> world = new CommandHelloWorld("World").queue();
    System.out.println(world.get());
}

public static void send(IServer server1,int f_i) {
    Message msg = null;
    try {
        //由于客户端配置的async="true"，我们用异步方式来获取结果，如果是同步方式，直接msg=server1.echoMsg(f_i)即可
        server1.echoMsg(f_i);
        Future<Message> future = RpcContext.getContext().getFuture();
        msg = future.get();
        System.out.println("msg:"+msg.getMsg()+","+msg.getData());
    } catch(Throwable e) {
        e.printStackTrace();
    }
}

public Future<Void> tryFinishProcessing() {
    executorService = Executors.newSingleThreadScheduledExecutor();
    final SettableFuture<Void> settableFuture = SettableFuture.create();

    scheduledFuture = executorService.scheduleAtFixedRate(new CheckFinishedTask(settableFuture), 0, 1, TimeUnit.SECONDS);
    return settableFuture;
}

public Move decision(final Board b) {
    // get maximum move

    final ArrayList<Move> moves = b.getMoves(color);
    if(moves.size() == 0)
        return null;

    Vector<Future<Float>> costs = new Vector<Future<Float>>(moves.size());
    costs.setSize(moves.size());

        ExecutorService exec = Executors.newFixedThreadPool(moves.size());
        try {
            for (int i = 0; i < moves.size(); i++) {
                final Move move = moves.get(i);
                Future<Float> result = exec.submit(new Callable<Float>() {

                    @Override
                    public Float call() {
                        ArrayList<Move> state = new ArrayList<Move>();
                        state.add(move);

                        float tmp = minValue(b, state, Float.NEGATIVE_INFINITY, Float.POSITIVE_INFINITY, 1);
                        return tmp;
                    }
                });
                costs.set(i, result);
            }
        } finally {
            exec.shutdown();
        }

        // max
        int maxi = -1;
    float max = Float.NEGATIVE_INFINITY;
        for(int i = 0; i < moves.size(); i++) {
            float cost;
        try {
            cost = costs.get(i).get();
        } catch (Exception e) {
            try {
                Thread.sleep(300);
            } catch (InterruptedException e1) {
            }
            continue;
        }
            if(cost >= max) {
                if(Math.abs(cost-max) < 0.1) // add a little random element
                    if(rand.nextBoolean())
                        continue;

                max = cost;
                maxi = i;
            }
        }

        return moves.get(maxi);
}

@Test
public void testPostOpWithRedirect() {
  // POST operation with redirect
  Future<String> future1 = contentLengthFuture(redirectResponse);
  Future<String> future2 = contentLengthFuture(errResponse);
  try {
    FSDataOutputStream os = fs.append(p);
    os.write(new byte[]{0});
    os.close();
    Assert.fail();
  } catch (IOException ioe) {} // expected
  Assert.assertEquals("0", getContentLength(future1));
  Assert.assertEquals("chunked", getContentLength(future2));
}

/**
 * Send the device request asynchronously
 *
 * @param clientDeviceRequest the device request to send
 * @return the Future instance that holds the response
 */
public Future<ClientDeviceResponse> sendDeviceRequestAsync(
    ClientDeviceRequest clientDeviceRequest) {

  return executorService.submit(
      new Callable<ClientDeviceResponse>() {
        @Override
        public ClientDeviceResponse call() {
          return sendDeviceRequest(clientDeviceRequest);
        }
      });
}

public void testOpenContextsAfterRejections() throws InterruptedException {
    createIndex("test");
    ensureGreen("test");
    final int docs = scaledRandomIntBetween(20, 50);
    for (int i = 0; i < docs; i++) {
        client().prepareIndex("test", "type", Integer.toString(i)).setSource("field", "value").execute().actionGet();
    }
    IndicesStatsResponse indicesStats = client().admin().indices().prepareStats().execute().actionGet();
    assertThat(indicesStats.getTotal().getSearch().getOpenContexts(), equalTo(0L));
    refresh();

    int numSearches = 10;
    Future<SearchResponse>[] responses = new Future[numSearches];
    SearchType searchType = randomFrom(SearchType.DEFAULT, SearchType.QUERY_THEN_FETCH, SearchType.DFS_QUERY_THEN_FETCH);
    logger.info("search type is {}", searchType);
    for (int i = 0; i < numSearches; i++) {
        responses[i] = client().prepareSearch()
                .setQuery(matchAllQuery())
                .setSearchType(searchType)
                .execute();
    }
    for (int i = 0; i < numSearches; i++) {
        try {
            responses[i].get();
        } catch (Exception t) {
        }
    }
    awaitBusy(() -> client().admin().indices().prepareStats().execute().actionGet().getTotal().getSearch().getOpenContexts() == 0, 1, TimeUnit.SECONDS);
    indicesStats = client().admin().indices().prepareStats().execute().actionGet();
    assertThat(indicesStats.getTotal().getSearch().getOpenContexts(), equalTo(0L));
}

/**
 * Collects the Part ETags for initiating the complete multi-part copy
 * request. This is blocking as it waits until all the upload part threads
 * complete.
 */
private List<PartETag> collectPartETags() {

    final List<PartETag> partETags = new ArrayList<PartETag>();
    for (Future<PartETag> future : futures) {
        try {
            partETags.add(future.get());
        } catch (Exception e) {
            throw new SdkClientException("Unable to copy part: "
                    + e.getCause().getMessage(), e.getCause());
        }
    }
    return partETags;
}

public <T> Future<T> cacheCall(Callable<T> compensableCall,Content content){
    logCtx.getLogCache().cacheLogList(Arrays.asList(content));

    mapCallable.put(compensableCall,new Object[3]);
    CompensableCallerWrapper<T> compensableCallWrapper = new CompensableCallerWrapper<T>(compensableCall);
    Future<T> future = new ExecuteTriggerByGetFuture<T>(compensableCallWrapper);
    return future;
}

@Test
public void testAddResourceRefAddResourceConcurrency() throws Exception {
  startEmptyStore();
  final String key = "key1";
  final String fileName = "foo.jar";
  final String user = "user";
  final ApplicationId id = createAppId(1, 1L);
  // add the resource and add the resource ref at the same time
  ExecutorService exec = Executors.newFixedThreadPool(2);
  final CountDownLatch start = new CountDownLatch(1);
  Callable<String> addKeyTask = new Callable<String>() {
    public String call() throws Exception {
      start.await();
      return store.addResource(key, fileName);
    }
  };
  Callable<String> addAppIdTask = new Callable<String>() {
    public String call() throws Exception {
      start.await();
      return store.addResourceReference(key,
          new SharedCacheResourceReference(id, user));
    }
  };
  Future<String> addAppIdFuture = exec.submit(addAppIdTask);
  Future<String> addKeyFuture = exec.submit(addKeyTask);
  // start them at the same time
  start.countDown();
  // get the results
  String addKeyResult = addKeyFuture.get();
  String addAppIdResult = addAppIdFuture.get();
  assertEquals(fileName, addKeyResult);
  System.out.println("addAppId() result: " + addAppIdResult);
  // it may be null or the fileName depending on the timing
  assertTrue(addAppIdResult == null || addAppIdResult.equals(fileName));
  exec.shutdown();
}

@Test(expected = ExecutionException.class)
public void shouldRetryOnMakeResponseException() throws RestException, ExecutionException, InterruptedException {
    String url = "http://dummy.com/test";

    MockResponse.builder()
            .withURL(url)
            .withMethod(POST)
            .withStatusCode(201)
            .withResponseHeader(ContentType.HEADER_NAME, ContentType.TEXT_PLAIN.toString())
            .withResponseBody(DummyCallbackProcessor.THROW)
            .build();

    final AtomicInteger retries = new AtomicInteger();

    Future<Response> response = RestClient.getDefault().withRetryStrategy(new RetryStrategy() {
        @Override
        public RetryResponse shouldRetry(Request req, Response r, Exception e, int rs) {
            retries.getAndIncrement();
            return new RetryResponse(false, 0);
        }
    }).asyncPost(url);

    LockSupport.parkNanos(2000000);

    assertEquals(1, retries.get());

    response.get();
}

@Async
@Override
public Future<List<EventLoggerDTO>> getEventLogForProcessModelAndSubject(final int processModelId,
    final String subject) {
  final List<EventLogEntry> results =
      eventLogRepository.getEventLogForProcessModelAndSubject(processModelId, subject);
  final List<EventLoggerDTO> eventLog = createEventLoggerDTO(results);
  return new AsyncResult<List<EventLoggerDTO>>(eventLog);
}

private <T> List<Future<T>> submitAll(ExecutorService executor,
    Collection<? extends Callable<T>> tasks) {
  List<Future<T>> result = new ArrayList<Future<T>>(tasks.size());

  for (Callable<T> task : tasks) {
    result.add(executor.submit(task));
  }

  return result;
}

private void scheduleCpuUtilizationTask() {
    if (executor != null) {
        executor.shutdownNow();
        executor = null;
    }

    executor = Executors.newSingleThreadScheduledExecutor();
    @SuppressWarnings("unused") // Prevent downstream linter warnings.
            Future<?> possiblyIgnoredError = executor.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            cpuUtilizationTask();
        }
    }, 0, CPU_STAT_SAMPLE_PERIOD_MS, TimeUnit.MILLISECONDS);
}

/**
 * 从缓存中获取.
 *
 * @param keys keys
 * @return Map<String, Object>
 */
public Map<String, Object> asyncGetMulti(String[] keys) {
    Map<String, Object> map = null;
    Future<Map<String, Object>> f = memcachedClient.asyncGetBulk(keys);
    try {
        map = f.get(SpyMemcachedConstants.DEFAULT_TIMEOUT, SpyMemcachedConstants.DEFAULT_TIMEUNIT);
    } catch (Exception e) {
        f.cancel(false);
    }
    return map;
}

@Test
public void testGetUninterruptedlyNoInterrupt() throws Exception {
    CompletableFuture<String> future = new CompletableFuture<>();

    Future<Object> testFuture = runAsync(() ->
        AsyncUtils.getUninterruptedly(future)
    );

    future.complete("something");

    assertEquals("something", get(testFuture));
}

private void startDelayedDrag(DragEvent dragEvent) {
  Future<?> task = delayedDragService.schedule(
      () -> FxUtils.runOnFxThread(tab::onDragOver),
      DRAG_FOCUS_DELAY, TimeUnit.MILLISECONDS);

  setOnDragExited(__de -> task.cancel(false));
}

@Override
public void scheduleCheck(final EndPoint endPoint) {
    final Future<?> future = scheduledExecutorService.scheduleWithFixedDelay(() -> checkEndPoint(endPoint), 0,
        DEFAULT_CHECK_DELAY, TimeUnit.SECONDS);

    endPointFutures.put(endPoint, future);

    LOGGER.info(String.format("%s scheduled", endPoint));
}

/**
 * Completed submit(runnable) returns successfully
 */
public void testSubmitRunnable() throws Exception {
    ExecutorService e = new DirectExecutorService();
    Future<?> future = e.submit(new NoOpRunnable());
    future.get();
    assertTrue(future.isDone());
}

private void saveToFile() {
    if (!Environment.getExternalStorageState().equals(Environment.MEDIA_MOUNTED)) {
        Toast.makeText(this, R.string.gallery_save_file_not_have_external_storage, Toast.LENGTH_SHORT).show();
        return;
    }

    final Future<File> future = getImageLoader()
            .load(mPath)
            .downloadOnly(Target.SIZE_ORIGINAL, Target.SIZE_ORIGINAL);

    AppOperator.runOnThread(new Runnable() {
        @Override
        public void run() {
            try {
                File sourceFile = future.get();
                if (sourceFile == null || !sourceFile.exists())
                    return;
                String extension = BitmapUtil.getExtension(sourceFile.getAbsolutePath());
                String extDir = Environment.getExternalStoragePublicDirectory(Environment.DIRECTORY_PICTURES)
                        .getAbsolutePath() + File.separator + "开源中国";
                File extDirFile = new File(extDir);
                if (!extDirFile.exists()) {
                    if (!extDirFile.mkdirs()) {
                        // If mk dir error
                        callSaveStatus(false, null);
                        return;
                    }
                }
                final File saveFile = new File(extDirFile, String.format("IMG_%s.%s", System.currentTimeMillis(), extension));
                final boolean isSuccess = StreamUtil.copyFile(sourceFile, saveFile);
                callSaveStatus(isSuccess, saveFile);
            } catch (Exception e) {
                e.printStackTrace();
                callSaveStatus(false, null);
            }
        }
    });
}

private static void writeRequest(AsyncChannelWrapper channel, ByteBuffer request,
        long timeout) throws TimeoutException, InterruptedException, ExecutionException {
    int toWrite = request.limit();

    Future<Integer> fWrite = channel.write(request);
    Integer thisWrite = fWrite.get(timeout, TimeUnit.MILLISECONDS);
    toWrite -= thisWrite.intValue();

    while (toWrite > 0) {
        fWrite = channel.write(request);
        thisWrite = fWrite.get(timeout, TimeUnit.MILLISECONDS);
        toWrite -= thisWrite.intValue();
    }
}