import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
public synchronized ListenableFuture<Response> sendRawRequest(Request req) {
  SettableListenableFuture<Response> slf = new SettableListenableFuture<Response>();
  if(isConnected()) {
    try {
      if(log.isDebugEnabled()) {
        log.debug("Sending Message:{}",  JsonFormat.printer().omittingInsignificantWhitespace().print(req));
      }
    } catch (InvalidProtocolBufferException e) {
      throw new RuntimeException(e);
    }
    responseQueue.add(slf);
    client.write(WebSocketFrameParser.makeWebSocketFrame(req.toByteArray().length, WebSocketOpCode.Binary.getValue(), false).getRawFrame());
    client.write(ByteBuffer.wrap(req.toByteArray()));
    return slf;
  }
  slf.setFailure(new IOException("Client is not connected!"));
  return slf;
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
public ListenableFuture<?> connect() {
  connectCalled = true;
  log.info("Connecting to:{}", host);
  ListenableFuture<?> lf = client.connect();
  lf.addCallback(new FutureCallback<Object>() {

    @Override
    public void handleResult(Object result) {
      log.info("Connected to:{}", host);    
    }

    @Override
    public void handleFailure(Throwable t) {
    }
    
  });
  return lf;
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Override
protected void runItem(ExecutionAssistant assistant) {
  try {
    List<? extends ListenableFuture<?>> scriptFutures = assistant.getGlobalRunningFutureSet();
    double doneCount = 0;
    Iterator<? extends ListenableFuture<?>> it = scriptFutures.iterator();
    while (it.hasNext()) {
      if (it.next().isDone()) {
        doneCount++;
      }
    }
    
    slf.setResult((doneCount / scriptFutures.size()) * 100);
  } catch (Exception e) {
    slf.setFailure(e);
  }
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Override
public void markGlobalFailure() {
  if (! markedFailure.get() && markedFailure.compareAndSet(false, true)) {
    synchronized (failureListeners) {
      for (Runnable r :  failureListeners) {
        ExceptionUtils.runRunnable(r);
      }
      failureListeners.clear();
    }
    List<ListenableFuture<StepResult>> futures = this.futures.get();
    if (futures != null) {
      // try to short cut any steps we can
      // Sadly this is a duplicate from other cancels, but since we are not garunteed to be 
      // able to cancel here, we still need those points
      FutureUtils.cancelIncompleteFutures(futures, true);
    }
  }
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Override
public ListenableFuture<?> executeIfStillRunning(ExecutionItem item, boolean forceAsync) {
  // the existence of the scheduler (and possibly limiter) indicate still running
  SubmitterExecutor limiter = this.limiter;
  if (limiter != null && ! item.isChainExecutor()) {
    return limiter.submit(wrapInRunnable(item));
  } else {
    PriorityScheduler scheduler = this.scheduler.get();
    if (scheduler != null) {
      if (forceAsync) {
        ExecuteOnGetFutureTask<?> result = new ExecuteOnGetFutureTask<Void>(wrapInRunnable(item));
        scheduler.execute(result);
        return result;
      } else {
        item.itemReadyForExecution(this);
      }
    }
  }
  return ImmediateResultListenableFuture.NULL_RESULT;
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Test
public void addParallelStepsTest() throws InterruptedException {
  ParallelScriptBuilder pBuilder = new ParallelScriptBuilder();
  pBuilder.addStep(new TestStep(), 10);
  final Collection<? extends ListenableFuture<?>> futures = pBuilder.currentStep.getFutures();
  assertFalse(futures.isEmpty());
  
  sBuilder.addStep(new TestStep());
  sBuilder.addStep(new TestStep());
  sBuilder.addSteps(pBuilder);
  sBuilder.addStep(new TestStep() {
    @Override
    public void handleRunStart() {
      Iterator<? extends ListenableFuture<?>> it = futures.iterator();
      while (it.hasNext()) {
        assertTrue(it.next().isDone());
      }
    }
  });
  
  assertNull(StepResultCollectionUtils.getFailedResult(sBuilder.build().startScript()));
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Test
public void inParallelOnlyTest() throws InterruptedException, ExecutionException {
  String identifier = StringUtils.makeRandomString(5);
  int runTime = 2;
  int runCount = 10;
  TestStep step = new TestStep(identifier, runTime);
  ParallelScriptBuilder builder = new ParallelScriptBuilder();
  builder.addStep(step, runCount);
  
  List<? extends ListenableFuture<StepResult>> futures = builder.build().startScript();
  assertEquals(runCount, futures.size());
  
  StepResult tr = futures.get(0).get();
  
  assertNull(tr.getError());
  assertEquals(identifier, tr.getDescription());
  assertTrue(tr.getRunTime(TimeUnit.MILLISECONDS) >= runTime);
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Test
public void inParallelSequenceChainsTest() throws InterruptedException, TimeoutException {
  final List<TestStep> steps1 = makeTestSteps(null, TEST_COMPLEXITY);
  final List<TestStep> steps2 = makeTestSteps(null, TEST_COMPLEXITY);
  ParallelScriptBuilder pBuilder = new ParallelScriptBuilder();
  SequentialScriptBuilder sBuilder1 = new SequentialScriptBuilder();
  addSteps(steps1, sBuilder1);
  SequentialScriptBuilder sBuilder2 = new SequentialScriptBuilder();
  addSteps(steps2, sBuilder2);
  pBuilder.addSteps(sBuilder1);
  pBuilder.addSteps(sBuilder2);
  
  assertEquals(4, pBuilder.getMaximumNeededThreadCount());
  List<? extends ListenableFuture<StepResult>> futures = pBuilder.build().startScript();
  assertEquals(TEST_COMPLEXITY * 2, futures.size());
  
  FutureUtils.blockTillAllComplete(futures, 10 * 1000);
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Test
public void addProgressFutureTest() throws InterruptedException, TimeoutException {
  SequentialScriptBuilder builder = new SequentialScriptBuilder();
  builder.addStep(new TestStep());
  builder.addStep(new TestStep());
  ListenableFuture<Double> future = builder.addProgressFuture();
  builder.addStep(new TestStep());
  builder.addStep(new TestStep());
  
  builder.build().startScript();
  
  final AsyncVerifier av = new AsyncVerifier();
  future.addCallback(new FutureCallback<Double>() {
    @Override
    public void handleResult(Double result) {
      av.assertEquals(50., result);
      av.signalComplete();
    }

    @Override
    public void handleFailure(Throwable t) {
      av.fail(t);
    }
  });
  av.waitForTest();
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
/**
 * This performs a write to the websocket connection.  This write will use the provided mask and OpCode values, ignoring the 
 * defaults.
 * 
 * Every {@link ByteBuffer} written is seen as an individual websocketFrame.
 * 
 * @param bb the {@link ByteBuffer} to write to frame and write to the websocket.
 * @param opCode the opCode to use in the websocket frame.
 * @param mask sets whether or not to mask the websocket data. true to mask, false to not.
 * @return a {@link ListenableFuture} that will be completed once the frame has been fully written to the socket.
 */
public ListenableFuture<?> write(final ByteBuffer bb, final byte opCode, final boolean mask) {
  if(connectFuture.isDone()) {
    WebSocketFrame wsFrame = WebSocketFrameParser.makeWebSocketFrame(bb.remaining(), opCode, mask);
    ByteBuffer data = bb;
    if(mask) {
      data = wsFrame.unmaskPayload(bb);
    }
    synchronized(this) {
      hsc.write(wsFrame.getRawFrame());
      return hsc.write(data);
    }
  } else {
    throw new IllegalStateException("Must be connected first!");
  }
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
/**
 * Sends an asynchronous HTTP request.
 * 
 * @param url the {@link URL} to send the request too.
 * @param rt the {@link HTTPRequestType} to use on the request.
 * @param bb the data to put in the body for this request.
 * @return an {@link ListenableFuture} containing a {@link HTTPResponseData} object that will be completed when the request is finished, 
 * successfully or with errors.
 */
public ListenableFuture<HTTPResponseData> requestAsync(final URL url, final HTTPRequestType rt, final ByteBuffer bb) {
  boolean ssl = false;
  int port = HTTPConstants.DEFAULT_HTTP_PORT;
  String host = url.getHost();
  if(url.getProtocol().equalsIgnoreCase("https")) {
    port = HTTPConstants.DEFAULT_HTTPS_PORT;
    ssl = true;
  }
  if(url.getPort() != -1) {
    port = url.getPort();
  }
  HTTPRequestBuilder hrb = new HTTPRequestBuilder(url);
  hrb.setRequestType(rt);
  return requestAsync(new HTTPAddress(host, port, ssl), hrb.build(), bb);
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
/**
 * Constructs a new parser for a given file.
 * 
 * @param executor Executor that computation can be threaded out to
 * @param hprofFile File that should be parsed
 */
public HprofParser(SubmitterExecutorInterface executor, File hprofFile) {
  ArgumentVerifier.assertNotNull(hprofFile, "hprofFile");
  if (! hprofFile.exists()) {
    throw new IllegalArgumentException("File does not exist: " + hprofFile);
  } else if (! hprofFile.canRead()) {
    throw new IllegalArgumentException("Can not read file: " + hprofFile);
  }
  if (VERBOSE) {  // use single thread in verbose so the out makes sense
    this.executor = SameThreadSubmitterExecutor.instance();
  } else {
    this.executor = executor;
  }
  this.hprofFile = hprofFile;
  parsingFutures = Collections.synchronizedList(new ArrayList<ListenableFuture<?>>());
  classMap = Collections.synchronizedMap(new HashMap<Long, ClassDefinition>());
  instances = Collections.synchronizedMap(new HashMap<Long, Instance>());
  stringMap = new HashMap<>();
  instanceSummary = new HashMap<>();
  arraySummary = new HashMap<>();
  leafInstances = new ArrayList<>();
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Override
public ListenableFuture<?> write(final MergedByteBuffers mbb) {
  if(isClosed()) {
    return FutureUtils.immediateFailureFuture(new IOException("Connection is Closed"));
  }
  synchronized(writerLock) {
    final SettableListenableFuture<Long> slf = makeClientSettableListenableFuture();
    lastWriteFuture = slf;
    final boolean needNotify = !canWrite();
    if(sslProcessor != null && sslProcessor.handShakeStarted()) {
      try {
        writeBuffers.add(sslProcessor.encrypt(mbb));
      } catch (EncryptionException e) {
        this.close(e);
        return lastWriteFuture;
      }
    } else {
      writeBuffers.add(mbb);
    }
    writeFutures.add(new Pair<>(writeBuffers.getTotalConsumedBytes()+writeBuffers.remaining(), slf));
    if(needNotify && se != null && channel.isConnected()) {
      se.setClientOperations(this);
    }
    return lastWriteFuture;
  }
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Test
public void tcpTimeout() throws Throwable {
  TCPClient client = SE.createTCPClient("2.0.0.2", port);
  client.setConnectionTimeout(1);
  assertTrue(!client.hasConnectionTimedOut());
  ListenableFuture<?> lf = client.connect();
  Thread.sleep(10);
  assertTrue(client.hasConnectionTimedOut());
  System.out.println(lf.isCancelled());
  System.out.println(lf.isDone());
  while(!lf.isCancelled() || !lf.isDone()) {
    Thread.sleep(10);
  }
  System.out.println(lf.isCancelled());
  System.out.println(lf.isDone());
  //assertTrue(lf.isCancelled());
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Test
public void submitScheduledCallableTest() throws InterruptedException, ExecutionException {
  TestCallable tc = new TestCallable();
  ListenableFuture<?> future = scheduler.submitScheduled(tc, DELAY_TIME);
  
  int runCount = 0;
  while (runCount == 0) {
    runCount = scheduler.tick(null);
  }
  
  assertEquals(1, runCount);
  
  assertTrue(tc.getDelayTillFirstRun() >= DELAY_TIME);
  assertTrue(future.isDone());
  assertTrue(future.get() == tc.getReturnedResult());
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Test
public void submitRunnableExceptionTest() throws InterruptedException {
  SubmitterExecutorFactory factory = getSubmitterExecutorFactory();
  try {
    SubmitterExecutor executor = factory.makeSubmitterExecutor(TEST_QTY, false);
    
    RuntimeException failure = new SuppressedStackRuntimeException();
    TestRuntimeFailureRunnable tr = new TestRuntimeFailureRunnable(failure);
    ListenableFuture<?> future = executor.submit(tr);
    // no exception should propagate
    
    try {
      future.get();
      fail("Exception should have thrown");
    } catch (ExecutionException e) {
      assertTrue(e.getCause() == failure);
    }
  } finally {
    factory.shutdown();
  }
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Test
public void watchAlreadyDoneFutureWithFailureTest() throws InterruptedException {
  final Throwable rootCause = new Exception();
  ListenableFuture<Object> lfResult = poller.watch(new AlreadyDoneFuture() {
    @Override
    public Object get() throws ExecutionException {
      throw new ExecutionException(rootCause);
    }

    @Override
    public Object get(long timeout, TimeUnit unit) throws ExecutionException {
      throw new ExecutionException(rootCause);
    }
  });
  
  assertTrue(lfResult.isDone());
  try {
    lfResult.get();
    fail("Exception should have thrown");
  } catch (ExecutionException e) {
    assertTrue(e.getCause() == rootCause);
  }
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Test
public void submitCallableExceptionTest() throws InterruptedException {
  SubmitterExecutorFactory factory = getSubmitterExecutorFactory();
  try {
    SubmitterExecutor executor = factory.makeSubmitterExecutor(TEST_QTY, false);
    
    final RuntimeException failure = new SuppressedStackRuntimeException();
    ListenableFuture<?> future = executor.submit(new Callable<Void>() {
      @Override
      public Void call() throws Exception {
        throw failure;
      }
    });
    // no exception should propagate
    
    try {
      future.get();
      fail("Exception should have thrown");
    } catch (ExecutionException e) {
      assertTrue(e.getCause() == failure);
    }
  } finally {
    factory.shutdown();
  }
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
@Test
public void shutdownNowIgnoreCanceledFuturesTest() {

  PrioritySchedulerServiceFactory factory = getPrioritySchedulerFactory();
  try {
    PriorityScheduler scheduler = factory.makePriorityScheduler(1);
    
    Runnable nonCanceledRunnable = new TestRunnable();
    scheduler.submitScheduled(nonCanceledRunnable, 1000 * 60 * 60);
    ListenableFuture<?> future = scheduler.submitScheduled(DoNothingRunnable.instance(), 
                                                           1000 * 60 * 60);
    
    future.cancel(false);
    
    List<Runnable> result = scheduler.shutdownNow();
    assertEquals(1, result.size()); // only canceled task removed
  } finally {
    factory.shutdown();
  }
}
 

import org.threadly.concurrent.future.ListenableFuture; //导入依赖的package包/类
/**
 * This is called once a task is ready to be executed (or if unable to execute immediately, 
 * queued).  In addition to the task itself, this function takes in any future which represents 
 * task execution (if available, otherwise {@code null}).  Passing in as a separate argument 
 * allows us to avoid a {@code instanceof} check, but does require it to be specified for 
 * pre-future listener completion support.
 * 
 * @param task Task to be executed
 * @param future Future to represent task completion or {@code null} if not available
 */
protected void executeOrQueue(Runnable task, ListenableFuture<?> future) {
  if (limitFutureListenersExecution || future == null) {
    executeOrQueueWrapper(new LimiterRunnableWrapper(task));
  } else {
    // we will release the limit restriction as soon as the future completes.
    // listeners should be invoked in order, so we just need to be the first listener here
    // We add a `SameThreadSubmitterExecutor` so that we get executed first as if it was async
    future.addListener(this::releaseExecutionLimit, SameThreadSubmitterExecutor.instance());

    if (canRunTask()) {
      executor.execute(task);
    } else {
      addToQueue(new TransparentRunnableContainer(task));
    }
  }
}