Java Code Examples for java.util.concurrent.Future

@SuppressWarnings("unchecked") public Future submit(Runnable task){
  Future result=null;
  if (!notificationService.getExecutorService().isTerminated() && !notificationService.getExecutorService().isShutdown() && task != null) {
    result=notificationService.getExecutorService().submit(task);
  }
  return result;
}
 

public DbFuture<Connection> connect() throws DbException {
  if (isClosed()) {
    throw new DbException("This connection manager is closed");
  }
  final DbFutureConcurrentProxy<Connection> future=new DbFutureConcurrentProxy<Connection>();
  Future<Connection> executorFuture=executorService.submit(new Callable<Connection>(){
    public Connection call() throws Exception {
      try {
        java.sql.Connection jdbcConnection=DriverManager.getConnection(jdbcUrl,properties);
        JdbcConnection connection=new JdbcConnection(JdbcConnectionManager.this,jdbcConnection);
synchronized (lock) {
          if (isClosed()) {
            connection.close(true);
            future.setException(new DbException("Connection manager closed"));
          }
 else {
            connections.add(connection);
            future.setValue(connection);
          }
        }
        return connection;
      }
 catch (      SQLException e) {
        future.setException(new DbException(e));
        e.printStackTrace();
        throw e;
      }
 finally {
        future.setDone();
      }
    }
  }
);
  future.setFuture(executorFuture);
  return future;
}
 

@PostConstruct public void start(){
  Future<?> future=executor.submit(new Runnable(){
    @Override public void run(){
      while (!Thread.interrupted()) {
        final List<T> entries=new ArrayList<T>(maxBatchSize);
        try {
          T first=queue.take();
          entries.add(first);
          queue.drainTo(entries,maxBatchSize - 1);
          handler.processBatch(entries);
        }
 catch (        InterruptedException e) {
          Thread.currentThread().interrupt();
        }
      }
    }
  }
);
  this.future.set(future);
}
 

/** 
 * Retrieve the contents for all regions, one region per task executor
 */
@Override public CommandResult run(ConfigurableApplicationContext mainContext,List<String> parameters){
  Map<String,GemfireTemplate> allRegionTemplates=CommandRegionUtils.getAllGemfireTemplates(mainContext);
  ExecutorService taskExecutor=Executors.newFixedThreadPool(allRegionTemplates.size());
  Collection tasks=new ArrayList<RegionFetcher>();
  CommandTimer timer=new CommandTimer();
  for (  String key : allRegionTemplates.keySet()) {
    GemfireTemplate oneTemplate=allRegionTemplates.get(key);
    if (parallelFetch) {
      tasks.add(new RegionFetcher(oneTemplate.getRegion().getName(),0,oneTemplate));
    }
 else {
      fetchOneRegion(oneTemplate.getRegion().getName(),5,oneTemplate);
    }
  }
  if (parallelFetch) {
    try {
      List<Future<?>> futures=taskExecutor.invokeAll(tasks);
      taskExecutor.shutdown();
      LOG.info("Fetched " + futures.size() + " regions in threads");
    }
 catch (    InterruptedException e) {
      e.printStackTrace();
    }
  }
  timer.stop();
  return new CommandResult(null,"Loading all regions took " + timer.getTimeDiffInSeconds() + " seconds");
}
 

/** 
 * Send the configuration requests we can only do after we have the features reply
 * @throws IOException
 */
void sendFeatureReplyConfiguration() throws IOException {
  OFSetConfig config=(OFSetConfig)factory.getMessage(OFType.SET_CONFIG);
  config.setMissSendLength((short)0xffff).setLengthU(OFSwitchConfig.MINIMUM_LENGTH);
  sw.write(config,null);
  sw.write(factory.getMessage(OFType.GET_CONFIG_REQUEST),null);
  OFStatisticsRequest req=new OFStatisticsRequest();
  req.setStatisticType(OFStatisticsType.DESC);
  req.setLengthU(req.getLengthU());
  Future<List<OFStatistics>> dfuture=sw.getStatistics(req);
  sw.setAttribute(IOFSwitch.SWITCH_DESCRIPTION_FUTURE,dfuture);
}
 

@Override public String getPermissionLimit(final Player p,final String limit){
  String result=null;
  if (mChat) {
    result=Reader.getInfo(p.getName(),InfoType.USER,p.getWorld().getName(),"admincmd." + limit);
  }
  if (result == null || (result != null && result.isEmpty())) {
    final Pattern regex=Pattern.compile("admincmd\\." + limit.toLowerCase() + "\\.[0-9]+");
    Set<PermissionAttachmentInfo> permissions=null;
    if (ACHelper.isMainThread()) {
      permissions=p.getEffectivePermissions();
    }
 else {
      final Callable<Set<PermissionAttachmentInfo>> perms=new Callable<Set<PermissionAttachmentInfo>>(){
        @Override public Set<PermissionAttachmentInfo> call() throws Exception {
          return p.getEffectivePermissions();
        }
      }
;
      final Future<Set<PermissionAttachmentInfo>> permTask=ACPluginManager.getScheduler().callSyncMethod(ACPluginManager.getCorePlugin(),perms);
      try {
        permissions=permTask.get();
        DebugLog.INSTANCE.info("Perms got for " + p.getName());
      }
 catch (      final InterruptedException e) {
        DebugLog.INSTANCE.info("Problem while gettings ASYNC perm of " + p.getName());
      }
catch (      final ExecutionException e) {
        DebugLog.INSTANCE.info("Problem while gettings ASYNC perm of " + p.getName());
      }
    }
    return permissionCheck(permissions,regex);
  }
 else {
    return result;
  }
}
 

@Override public <T>List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
  List<Future<T>> futures=new ArrayList<Future<T>>(tasks.size());
  for (  Callable<T> c : tasks) {
    futures.add(submit(c));
  }
  return futures;
}
 

private void assertSuccess(List<Future<Boolean>> errors,long timeout,TimeUnit unit) throws Exception {
  boolean fail=false;
  for (  Future<Boolean> f : errors) {
    Boolean e=f.get(timeout,unit);
    if (!e) {
      fail=true;
    }
  }
  if (fail) {
    throw new RuntimeException("See log for details.");
  }
}
 

public Future<?> scheduleDecayTask(final Npc npc){
  final World world=npc.getActiveRegion().getWorld();
  return ThreadPoolManager.getInstance().schedule(new Runnable(){
    @Override public void run(){
      PacketSendUtility.broadcastPacket(npc,new SM_DELETE(npc));
      world.despawn(npc);
    }
  }
,DECAY_DEFAULT_DELAY);
}
 

@Test public void loadTest() throws ExecutionException, InterruptedException, IOException {
  client=newEventClient(asList(baseUri));
  List<Future<Void>> futures=newArrayList();
  for (int i=0; i < 100; i++) {
    futures.add(client.post(TestingUtils.getEvents()));
  }
  for (  Future<Void> future : futures) {
    future.get();
    System.out.println("future " + future);
  }
  assertEquals(servlet.lastPath,"/v2/event");
  assertEquals(servlet.lastBody,getNormalizedJson("events.json"));
}
 

/** 
 * Tests that getStatus() is relayed correctly.
 */
@Test public void testGetStatus() throws Exception {
  Future<Integer> future=executor.submit(new Callable<Integer>(){
    public Integer call() throws Exception {
      return st.getStatus();
    }
  }
);
  Operation<?> op=st.getNextOperation();
  assertTrue(op instanceof GetStatus);
  st.postResult(new Result<Integer>(42));
  assertEquals(new Integer(42),future.get());
}
 

/** 
 * @see com.sworddance.taskcontrol.FutureListener#futureSet(java.util.concurrent.Future,Object)
 */
@SuppressWarnings({"hiding","unchecked"}) public <P extends Future<MV>>void futureSet(P future,MV returnedValue){
  checkDoneStateAndSaveStack();
  this.readListenersLock.lock();
  try {
    this.setMonitoredFuture(future);
    if (this.returnedFuture == null) {
      this.setReturnedFuture((Future<RV>)this.monitoredFuture.get());
    }
    if (this.returnedFuture == null || this.returnedFuture.get() == null) {
      this.setReturnedValue((RV)returnedValue);
    }
    this.done.countDown();
    for (    FutureListener<RV> futureListener : NotNullIterator.<FutureListener<RV>>newNotNullIterator(listeners)) {
      notifyListener(futureListener);
    }
  }
  finally {
    this.readListenersLock.unlock();
  }
  clear();
}
 

public MultithreadedAnalyzer(Analyzer analyzer,int threads){
  if (analyzer == null) {
    throw new NullPointerException("analyzer can't be null");
  }
  if (threads < 1) {
    throw new IllegalArgumentException("thread count must be greater than or equal to 1");
  }
  this.analyzer=analyzer;
  exec=Executors.newFixedThreadPool(threads);
  results=new ConcurrentHashMap<Integer,Future<Report>>();
}
 

/** 
 * Cancels any pending (or potentially active) requests associated with the passed Context. <p> <b>Note:</b> This will only affect requests which were created with a non-null android Context. This method is intended to be used in the onDestroy method of your android activities to destroy all requests which are no longer required.
 * @param context the android Context instance associated to the request.
 * @param mayInterruptIfRunning specifies if active requests should be cancelled along with pending requests.
 */
public void cancelRequests(Context context,boolean mayInterruptIfRunning){
  List<WeakReference<Future<?>>> requestList=requestMap.get(context);
  if (requestList != null) {
    for (    WeakReference<Future<?>> requestRef : requestList) {
      Future<?> request=requestRef.get();
      if (request != null) {
        request.cancel(mayInterruptIfRunning);
      }
    }
  }
  requestMap.remove(context);
}
 

/** 
 * Loads the specified RSS feed URI asynchronously. For the specified priority to determine the relative loading order of RSS feeds, this loader must have been constructed with  {@link #priority()} or {@link #priority(int)}. Otherwise, RSS feeds are loaded in FIFO order. <p> Returns  {@code null} if the RSS feed URI cannot be scheduled for loadingdue to resource constraints or if  {@link #stop()} has been previouslycalled. <p> Completed RSS feed loads can be retrieved by calling  {@link #take()}. Alternatively, non-blocking polling is possible with  {@link #poll()}.
 * @param uri RSS feed URI to be loaded
 * @param priority larger integer gives higher priority
 * @return Future representing the RSS feed scheduled for loading,{@code null} if scheduling failed
 */
public Future<RSSFeed> load(String uri,int priority){
  if (uri == null) {
    throw new IllegalArgumentException("RSS feed URI must not be null.");
  }
  if (stopped) {
    return null;
  }
  final RSSFuture future=new RSSFuture(uri,priority);
  final boolean ok=in.offer(future);
  if (!ok || stopped) {
    return null;
  }
  return future;
}
 

/** 
 * @since 3.0
 */
public Future asyncGetRemoteServiceReferences(final ID[] idFilter,final String clazz,final RegularExpression filter){
  ExecutorService executor=Executors.newCachedThreadPool();
  return executor.submit(new Runnable(){
    public void run(){
      try {
        getRemoteServiceReferences(idFilter,clazz,filter);
      }
 catch (      InvalidSyntaxException e) {
        e.printStackTrace();
      }
    }
  }
);
}
 

public final Future<State> start(){
  lock.lock();
  try {
    if (state == State.NEW) {
      state=State.STARTING;
      doStart();
    }
  }
 catch (  Throwable startupFailure) {
    notifyFailed(startupFailure);
  }
 finally {
    lock.unlock();
  }
  return startup;
}
 

@Test public void testCLI() throws IOException, RDFHandlerException, RDFParseException {
  assumeOnlineAllowed();
  final File outFile=File.createTempFile("crawler-test",".nq",tempDirectory);
  outFile.delete();
  logger.info("Outfile: " + outFile.getAbsolutePath());
  final Future<?> future=Executors.newSingleThreadExecutor().submit(new Runnable(){
    @Override public void run(){
      try {
        ToolRunner.main(String.format("crawler -f nquads --maxpages 50 --maxdepth 1 --politenessdelay 500 -o %s " + "http://eventiesagre.it/",outFile.getAbsolutePath()).split(" "));
      }
 catch (      Exception e) {
        e.printStackTrace();
      }
    }
  }
);
  try {
    future.get(10,TimeUnit.SECONDS);
  }
 catch (  Exception e) {
    if (!(e instanceof TimeoutException)) {
      e.printStackTrace();
    }
  }
  assertTrue("The output file has not been created.",outFile.exists());
  final String[] lines=FileUtils.readFileLines(outFile);
  final StringBuilder allLinesExceptLast=new StringBuilder();
  for (int i=0; i < lines.length - 1; i++) {
    allLinesExceptLast.append(lines[i]);
  }
  final Statement[] statements=RDFUtils.parseRDF(RDFFormat.NQUADS,allLinesExceptLast.toString());
  assertTrue(statements.length > 0);
}
 

/** 
 * Spawns multiple threads, one for each controller in the list of controllers, and within each thread, delegates to the controller's handleRequest() method. Once all the threads are complete, the ModelAndView objects returned from each of the handleRequest() methods are merged into a single view. The view name for the model is set to the specified view name. If an exception is thrown by any of the controllers in the chain, this exception is propagated up from the handleRequest() method of the ChainedController.
 * @param request  the HttpServletRequest object.
 * @param response the HttpServletResponse object.
 * @return a merged ModelAndView object.
 * @throws Exception if one is thrown from the controllers in the chain.
 */
@SuppressWarnings("unchecked") private ModelAndView handleRequestParallely(HttpServletRequest request,HttpServletResponse response) throws Exception {
  ExecutorService service=Executors.newCachedThreadPool();
  int numberOfControllers=controllers.size();
  CallableController[] callables=new CallableController[numberOfControllers];
  Future<ModelAndView>[] futures=new Future[numberOfControllers];
  for (int i=0; i < numberOfControllers; i++) {
    callables[i]=new CallableController(controllers.get(i),request,response);
    futures[i]=service.submit(callables[i]);
  }
  ModelAndView mergedModel=new ModelAndView();
  for (  Future<ModelAndView> future : futures) {
    ModelAndView model=future.get();
    if (model != null) {
      mergedModel.addAllObjects(model.getModel());
    }
  }
  if (StringUtils.isNotEmpty(this.viewName)) {
    mergedModel.setViewName(this.viewName);
  }
  return mergedModel;
}
 

/** 
 * Spawns a process defined by command. Process output is discarded
 * @param command the command to be executed
 * @return
 * @throws InterruptedException
 * @throws ExecutionException
 */
public Process spawn(String... command) throws InterruptedException, ExecutionException {
  Future<Process> processFuture=service.submit(new SpawnedProcess(command));
  Process process=processFuture.get();
  service.submit(new ProcessOutputConsumer(new ProcessWithId(process,command[0])));
  shutdownThreads.addHookFor(process);
  return process;
}
 

/** 
 * Wait for all tasks to finish.
 * @param futures
 * @return true if all tasks returned true or false otherwise.  
 */
private boolean waitForTasksToFinish(){
  for (  Future<Boolean> future : futures) {
    try {
      if (!future.get()) {
        cancel();
        return false;
      }
    }
 catch (    Throwable t) {
      cancel();
      return false;
    }
  }
  return true;
}
 

private void performInstalls(ListeningExecutorService es,List<Pair<Host,List<Uri<Installer>>>> floggles){
  InstallerCache installers=new InstallerCache(this);
  final List<Future<?>> futures=Lists.newArrayList();
  for (  Pair<Host,List<Uri<Installer>>> pair : floggles) {
    List<Pair<Uri<Installer>,Installer>> real=Lists.newArrayList();
    for (    Uri<Installer> uri : pair.getRight()) {
      real.addAll(installers.lookup(uri));
    }
    futures.add(installAllOnHost(es,pair.getLeft(),real));
  }
  for (  Future<?> future : futures) {
    try {
      future.get();
    }
 catch (    Exception e) {
      log.warn(e,"Exception trying to execute installation");
      throw new UnsupportedOperationException("Not Yet Implemented!",e);
    }
  }
  installers.finished();
}
 

/** 
 * Delegate method actually performing the indexing work for objects implementing ContentStream
 * @param cs ContentStream to ingest
 * @param fields content specific fields
 * @param size size of the content - used to determine the Solr timeout, notused to populate meta-data
 * @throws IngesterException if there was an error processing a specificcontent, but the Solr server is probably fine.
 */
private void ingest(ContentStream cs,Map<String,String> fields,final long size) throws IngesterException {
  final ContentStreamUpdateRequest up=new ContentStreamUpdateRequest("/update/extract");
  up.addContentStream(cs);
  setFields(up,fields);
  up.setAction(AbstractUpdateRequest.ACTION.COMMIT,true,true);
  final String contentType=cs.getContentType();
  if (contentType != null && !contentType.trim().equals("")) {
    up.setParam("stream.contentType",contentType);
  }
  up.setParam("commit","false");
  final Future<?> f=upRequestExecutor.submit(new UpRequestTask(up));
  try {
    f.get(getTimeout(size),TimeUnit.SECONDS);
  }
 catch (  TimeoutException te) {
    logger.log(Level.WARNING,"Solr timeout encountered, trying to restart Solr");
    hardSolrRestart();
    throw new IngesterException("Solr index request time out for id: " + fields.get("id") + ", name: "+ fields.get("file_name"));
  }
catch (  Exception e) {
    throw new IngesterException("Problem posting content to Solr, id: " + fields.get("id") + ", name: "+ fields.get("file_name"),e);
  }
  uncommitedIngests=true;
}
 

private void executeCallables(List<SshCallable> sshCallables) throws IOException {
  ExecutorService e=Executors.newCachedThreadPool();
  List<Future<SshCallable>> futureList=Lists.newArrayListWithCapacity(sshCallables.size());
  for (  SshCallable sshCallable : sshCallables) {
    futureList.add(e.submit(sshCallable));
  }
  waitForSshCommandCompletion(futureList);
}
 

private Future<Object> runInSeperateThread(final ExecutorService executor,final AzkabanProcess process) throws InterruptedException {
  Future<Object> result=executor.submit(new Callable<Object>(){
    public Object call() throws IOException {
      process.run();
      return null;
    }
  }
);
  process.awaitStartup();
  return result;
}
 

@Test public void test_read_write() throws Exception {
  ExecutorService service=Executors.newCachedThreadPool();
  Future<Boolean> readerFuture=service.submit(new Callable<Boolean>(){
    @Override public Boolean call() throws Exception {
      byte[] buffer=new byte[1000];
      readerStarted=true;
      assertEquals(0,pipe.inputStream.read(buffer,5,0));
      assertFalse(writerStarted);
      assertEquals(2,pipe.inputStream.read(buffer,10,2));
      assertEquals("01",new String(buffer,10,2,Charsets.UTF_8));
      assertTrue(writerStarted);
      assertEquals(8,pipe.inputStream.read(buffer,1,10));
      assertEquals("23456789",new String(buffer,1,8,Charsets.UTF_8));
      assertFalse(writerClosed);
      assertEquals(-1,pipe.inputStream.read(buffer,1,10));
      assertTrue(writerClosed);
      return true;
    }
  }
);
  Thread.sleep(500);
  assertTrue(readerStarted);
  Future<Boolean> writerFuture=service.submit(new Callable<Boolean>(){
    @Override public Boolean call() throws Exception {
      writerStarted=true;
      pipe.outputStream.write(TEST_BYTES);
      Thread.sleep(500);
      pipe.outputStream.close();
      writerClosed=true;
      return true;
    }
  }
);
  assertTrue(readerFuture.get());
  assertTrue(writerFuture.get());
}
 

/** 
 * Fire the specified event asynchronously.
 * @param < T > the result type
 * @param event the specified event
 * @return future result
 * @throws EventException event exception
 */
public <T>Future<T> fireEventAsynchronously(final Event<?> event) throws EventException {
  final ExecutorService executorService=Executors.newSingleThreadExecutor();
  final FutureTask<T> futureTask=new FutureTask<T>(new Callable<T>(){
    @Override public T call() throws Exception {
      synchronizedEventQueue.fireEvent(event);
      return null;
    }
  }
);
  executorService.execute(futureTask);
  return futureTask;
}
 

public KafkaConsumer(String topic,Properties props,int numThreads){
  LOG.info("# of threads: " + numThreads);
  executor=Executors.newFixedThreadPool(numThreads);
  workers=new ArrayList<Future<?>>(numThreads);
  ConsumerConfig consumerConfig=new ConsumerConfig(props);
  consumerConnector=kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig);
  streams=consumerConnector.createMessageStreamsByFilter(new Whitelist(topic),numThreads);
  consumed=Metrics.newMeter(new MetricName("bagheera","consumer",topic + ".consumed"),"messages",TimeUnit.SECONDS);
}
 

public Object getInstance(String name){
  Future<Object> future=instances.get(name);
  if (future != null && future.isDone()) {
    try {
      return future.get();
    }
 catch (    InterruptedException e) {
      Thread.currentThread().interrupt();
      return null;
    }
catch (    ExecutionException e) {
      return null;
    }
  }
 else {
    return null;
  }
}
 

@Test(timeout=1000) public void testRetriesOnFailure() throws InterruptedException, ExecutionException {
  new GlobalBackoffThrottle(new ExponentialBackoff(),handlers);
  manager.enqueue(ResponseType.QuotaExceeded);
  manager.enqueue(ResponseType.QuotaExceeded);
  manager.enqueue(ResponseType.QuotaExceeded);
  manager.enqueue(ResponseType.QuotaExceeded);
  manager.enqueue(ResponseType.Success);
  Future<Response> fut=cut.pushMessage(msg);
  assertThat(fut.get().getResponseType(),is(ResponseType.Success));
}
 

@Test public void testFutureDone() throws Exception {
  LOG.info("Submitting task to Camel");
  Future<String> future=template.asyncRequestBody("seda:quote","Hello Camel",String.class);
  LOG.info("Task submitted and we got a Future handle");
  boolean done=false;
  while (!done) {
    done=future.isDone();
    LOG.info("Is the task done? " + done);
    if (!done) {
      Thread.sleep(2000);
    }
  }
  String answer=future.get();
  LOG.info("The answer is: " + answer);
}
 

public Future<byte[]> makeAsyncCall(ApiProxy.Environment environment,String packageName,String methodName,final byte[] bytes,ApiProxy.ApiConfig apiConfig){
  return ExecutorFactory.wrap(new Callable<byte[]>(){
    public byte[] call() throws Exception {
      return bytes;
    }
  }
);
}
 

private void retrieveAsynchronousResults(Element inputs,List<Callable<Content>> callables,PID pid,boolean failOnException) throws FedoraException {
  Collection<Future<Content>> futures=new ArrayList<Future<Content>>(callables.size());
  if (GroupsThreadStore.getGroups() != null) {
    String groups=GroupsThreadStore.getGroups();
    for (    Callable<Content> c : callables) {
      if (GroupForwardingCallable.class.isInstance(c)) {
        GroupForwardingCallable rfc=(GroupForwardingCallable)c;
        rfc.setGroups(groups);
      }
    }
  }
  for (  Callable<Content> callable : callables) {
    futures.add(executor.submit(callable));
  }
  for (  Future<Content> future : futures) {
    try {
      Content results=future.get(serviceTimeout,TimeUnit.MILLISECONDS);
      if (results != null) {
        inputs.addContent(results);
      }
    }
 catch (    InterruptedException e) {
      LOG.warn("Attempt to get asynchronous results was interrupted for " + pid.getPid(),e);
      return;
    }
catch (    Exception e) {
      if (failOnException) {
        throw new ServiceException("Failed to get asynchronous results for " + pid.getPid(),e);
      }
      LOG.error("Failed to get asynchronous results for " + pid.getPid() + ", continuing.",e);
    }
  }
}
 

/** 
 * {@inheritDoc} 
 */
public PoolStatus observe(){
  final Future<StatusCode> result=this.executor.submit(new Validator());
  StatusCode code;
  String description=null;
  try {
    code=result.get(this.maxWait,TimeUnit.MILLISECONDS);
  }
 catch (  final InterruptedException e) {
    code=StatusCode.UNKNOWN;
    description="Validator thread interrupted during pool validation.";
  }
catch (  final TimeoutException e) {
    code=StatusCode.WARN;
    description=String.format("Pool validation timed out.  Max wait is %s ms.",this.maxWait);
  }
catch (  final Exception e) {
    code=StatusCode.ERROR;
    description=e.getMessage();
  }
  return new PoolStatus(code,description,getActiveCount(),getIdleCount());
}
 

/** 
 * Method run implements the Runnable run method. 
 */
private void run(){
  Version.printBanner();
  if (LOG.isInfoEnabled())   logInfo("starting");
  registerShutdownHook();
  try {
    cascadeStats.markStartedThenRunning();
    initializeNewJobsMap();
    int numThreads=getMaxConcurrentFlows(properties,maxConcurrentFlows);
    if (numThreads == 0)     numThreads=jobsMap.size();
    int numLocalFlows=numLocalFlows();
    boolean runFlowsLocal=numLocalFlows > 1;
    if (runFlowsLocal)     numThreads=1;
    if (LOG.isInfoEnabled()) {
      logInfo(" parallel execution is enabled: " + !runFlowsLocal);
      logInfo(" starting flows: " + jobsMap.size());
      logInfo(" allocating threads: " + numThreads);
    }
    List<Future<Throwable>> futures=spawnStrategy.start(this,numThreads,jobsMap.values());
    for (    Future<Throwable> future : futures) {
      throwable=future.get();
      if (throwable != null) {
        cascadeStats.markFailed(throwable);
        if (!stop)         internalStopAllFlows();
        handleExecutorShutdown();
        break;
      }
    }
  }
 catch (  Throwable throwable) {
    this.throwable=throwable;
  }
 finally {
    if (!cascadeStats.isFinished())     cascadeStats.markSuccessful();
    deregisterShutdownHook();
  }
}
 

@Override public void addReadings(List<Reading> readings){
  MutationBatch m=keyspace.prepareMutationBatch();
  for (  Reading reading : readings) {
    m.withRow(columnFamilyInfo,reading.getSensorId()).putColumn(reading.getTimestamp(),reading,ReadingSerializer.get(),ttl);
  }
  try {
    Future<OperationResult<Void>> future=m.executeAsync();
    OperationResult<Void> result=future.get();
  }
 catch (  ConnectionException e) {
    throw new RuntimeException("Storage of readings failed",e);
  }
catch (  InterruptedException e) {
    throw new RuntimeException("Storage of readings failed",e);
  }
catch (  ExecutionException e) {
    throw new RuntimeException("Storage of readings failed",e);
  }
}
 

@Test public void testUniquenessInSeveralThreads() throws Exception {
  final Set<Scope> uuids=Collections.synchronizedSet(new HashSet<Scope>());
  final Set<String> uuidStrings=Collections.synchronizedSet(new HashSet<String>());
  List<Future<Boolean>> futures=new ArrayList<Future<Boolean>>();
  int numThreads=100;
  ExecutorService executor=Executors.newCachedThreadPool();
  for (int i=0; i < numThreads; i++) {
    futures.add(executor.submit(new UUIDFactoryTestRunnable(factory,uuids,uuidStrings)));
  }
  for (  Future<Boolean> future : futures) {
    assertTrue(future.get());
  }
  assertEquals("Generated non-unique hashing UUIDs.",NUM_TESTS * numThreads,uuids.size());
  assertEquals("Generated UUIDs with duplicate String values.",NUM_TESTS * numThreads,uuidStrings.size());
}
 

@Override public Future<T> take() throws InterruptedException {
  if (tasksCounter.get() == 0) {
    return null;
  }
  try {
    return super.take();
  }
 catch (  Exception e) {
    throw new IllegalStateException("CompletionService failed",e);
  }
 finally {
    tasksCounter.getAndDecrement();
  }
}
 

@Override public Object[] invoke(final String commandName,final Object[] params) throws Exception {
  final ServiceInstanceImpl[] instances=this.getInstances();
  final List<Future<Object>> futures=new ArrayList<Future<Object>>();
  for (  final ServiceInstanceImpl instance : instances) {
    final Future<Object> future=instance.invokeAsync(commandName,params);
    futures.add(future);
  }
  final long start=System.currentTimeMillis();
  final long end=start + DEFAULT_INVOKE_TIMEOUT;
  Exception firstException=null;
  final Object[] results=new Object[instances.length];
  for (int i=0; i < results.length; i++) {
    final Future<Object> future=futures.get(i);
    try {
      results[i]=future.get(end - System.currentTimeMillis(),TimeUnit.MILLISECONDS);
    }
 catch (    final Exception e) {
      results[i]=e;
      if (firstException == null) {
        firstException=e;
      }
    }
  }
  if (firstException != null) {
    throw firstException;
  }
  return results;
}
 

/** 
 * Retrieves future value or throws IllegalStateException if the future value can not be retrieved anymore.
 * @return
 */
private <T>T valueOrNotReady(Future<T> future){
  try {
    return future.get();
  }
 catch (  Exception ex) {
    throw new IllegalStateException("ComputeContext is not ready.",ex);
  }
}
 

@Test public void testConnectListenerThenDisconnectThenConnectHandler() throws Exception {
  final CountDownLatch connectLatch=new CountDownLatch(2);
  final CountDownLatch disconnectLatch=new CountDownLatch(1);
  bayeux.getChannel("/meta/connect").addListener(new ServerChannel.MessageListener(){
    public boolean onMessage(    ServerSession from,    ServerChannel channel,    ServerMessage.Mutable message){
      connectLatch.countDown();
      if (connectLatch.getCount() == 0)       await(disconnectLatch);
      return true;
    }
  }
);
  Request handshake=newBayeuxRequest("[{" + "\"channel\": \"/meta/handshake\"," + "\"version\": \"1.0\","+ "\"minimumVersion\": \"1.0\","+ "\"supportedConnectionTypes\": [\"long-polling\"]"+ "}]");
  ContentResponse response=handshake.send().get(5,TimeUnit.SECONDS);
  Assert.assertEquals(200,response.status());
  String clientId=extractClientId(response);
  String channelName="/foo";
  Request subscribe=newBayeuxRequest("[{" + "\"clientId\": \"" + clientId + "\","+ "\"channel\": \"/meta/subscribe\","+ "\"subscription\": \""+ channelName+ "\""+ "}]");
  response=subscribe.send().get(5,TimeUnit.SECONDS);
  Assert.assertEquals(200,response.status());
  Request connect1=newBayeuxRequest("[{" + "\"channel\": \"/meta/connect\"," + "\"clientId\": \"" + clientId + "\","+ "\"connectionType\": \"long-polling\""+ "}]");
  response=connect1.send().get(5,TimeUnit.SECONDS);
  Assert.assertEquals(200,response.status());
  Request connect2=newBayeuxRequest("[{" + "\"channel\": \"/meta/connect\"," + "\"clientId\": \"" + clientId + "\","+ "\"connectionType\": \"long-polling\""+ "}]");
  Future<ContentResponse> futureResponse=connect2.send();
  Assert.assertTrue(connectLatch.await(5,TimeUnit.SECONDS));
  Request disconnect=newBayeuxRequest("[{" + "\"channel\": \"/meta/disconnect\"," + "\"clientId\": \"" + clientId + "\""+ "}]");
  response=disconnect.send().get(5,TimeUnit.SECONDS);
  Assert.assertEquals(200,response.status());
  disconnectLatch.countDown();
  response=futureResponse.get(timeout * 2,TimeUnit.SECONDS);
  Assert.assertEquals(200,response.status());
  Assert.assertTrue(response.contentAsString().toLowerCase().contains("unknown"));
  Assert.assertNull(bayeux.getSession(clientId));
}
 

/** 
 * Uses http://blockchain.info/ to return transaction from given transaction hash.
 * @param txHash Hash of transaction to find
 * @param connect Connect transaction to it's inputs
 * @return
 */
protected Transaction getTransaction(Sha256Hash txHash,boolean connect){
  getBlockTime(txHash);
  try {
    String expr="/block-index/[0-9]*\">[0-9]*";
    Pattern patt=Pattern.compile(expr,Pattern.DOTALL | Pattern.UNIX_LINES);
    URL url=new URL("http://blockchain.info/tx-index/" + txHash);
    Matcher m=patt.matcher(getURLContent(url));
    URL url2;
    if (m.find()) {
      String parsed=m.group();
      String blockNumber=parsed.substring(parsed.indexOf(">") + 1);
      url2=new URL("http://blockchain.info/block-height/" + blockNumber);
      String expr2="<td>Hash</td>.*<td><a href=\"/block-index/[0-9]*/[a-z 0-9]*";
      Pattern patt2=Pattern.compile(expr2,Pattern.DOTALL | Pattern.UNIX_LINES);
      Matcher m2=patt2.matcher(getURLContent(url2));
      if (m2.find()) {
        String blockHash=m2.group().substring(m2.group().lastIndexOf("/") + 1);
        Future<Block> fb=peerGroup.getConnectedPeers().get(0).getBlock(new Sha256Hash(blockHash));
        Block b=fb.get();
        Transaction t=null;
        for (        Transaction t1 : b.getTransactions()) {
          if (t1.getHash().equals(txHash)) {
            t=t1;
          }
        }
        if (connect) {
          for (          TransactionInput input : t.getInputs()) {
            input.getOutpoint().setFromTx(getTransaction(input.getOutpoint().getHash(),false));
          }
        }
        return t;
      }
    }
  }
 catch (  Exception e) {
    e.printStackTrace();
  }
  return null;
}
 

@SuppressWarnings("unchecked") public Future<Boolean> publish(final ByteBuffer message,final String topic){
  Preconditions.checkNotNull(message,"Message can be empty but not null");
  Preconditions.checkNotNull(topic,"Topic can be empty but not null");
  final Channel channel=reconnectHandler.channel();
  logger.trace(channel != null ? "Publishing full message of length {} to topic[{}]" : "Not currently connected to a server, dropping message of length {} to topic[{}]",asArray(message.remaining(),topic));
  return channel != null ? new NettyToJDKFuture(channel.write(new ApplicationMessage(message,topic))) : NettyToJDKFuture.WRITE_FAILED;
}
 

@Test public void invokeMultiThreaded() throws Exception {
  String output=null;
  String response=null;
  Collection<Callable<String>> callables=new ArrayList<Callable<String>>();
  for (int i=0; i < _noOfThreads; i++) {
    callables.add(new WebServiceInvoker(i));
  }
  ExecutorService executorService=Executors.newFixedThreadPool(DEFAULT_THREAD_COUNT);
  Collection<Future<String>> futures=executorService.invokeAll(callables);
  Assert.assertEquals(futures.size(),_noOfThreads);
  int i=0;
  for (  Future<String> future : futures) {
    response=future.get();
    output="<SOAP-ENV:Envelope xmlns:SOAP-ENV=\"http://schemas.xmlsoap.org/soap/envelope/\">" + "<SOAP-ENV:Body>" + "   <test:sayHelloResponse xmlns:test=\"urn:switchyard-component-soap:test-ws:1.0\">"+ "      <return>Hello Thread " + i + "! The soapAction received is </return>"+ "   </test:sayHelloResponse>"+ "</SOAP-ENV:Body></SOAP-ENV:Envelope>";
    XMLAssert.assertXMLEqual(output,response);
    i++;
  }
}
 

public synchronized Future<Runnable> executeAsync(){
  if (queue.size() == 0) {
    throw new AssertionFailedError();
  }
  Runnable runnable=queue.removeFirst();
  FutureTask<Runnable> future=new FutureTask<Runnable>(runnable,runnable);
  new Thread(future).start();
  return future;
}
 

/** 
 * Execute the # {@link InvocationContext}, unwrap the results from their # {@link AsyncResult}if necessary and fire a post-execution event.
 * @return The result of the method invocation, unwrapped if #{@literal popResultsFromFuture} is true (ie: the return type of the method is a # {@link Future}).
 * @throws Exception Includes any exception thrown by the invoked method.
 */
public Object executeInvocationContext() throws Exception {
  Object result;
  if (ic == null || ic.getMethod() == null) {
    throw new InternalException("Failed to provide an InvocationContext/method to this " + this.getClass().getName());
  }
  final Method method=ic.getMethod();
  if (log.isTraceEnabled()) {
    log.trace("Running Invocation Context for " + method.getName());
  }
  final ArrayList<Annotation> qualifiers=new ArrayList<Annotation>();
  for (  Annotation ant : method.getAnnotations()) {
    if (beanMan.isQualifier(ant.annotationType())) {
      qualifiers.add(ant);
    }
  }
  ic.getContextData().put(INVOKED_IN_THREAD,Boolean.TRUE);
  result=ic.proceed();
  if (popResultsFromFuture) {
    result=((Future)result).get();
  }
  if (result != null) {
    if (log.isTraceEnabled()) {
      log.trace("Firing post execution event result: " + result);
    }
    beanMan.fireEvent(result,qualifiers.toArray(new Annotation[qualifiers.size()]));
  }
 else {
    if (log.isTraceEnabled()) {
      if (method.getReturnType().equals(Void.TYPE)) {
        log.trace("Method invocation on " + method.getName() + ":"+ method.getClass().getName()+ " returns void, so not firing a post-execution event");
      }
 else {
        log.trace("Method invocation on " + method.getName() + ":"+ method.getClass().getName()+ " returned null, so not firing an event");
      }
    }
  }
  return result;
}
 

@Test public void testSimulation() throws Exception {
  final int threadQty=20;
  final int executionQty=50;
  final AtomicInteger optimisticTries=new AtomicInteger();
  final AtomicInteger promotedLockTries=new AtomicInteger();
  final AtomicInteger failures=new AtomicInteger();
  final AtomicInteger errors=new AtomicInteger();
  final SummaryStatistics timingStats=new SynchronizedSummaryStatistics();
  List<Future<Void>> procs=Lists.newArrayList();
  ExecutorService executorService=Executors.newFixedThreadPool(threadQty);
  for (int i=0; i < threadQty; ++i) {
    Callable<Void> proc=new Callable<Void>(){
      @Override public Void call() throws Exception {
        doSimulation(executionQty,timingStats,optimisticTries,promotedLockTries,failures,errors);
        return null;
      }
    }
;
    procs.add(executorService.submit(proc));
  }
  for (  Future<Void> f : procs) {
    f.get();
  }
  System.out.println("OptimisticTries: " + optimisticTries.get());
  System.out.println("PromotedLockTries: " + promotedLockTries.get());
  System.out.println("Failures: " + failures.get());
  System.out.println("Errors: " + errors.get());
  System.out.println();
  System.out.println("Avg time: " + timingStats.getMean());
  System.out.println("Max time: " + timingStats.getMax());
  System.out.println("Min time: " + timingStats.getMin());
  System.out.println("Qty: " + timingStats.getN());
  Assert.assertEquals(errors.get(),0);
  Assert.assertTrue(optimisticTries.get() > 0);
  Assert.assertTrue(promotedLockTries.get() > 0);
}
 

private List<KamNode> searchKAMNodes(){
  ExecutorService e=Executors.newSingleThreadExecutor();
  Future<List<KamNode>> future=e.submit(buildCallable());
  while (!(future.isDone() || future.isCancelled()) && !e.isShutdown()) {
    try {
      if (halt) {
        e.shutdownNow();
        future.cancel(true);
      }
      Thread.sleep(100);
    }
 catch (    InterruptedException ex) {
      halt=true;
    }
  }
  if (future.isCancelled()) {
    return null;
  }
  try {
    return future.get();
  }
 catch (  InterruptedException ex) {
    log.warn("Error searching kam nodes",ex);
    return null;
  }
catch (  ExecutionException ex) {
    log.warn("Error searching kam nodes",ex);
    return null;
  }
}
 

@Override public <V>Future<V> submitLimited(final Rejectable<V> r){
  Callable<V> task=new Callable<V>(){
    Rejectable<V> o=r;
    @Override public V call() throws Exception {
      long num=numLimited.decrementAndGet();
      if (num <= maxNumWaitingLimitedTasks)       return o.call();
      o.rejected();
      return null;
    }
  }
;
  numLimited.incrementAndGet();
  Future<V> ret=executor.submit(task);
  return ret;
}
 

@Override public Future<DirectMemoryResponse> asyncPut(final DirectMemoryRequest request) throws DirectMemoryException {
  return Executors.newSingleThreadExecutor().submit(new Callable<DirectMemoryResponse>(){
    @Override public DirectMemoryResponse call() throws Exception {
      return put(request);
    }
  }
);
}
 

@Override public int get(String key,Object value){
  Future<Object> f=client.asyncGet(key);
  try {
    if (f.get() == null) {
      System.out.println("GET: error getting data");
      return -1;
    }
  }
 catch (  InterruptedException e) {
    System.out.println("GET Interrupted");
  }
catch (  ExecutionException e) {
    System.out.println("GET Execution");
    e.printStackTrace();
    return -2;
  }
catch (  RuntimeException e) {
    System.out.println("GET Runtime");
    return -3;
  }
  return 0;
}
 

@Test public void multiThreadTest() throws InterruptedException {
  final SyncDialogueHelper helper=new SyncDialogueHelper(endpoint);
  final int EMERGENCY_COUNT=45;
  final int THREAD_COUNT=10;
  Collection<Callable<Void>> tasks=new ArrayList<Callable<Void>>();
  for (int i=0; i < EMERGENCY_COUNT; i++) {
    final Emergency emergency=new Emergency("Emergency" + i,new Date(),"Fire" + i,10);
    tasks.add(new Callable<Void>(){
      public Void call() throws Exception {
        helper.invokeInform("me","you",emergency);
        helper.invokeRequest("coordinateEmergency",new LinkedHashMap<String,Object>());
        helper.getReturn(false);
        return null;
      }
    }
);
  }
  ExecutorService executorService=Executors.newFixedThreadPool(THREAD_COUNT);
  List<Future<Void>> futures=executorService.invokeAll(tasks);
  assertEquals(futures.size(),EMERGENCY_COUNT);
}
 

/** 
 * Loads the specified RSS feed URI asynchronously. For the specified priority to determine the relative loading order of RSS feeds, this loader must have been constructed with  {@link #priority()} or {@link #priority(int)}. Otherwise, RSS feeds are loaded in FIFO order. <p> Returns  {@code null} if the RSS feed URI cannot be scheduled for loadingdue to resource constraints or if  {@link #stop()} has been previouslycalled. <p> Completed RSS feed loads can be retrieved by calling  {@link #take()}. Alternatively, non-blocking polling is possible with  {@link #poll()}.
 * @param uri RSS feed URI to be loaded
 * @param priority larger integer gives higher priority
 * @return Future representing the RSS feed scheduled for loading,{@code null} if scheduling failed
 */
public Future<RSSFeed> load(String uri,int priority){
  if (uri == null) {
    throw new IllegalArgumentException("RSS feed URI must not be null.");
  }
  if (stopped) {
    return null;
  }
  final RSSFuture future=new RSSFuture(uri,priority);
  final boolean ok=in.offer(future);
  if (!ok || stopped) {
    return null;
  }
  return future;
}
 

@Test public void testSimpleOperations() throws Exception {
  Future<Boolean> setResult=memcachedClient.set("/test/person/1",0,jsonBuilder().startObject().field("test","value").endObject().bytes().copyBytesArray().array());
  assertThat(setResult.get(10,TimeUnit.SECONDS),equalTo(true));
  ClusterHealthResponse health=node.client().admin().cluster().prepareHealth().setWaitForYellowStatus().execute().actionGet();
  assertThat(health.timedOut(),equalTo(false));
  String getResult=(String)memcachedClient.get("/_refresh");
  System.out.println("REFRESH " + getResult);
  assertThat(getResult,Matchers.containsString("\"total\":10"));
  assertThat(getResult,Matchers.containsString("\"successful\":5"));
  assertThat(getResult,Matchers.containsString("\"failed\":0"));
  getResult=(String)memcachedClient.get("/test/person/1");
  System.out.println("GET " + getResult);
  assertThat(getResult,Matchers.containsString("\"_index\":\"test\""));
  assertThat(getResult,Matchers.containsString("\"_type\":\"person\""));
  assertThat(getResult,Matchers.containsString("\"_id\":\"1\""));
  Future<Boolean> deleteResult=memcachedClient.delete("/test/person/1");
  assertThat(deleteResult.get(10,TimeUnit.SECONDS),equalTo(true));
  getResult=(String)memcachedClient.get("/_refresh");
  System.out.println("REFRESH " + getResult);
  assertThat(getResult,Matchers.containsString("\"total\":10"));
  assertThat(getResult,Matchers.containsString("\"successful\":5"));
  assertThat(getResult,Matchers.containsString("\"failed\":0"));
  getResult=(String)memcachedClient.get("/test/person/1");
  System.out.println("GET " + getResult);
}
 

public void createAndSubmitAddResourceTasks(final SyncChunk syncChunk){
  if (syncChunk.isSetResources()) {
    for (    Resource res : syncChunk.getResources()) {
      ElemInfo resInfo=new ElemInfo();
      resInfo.guid=res.getGuid();
      resInfo.usn=res.getUpdateSequenceNum();
      final RetrieveAndAddResourceTask task=new RetrieveAndAddResourceTask(resInfo,util);
      final Future<Result> future=RP.submit(task);
      final int usn=(res.getUpdateSequenceNum() > 0) ? res.getUpdateSequenceNum() : fromUSN - 1;
      tasks.put(future,usn);
    }
  }
}
 

@Override public void committed(XaDataSource dataSource,int identifier,long txId,Integer externalAuthorServerId){
  int replicationFactor=desiredReplicationFactor;
  if (externalAuthorServerId != null)   replicationFactor--;
  if (replicationFactor == 0)   return;
  Collection<Future<Void>> committers=new HashSet<Future<Void>>();
  try {
    int successfulReplications=0;
    Iterator<Slave> slaves=filter(replicationStrategy.prioritize(broker.getSlaves()),externalAuthorServerId);
    CompletionNotifier notifier=new CompletionNotifier();
    for (int i=0; i < replicationFactor && slaves.hasNext(); i++)     committers.add(slaveCommitters.submit(slaveCommitter(dataSource,identifier,slaves.next(),txId,notifier)));
    Collection<Future<Void>> toAdd=new ArrayList<Future<Void>>();
    Collection<Future<Void>> toRemove=new ArrayList<Future<Void>>();
    while (!committers.isEmpty() && successfulReplications < replicationFactor) {
      toAdd.clear();
      toRemove.clear();
      for (      Future<Void> committer : committers) {
        if (!committer.isDone())         continue;
        if (isSuccessfull(committer))         successfulReplications++;
 else         if (slaves.hasNext())         toAdd.add(slaveCommitters.submit(slaveCommitter(dataSource,identifier,slaves.next(),txId,notifier)));
        toRemove.add(committer);
      }
      if (!toAdd.isEmpty())       committers.addAll(toAdd);
      if (!toRemove.isEmpty())       committers.removeAll(toRemove);
      if (!committers.isEmpty())       notifier.waitForAnyCompletion();
    }
    if (!(successfulReplications >= replicationFactor))     log.logMessage("Transaction " + txId + " for "+ dataSource.getName()+ " couldn't commit on enough slaves, desired "+ replicationFactor+ ", but could only commit at "+ successfulReplications);
  }
 catch (  Throwable t) {
    t.printStackTrace();
    log.logMessage("Unknown error commit master transaction at slave",t);
  }
 finally {
    for (    Future<Void> committer : committers)     committer.cancel(false);
  }
}
 

/** 
 * Send a single event to the collector
 * @param eventPayload Event to sent, created by the EventBuilder
 * @return true on success (collector got the event), false otherwise (event was lost)
 */
public Future<Boolean> send(final String eventPayload){
  if (client == null || client.isClosed()) {
    client=new AsyncHttpClient(clientConfig);
  }
  try {
    final AsyncHttpClient.BoundRequestBuilder requestBuilder=client.prepareGet(collectorURI + eventPayload);
    log.debug("Sending event to collector: {}",eventPayload);
    activeRequests.incrementAndGet();
    return client.executeRequest(requestBuilder.build(),new AsyncCompletionHandler<Boolean>(){
      @Override public Boolean onCompleted(      final Response response){
        activeRequests.decrementAndGet();
        if (response.getStatusCode() == 202) {
          return true;
        }
 else {
          log.warn("Received response from collector {}: {}",response.getStatusCode(),response.getStatusText());
          return false;
        }
      }
      @Override public void onThrowable(      final Throwable t){
        activeRequests.decrementAndGet();
      }
    }
);
  }
 catch (  IOException e) {
    client.close();
    return null;
  }
}