Java Code Examples for java.util.concurrent.BlockingQueue

@Test public void testRetry() throws InterruptedException {
  AsyncTaskExecutor<Void> executor=new AsyncTaskExecutor<Void>();
  AsyncFutureImpl task1=new AsyncFutureImpl();
  executor.schedule(task1);
  @SuppressWarnings("rawtypes") BlockingQueue queue=field("queue").ofType(BlockingQueue.class).in(executor).get();
  Thread.sleep(2000L);
  Assert.assertEquals(2,task1.calls.get());
  Assert.assertTrue(queue.isEmpty());
}
 

/** 
 * @param queueName the identifier of the input queue
 * @param capacity the number of elements which should fit into the inputqueue. This should be at least twice times bigger than batchSize.
 * @param batchSize the number of elements to feed at once into main outputqueue.
 * @return the newly registered queue
 */
public BlockingQueue<JTweet> register(String queueName,int capacity,int batchSize){
  BlockingQueue q=new LinkedBlockingQueue<JTweet>(capacity);
  QueueInfo qInfo=new QueueInfo(queueName,q);
  for (  QueueInfo<JTweet> qi : inputQueues) {
    if (qi.getName().equals(queueName))     throw new IllegalStateException("cannot register queue. Queue " + queueName + " already exists");
  }
  qInfo.setBatchSize(batchSize);
  inputQueues.add(qInfo);
  int sum=0;
  for (  QueueInfo<JTweet> qi : inputQueues) {
    sum+=qi.getBatchSize();
  }
  int mainCapacity=resolver.getInputQueue().remainingCapacity() + resolver.getInputQueue().size();
  if (sum * 2 > mainCapacity)   throw new IllegalStateException("cannot register queue " + queueName + " because it"+ " would increas capacity of all input queues too much ("+ sum+ ") and "+ " can block main queue too often, where the capacity is only:"+ mainCapacity);
  return qInfo.getQueue();
}
 

public void unregister(Receiver receiver,BlockingQueue<SoftReference<M>> q){
  for (  Registration r : regs) {
    if (r.receiver == receiver) {
synchronized (r) {
        if (q == null) {
          regs.remove(r);
        }
 else {
          r.setQueue(q);
        }
      }
    }
  }
}
 

@SuppressWarnings("unchecked") public TaskControl(Comparator<PrioritizedTask> activeComparator,int maxThreads,ThreadFactory threadFactory,Log log){
  this.log=log;
  ApplicationIllegalArgumentException.notNull(activeComparator,"activeComparator");
  this.eligibleTasks=new PriorityBlockingQueue<PrioritizedTask>(20,activeComparator);
  this.stateChangeNotificator=new ReentrantLock();
  this.newTasks=this.stateChangeNotificator.newCondition();
  this.runningTasks=new AtomicInteger(0);
  this.threadFactory=threadFactory;
  int keepAliveTime=10;
  int corePoolSize=1;
  this.executor=new ThreadPoolExecutor(corePoolSize,Math.max(corePoolSize,maxThreads),keepAliveTime,MICROSECONDS,(BlockingQueue)this.eligibleTasks,threadFactory);
  this.stayActive=true;
}
 

protected void configure(){
  super.configure();
  bind(TransformingHttpCommandExecutorService.class).to(NioTransformingHttpCommandExecutorService.class);
  bind(new TypeLiteral<BlockingQueue<HttpCommandRendezvous<?>>>(){
  }
).to(new TypeLiteral<LinkedBlockingQueue<HttpCommandRendezvous<?>>>(){
  }
).in(Scopes.SINGLETON);
  bind(NioHttpCommandExecutionHandler.ConsumingNHttpEntityFactory.class).to(ConsumingNHttpEntityFactoryImpl.class).in(Scopes.SINGLETON);
  bind(NHttpRequestExecutionHandler.class).to(NioHttpCommandExecutionHandler.class).in(Scopes.SINGLETON);
  bind(ConnectionReuseStrategy.class).to(DefaultConnectionReuseStrategy.class).in(Scopes.SINGLETON);
  bind(ByteBufferAllocator.class).to(HeapByteBufferAllocator.class);
  bind(NioHttpCommandConnectionPool.Factory.class).to(Factory.class).in(Scopes.SINGLETON);
}
 

private int saveQueue(BlockingQueue<CommandData> q,String prefsFileName){
  Context context=MyPreferences.getContext();
  int count=0;
  SharedPreferencesUtil.delete(context,prefsFileName);
  if (q.size() > 0) {
    SharedPreferences sp=MyPreferences.getSharedPreferences(prefsFileName,MODE_PRIVATE);
    while (q.size() > 0) {
      CommandData cd=q.poll();
      cd.save(sp,count);
      MyLog.v(TAG,"Command saved: " + cd.toString());
      count+=1;
    }
    MyLog.d(TAG,"Queue saved to " + prefsFileName + ", "+ count+ " msgs");
  }
  return count;
}
 

/** 
 * Removes all available elements from this  {@link Iterable} and adds them to the given  {@link Collection}.
 */
public static <T>int drainTo(Iterable<? extends T> src,Collection<? super T> dst,int max){
  if (src instanceof BlockingQueue<?>) {
    return ((BlockingQueue<? extends T>)src).drainTo(dst,max);
  }
  int count=0;
  for (Iterator<? extends T> it=src.iterator(); it.hasNext(); ) {
    if (count >= max) {
      break;
    }
    dst.add(it.next());
    it.remove();
    ++count;
  }
  return count;
}
 

public AsyncAgiReader(ManagerConnection connection,List<String> environment,BlockingQueue<AsyncAgiEvent> asyncAgiEvents){
  this.request=new AgiRequestImpl(environment);
  this.asyncAgiEvents=asyncAgiEvents;
  request.setLocalAddress(connection.getLocalAddress());
  request.setLocalPort(connection.getLocalPort());
  request.setRemoteAddress(connection.getRemoteAddress());
  request.setRemotePort(connection.getRemotePort());
}
 

/** 
 * Thread that runs continuously and writes outstanding requests to Avro files. This thread also deals with rolling files over and dropping old files when the span limit is reached.
 * @param compressionLevel 
 */
public DiskWriterThread(BlockingQueue<Span> outstanding,TreeMap<Long,File> files,boolean buffer,int compressionLevel){
  this.outstanding=outstanding;
  this.files=files;
  this.doBuffer=buffer;
  this.compressionLevel=compressionLevel;
}
 

/** 
 * If we hit our limit, we should signal for more connections to be created on the fly
 * @throws SQLException
 */
@Test public void testGetConnectionLimitsHit() throws SQLException {
  reset(mockPartition,mockConnectionHandles,mockConnection);
  expect(mockConfig.getPoolAvailabilityThreshold()).andReturn(0).anyTimes();
  expect(mockPartition.isUnableToCreateMoreTransactions()).andReturn(false).anyTimes();
  expect(mockPartition.getFreeConnections()).andReturn(mockConnectionHandles).anyTimes();
  expect(mockPartition.getMaxConnections()).andReturn(10).anyTimes();
  expect(mockPartition.getAvailableConnections()).andReturn(1).anyTimes();
  BlockingQueue<Object> bq=new ArrayBlockingQueue<Object>(1);
  bq.add(new Object());
  expect(mockPartition.getPoolWatchThreadSignalQueue()).andReturn(bq);
  expect(mockConnectionHandles.poll()).andReturn(mockConnection).once();
  mockConnection.renewConnection();
  expectLastCall().once();
  replay(mockPartition,mockConnectionHandles,mockConnection);
  testClass.getConnection();
  verify(mockPartition,mockConnectionHandles,mockConnection);
}
 

@SuppressWarnings("unchecked") @Before public void setup(){
  catchup=new CatchUpService();
  messageDirector=mock(MessageDirector.class);
  enhancementConductor=mock(EnhancementConductor.class);
  when(enhancementConductor.isEmpty()).thenReturn(true);
  List<EnhancementMessage> collisionList=mock(List.class);
  when(collisionList.size()).thenReturn(0);
  when(enhancementConductor.getCollisionList()).thenReturn(collisionList);
  BlockingQueue<EnhancementMessage> pidQueue=mock(BlockingQueue.class);
  when(pidQueue.size()).thenReturn(0);
  when(enhancementConductor.getPidQueue()).thenReturn(pidQueue);
  FailedObjectHashMap failedPids=mock(FailedObjectHashMap.class);
  when(failedPids.size()).thenReturn(0);
  when(enhancementConductor.getFailedPids()).thenReturn(failedPids);
  techmd=mock(TechnicalMetadataEnhancementService.class);
  image=mock(ImageEnhancementService.class);
  delay=new DelayService();
  DelayEnhancement.init();
  services=new ArrayList<ObjectEnhancementService>();
  services.add(techmd);
  services.add(image);
  catchup.setMessageDirector(messageDirector);
  catchup.setenhancementConductor(enhancementConductor);
  catchup.setServices(services);
  catchup.setCatchUpCheckDelay(100L);
  catchup.setEnabled(true);
}
 

public LocalToRemoteHdfsMover(BlockingQueue<String> fileQueue,Configuration conf){
  this.fileQueue=fileQueue;
  this.conf=conf;
  this.setDaemon(true);
  this.setName("LocalToRemoteHdfsMover");
  this.start();
}
 

public RequestThread(Window<Integer,String,String> window,BlockingQueue<Integer> requestQueue,int id,int requestsPerThread){
  this.window=window;
  this.requestQueue=requestQueue;
  this.id=id;
  this.requestsPerThread=requestsPerThread;
}
 

public final synchronized void reorderQueue(BlockingQueue<Runnable> queue){
  LOG.info("Reordering a FileGrabber work queue (contains " + queue.size() + " potential datasets)");
  List<Runnable> workList=new ArrayList<Runnable>(queue.size());
  queue.drainTo(workList);
  Collections.sort(workList,new Comparator<Runnable>(){
    @Override public int compare(    Runnable o1,    Runnable o2){
      WorkEntry w1=(WorkEntry)o1;
      WorkEntry w2=(WorkEntry)o2;
      return Long.compare(w1.getLatestFileTimestamp(),w2.getLatestFileTimestamp());
    }
  }
);
  queue.addAll(workList);
}
 

@Test public void compareQueues() throws Exception {
  if ("true".equalsIgnoreCase(System.getProperty("com.github.coderplay.util.concurrent.runQueueTests","true"))) {
    for (int producerThread=1; producerThread <= 16; producerThread<<=1) {
      for (int consumerThread=1; consumerThread <= 16; consumerThread<<=1) {
        for (int i=0; i < RUNS; i++) {
          System.out.format("Producer threads:%d, consumer threads: %d, run %d\n",Integer.valueOf(producerThread),Integer.valueOf(consumerThread),Integer.valueOf(i));
          for (          Entry<String,BlockingQueue<Long>> entry : queueMap.entrySet()) {
            runOneQueue(entry.getKey(),entry.getValue(),producerThread,consumerThread);
          }
          runOneQueue("FastArrayBlockingQueue",getFastArrayBlockingQueue(BUFFER_SIZE,producerThread,consumerThread),producerThread,consumerThread);
        }
      }
    }
  }
}
 

public FilesProcessor(final String name,final Log log,final BlockingQueue<NativeCodeFile> classesToProcess,final ConcurrentLinkedQueue<NativeCodeFile> processedClasses,final int numberOfProcessorThreads,final Object monitor){
  this.name=name;
  this.log=log;
  this.classesToProcess=classesToProcess;
  this.processedClasses=processedClasses;
  this.numberOfProcessorThreads=numberOfProcessorThreads;
  this.monitor=monitor;
}
 

public Collection<BlockingQueue<SAMRecord>> getQueuesForQuery(String sequenceName,int start,int end,boolean overlaping,int nofQueues){
  List<BlockingQueue<SAMRecord>> list=new LinkedList<BlockingQueue<SAMRecord>>();
  int step=(int)Math.ceil(((float)end - start) / nofQueues);
  for (int i=0; i < nofQueues; i++) {
    BlockingQueue<SAMRecord> q=getBlockingQueueForQuery(sequenceName,start + (step * i),start + (step * (i + 1)),overlaping);
    list.add(q);
  }
  return list;
}
 

@Test public void testConnectionState() throws Exception {
  Timing timing=new Timing();
  CuratorFramework client=CuratorFrameworkFactory.newClient(server.getConnectString(),timing.session(),timing.connection(),new RetryOneTime(1));
  try {
    final BlockingQueue<ConnectionState> queue=new LinkedBlockingQueue<ConnectionState>();
    ConnectionStateListener listener=new ConnectionStateListener(){
      @Override public void stateChanged(      CuratorFramework client,      ConnectionState newState){
        queue.add(newState);
      }
    }
;
    client.getConnectionStateListenable().addListener(listener);
    client.start();
    Assert.assertEquals(queue.poll(timing.multiple(4).seconds(),TimeUnit.SECONDS),ConnectionState.CONNECTED);
    server.stop();
    Assert.assertEquals(queue.poll(timing.multiple(4).seconds(),TimeUnit.SECONDS),ConnectionState.SUSPENDED);
    Assert.assertEquals(queue.poll(timing.multiple(4).seconds(),TimeUnit.SECONDS),ConnectionState.LOST);
  }
  finally {
    Closeables.closeQuietly(client);
  }
}
 

@SuppressWarnings("unchecked") @Before public void setUp() throws Exception {
  context=new ClassPathXmlApplicationContext("TestMPContainer.xml");
  container=(MpContainer)context.getBean("container");
  assertNotNull(container.getSerializer());
  inputQueue=(BlockingQueue<Object>)context.getBean("inputQueue");
  outputQueue=(BlockingQueue<Object>)context.getBean("outputQueue");
}
 

public FizzBuzzQueueProcessor(final FizzBuzzStep fizzBuzzStep,final BlockingQueue<Long> fizzInputQueue,final BlockingQueue<Long> buzzInputQueue,final BlockingQueue<Boolean> fizzOutputQueue,final BlockingQueue<Boolean> buzzOutputQueue,final long count){
  this.fizzBuzzStep=fizzBuzzStep;
  this.fizzInputQueue=fizzInputQueue;
  this.buzzInputQueue=buzzInputQueue;
  this.fizzOutputQueue=fizzOutputQueue;
  this.buzzOutputQueue=buzzOutputQueue;
  this.count=count;
}
 

public FizzBuzzQueueConsumer(final FizzBuzzStep fizzBuzzStep,final BlockingQueue<Long> fizzInputQueue,final BlockingQueue<Long> buzzInputQueue,final BlockingQueue<Boolean> fizzOutputQueue,final BlockingQueue<Boolean> buzzOutputQueue){
  this.fizzBuzzStep=fizzBuzzStep;
  this.fizzInputQueue=fizzInputQueue;
  this.buzzInputQueue=buzzInputQueue;
  this.fizzOutputQueue=fizzOutputQueue;
  this.buzzOutputQueue=buzzOutputQueue;
}
 

private void checkProblemFactChanges(){
  BlockingQueue<ProblemFactChange> problemFactChangeQueue=basicPlumbingTermination.getProblemFactChangeQueue();
  if (!problemFactChangeQueue.isEmpty()) {
    solverScope.setRestartSolver(true);
    solverScope.setWorkingSolutionFromBestSolution();
    Score score=null;
    int count=0;
    ProblemFactChange problemFactChange=problemFactChangeQueue.poll();
    while (problemFactChange != null) {
      score=doProblemFactChange(problemFactChange);
      count++;
      problemFactChange=problemFactChangeQueue.poll();
    }
    bestSolutionRecaller.updateBestSolution(solverScope,solverScope.getWorkingSolution().cloneSolution());
    logger.info("Done {} ProblemFactChange(s): new score ({}) possibly uninitialized. Restarting solver.",count,score);
  }
}
 

/** 
 * Creates a new loader state.
 * @param taskList The loader.
 * @param queue The queue of tasks to execute.
 * @param app The application.
 * @param errorListener The error listener.
 */
public LoaderState(Loader taskList,BlockingQueue<LoadTask> queue,Application app,LoaderErrorListener errorListener){
  super("__loader");
  this.tasksTotal=queue.size();
  this.taskList=taskList;
  this.queue=queue;
  this.app=app;
  this.errorListener=errorListener;
}
 

/** 
 * The implementation for getPackets() function. This is a blocking API.
 * @param sessionid
 * @return A list of packets associated with the session
 */
@Override @LogMessageDocs({@LogMessageDoc(level="ERROR",message="Interrupted while waiting for session start",explanation="The thread was interrupted waiting " + "for the packet streamer session to start",recommendation=LogMessageDoc.CHECK_CONTROLLER),@LogMessageDoc(level="ERROR",message="Interrupted while waiting for packets",explanation="The thread was interrupted waiting " + "for packets",recommendation=LogMessageDoc.CHECK_CONTROLLER)}) public List<ByteBuffer> getPackets(String sessionid) throws org.apache.thrift.TException {
  List<ByteBuffer> packets=new ArrayList<ByteBuffer>();
  int count=0;
  while (!msgQueues.containsKey(sessionid) && count++ < 100) {
    log.debug("Queue for session {} doesn't exist yet.",sessionid);
    try {
      Thread.sleep(100);
    }
 catch (    InterruptedException e) {
      log.error("Interrupted while waiting for session start");
    }
  }
  if (count < 100) {
    SessionQueue pQueue=msgQueues.get(sessionid);
    BlockingQueue<ByteBuffer> queue=pQueue.getQueue();
    try {
      packets.add(queue.take());
      queue.drainTo(packets);
    }
 catch (    InterruptedException e) {
      log.error("Interrupted while waiting for packets");
    }
  }
  return packets;
}
 

/** 
 * Cleanup a file after it has been successfully processed. This can through both IOExceptions and runtime exceptions due to Preconditions failures. According to the link below, Solaris (I assume POSIX/linux) does atomic rename but Windows does not guarantee it. http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=4017593 To be truly correct, I need to check the return value (will likely fail in unix if moving from one volume to another instead of just within same volume)
 */
synchronized void changeState(String tag,State oldState,State newState) throws IOException {
  DFOData data=table.get(tag);
  Preconditions.checkArgument(data != null,"Tag " + tag + " has no data");
  Preconditions.checkArgument(tag.equals(data.tag),"Data associated with tag didn't match tag " + tag);
  if (LOG.isDebugEnabled()) {
    LOG.debug("Change " + data.s + "/"+ oldState+ " to "+ newState+ " : "+ tag);
  }
  if (oldState == null) {
    oldState=data.s;
  }
  Preconditions.checkState(data.s == oldState,"Expected state to be " + oldState + " but was "+ data.s);
  if (oldState == State.ERROR) {
    throw new IllegalStateException("Cannot move from error state");
  }
  if (newState == State.SENT) {
    getQueue(oldState).remove(tag);
    File sentFile=getFile(tag);
    data.s=newState;
    if (!sentFile.delete()) {
      LOG.error("Couldn't delete " + sentFile + " - can be safely manually deleted");
    }
    return;
  }
  File orig=getFile(tag);
  File newf=new File(getDir(newState),tag);
  boolean success=orig.renameTo(newf);
  if (!success) {
    throw new IOException("Move  " + orig + " -> "+ newf+ "failed!");
  }
  LOG.debug("old state is " + oldState);
  getQueue(oldState).remove(tag);
  BlockingQueue<String> q=getQueue(newState);
  if (q != null) {
    q.add(tag);
  }
  data.s=newState;
}
 

/** 
 * @param name
 * @param password
 * @param connectionTimeout
 * @throws IOException
 */
public TwitterStreamingConnection(String name,String password,int connectionTimeout) throws IOException {
  httpClient=new HttpClient();
  httpClient.getHttpConnectionManager().getParams().setConnectionTimeout(connectionTimeout);
  httpClient.getHttpConnectionManager().getParams().setSoTimeout(10 * 1000);
  httpClient.getParams().setAuthenticationPreemptive(true);
  httpClient.getState().setCredentials(AuthScope.ANY,new UsernamePasswordCredentials(name,password));
  doOpen();
  Executors.newSingleThreadExecutor(new ThreadFactory(){
    @Override public Thread newThread(    Runnable runnable){
      return new Thread(runnable,"TwitterStreamingConnection");
    }
  }
).execute(new Runnable(){
    @Override public void run(){
      BlockingQueue<String> queue=TwitterStreamingConnection.this.queue;
      String line;
      while ((line=readLine()) != null) {
        queue.add(line);
      }
    }
  }
);
}
 

/** 
 * Cleanup a file after it has been successfully processed. This can through both IOExceptions and runtime exceptions due to Preconditions failures. According to the link below, Solaris (I assume POSIX/linux) does atomic rename but Windows does not guarantee it. http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=4017593 To be truly correct, I need to check the return value (will likely fail in unix if moving from one volume to another instead of just within same volume)
 */
synchronized void changeState(String tag,State oldState,State newState) throws IOException {
  DFOData data=table.get(tag);
  Preconditions.checkArgument(data != null,"Tag " + tag + " has no data");
  Preconditions.checkArgument(tag.equals(data.tag),"Data associated with tag didn't match tag " + tag);
  if (LOG.isDebugEnabled()) {
    LOG.debug("Change " + data.s + "/"+ oldState+ " to "+ newState+ " : "+ tag);
  }
  if (oldState == null) {
    oldState=data.s;
  }
  Preconditions.checkState(data.s == oldState,"Expected state to be " + oldState + " but was "+ data.s);
  if (oldState == State.ERROR) {
    throw new IllegalStateException("Cannot move from error state");
  }
  if (newState == State.SENT) {
    getQueue(oldState).remove(tag);
    File sentFile=getFile(tag);
    data.s=newState;
    if (!sentFile.delete()) {
      LOG.error("Couldn't delete " + sentFile + " - can be safely manually deleted");
    }
    return;
  }
  File orig=getFile(tag);
  File newf=new File(getDir(newState),tag);
  boolean success=orig.renameTo(newf);
  if (!success) {
    throw new IOException("Move  " + orig + " -> "+ newf+ "failed!");
  }
  LOG.debug("old state is " + oldState);
  getQueue(oldState).remove(tag);
  BlockingQueue<String> q=getQueue(newState);
  if (q != null) {
    q.add(tag);
  }
  data.s=newState;
}
 

public FlowLauncher(JobRequest request,BlockingQueue<JobResponse> queue){
  super();
  this.request=request;
  this.queue=queue;
  response=request.createResponse();
}
 

/** 
 * Tests  {@link GerritHandler#addListener(GerritEventListener)}. With 10000 listeners added by 10 threads at the same time.
 * @throws Exception if so.
 */
@Test public void testAddListenerManyAtTheSameTime() throws Exception {
  final int nrOfListeners=100000;
  BlockingQueue<Runnable> listeners=new LinkedBlockingQueue<Runnable>(nrOfListeners);
  System.out.print("Creating Listeners");
  for (int i=0; i < nrOfListeners; i++) {
    listeners.add(new Runnable(){
      GerritEventListener listener=new ListenerMock();
      @Override public void run(){
        handler.addListener(listener);
      }
    }
);
    if (i % 1000 == 0) {
      System.out.print(".");
    }
  }
  System.out.println(".Done!");
  ThreadPoolExecutor executor=new ThreadPoolExecutor(100,100,1,TimeUnit.MINUTES,listeners);
  executor.prestartAllCoreThreads();
  executor.shutdown();
  do {
    System.out.printf("Waiting for listeners to be added...Running#: %5d  Left#: %5d  Count#: %5d\n",executor.getActiveCount(),listeners.size(),handler.getEventListenersCount());
  }
 while (!executor.awaitTermination(1,TimeUnit.SECONDS));
  System.out.printf("              Listeners are added...Running#: %5d  Left#: %5d  Count#: %5d\n",executor.getActiveCount(),listeners.size(),handler.getEventListenersCount());
  assertEquals(nrOfListeners,handler.getEventListenersCount());
}
 

@Inject public PipeliningBatchService(final Grid grid,final IndexProvider indexes,final FileSystem fs,final TransformProvider transforms){
  super(indexes);
  final BlockingQueue<Task> prepQ=grid.queue("grouperfish_prepare");
  final BlockingQueue<Task> runQ=grid.queue("grouperfish_run");
  final BlockingQueue<Task> putQ=grid.queue("grouperfish_putresult");
  final BlockingQueue<Task> cleanupQ=grid.queue("grouperfish_cleanup");
  final BlockingQueue<Task> failQ=grid.queue("grouperfish_fail");
  workers=new ImmutableList.Builder<Worker>().add(new Worker(failQ,prepQ,runQ,new FetchHandler(fs,indexes))).add(new Worker(failQ,runQ,putQ,new RunHandler(fs,transforms))).add(new Worker(failQ,putQ,cleanupQ,new PutHandler(grid,fs))).add(new Worker(failQ,cleanupQ,null,new CleanupHandler(fs))).build();
  prepareQueue=prepQ;
  log.info("Instantiated service: {}",getClass().getSimpleName());
}
 

Publisher(Configuration conf,BlockingQueue<Result> results){
  pool=Executors.newCachedThreadPool();
  this.conf=conf;
  this.results=results;
  setName(this.getClass().getSimpleName());
  setDaemon(true);
}
 

/** 
 * Get a reference to the specified table from the pool. <p> Create a new one if one is not available.
 * @param tableName
 * @return a reference to the specified table
 * @throws RuntimeException if there is a problem instantiating the HTable
 */
@Override public HTable getTable(String tableName){
  BlockingQueue<HTable> queue=tables.get(tableName);
  if (queue == null) {
synchronized (tables) {
      queue=tables.get(tableName);
      if (queue == null) {
        queue=new LinkedBlockingQueue<HTable>(this.maxSize);
        for (int i=0; i < this.maxSize; ++i) {
          queue.add(this.newHTable(tableName));
        }
        tables.put(tableName,queue);
      }
    }
  }
  try {
    return queue.take();
  }
 catch (  Exception ex) {
    return null;
  }
}
 

public AsyncExecutorImpl(final String poolName,final int minThreadCount,final int maxThreadCount,final long keepAliveSecs){
  this.poolName=poolName;
  this.minThreadCount=minThreadCount;
  this.maxThreadCount=maxThreadCount;
  this.keepAliveSecs=keepAliveSecs;
  final BlockingQueue<Runnable> backingQueue=new LinkedBlockingQueue<Runnable>();
  final String name="Async exec pool " + this.getName();
  this.threadPoolExecutor=new LocalThreadPoolExecutor(minThreadCount,maxThreadCount,keepAliveSecs,TimeUnit.SECONDS,backingQueue,new NamedThreadFactory(name));
}
 

public Thread tryMatchInThread(final CharSequence input,final String regex,final BlockingQueue<Object> atFinish){
  Thread t=new Thread(){
    public void run(){
      boolean result;
      try {
        result=tryMatch(input,regex);
      }
 catch (      Exception e) {
        atFinish.offer(e);
        return;
      }
      atFinish.offer(result);
    }
  }
;
  t.start();
  return t;
}
 

public LocalToRemoteHdfsMover(BlockingQueue<String> fileQueue,Configuration conf){
  this.fileQueue=fileQueue;
  this.conf=conf;
  this.setDaemon(true);
  this.setName("LocalToRemoteHdfsMover");
  this.start();
}
 

public LocalToRemoteHdfsMover(String group,BlockingQueue<String> fileQueue,Configuration conf){
  this.group=group;
  this.fileQueue=fileQueue;
  this.conf=conf;
  this.setDaemon(true);
  this.setName("LocalToRemoteHdfsMover-" + group);
  this.start();
}
 

public ConsumerIterator(String topic,BlockingQueue<FetchedDataChunk> queue,int consumerTimeoutMs,Decoder<T> decoder){
  super();
  this.topic=topic;
  this.queue=queue;
  this.consumerTimeoutMs=consumerTimeoutMs;
  this.decoder=decoder;
}
 

private static <T>T safeTake(BlockingQueue<T> queue){
  while (true) {
    try {
      return queue.take();
    }
 catch (    InterruptedException e) {
    }
  }
}
 

/** 
 * {@inheritDoc} 
 */
protected void doPublish(final ExtLogRecord record){
switch (state) {
case 0:
{
      if (stateUpdater.compareAndSet(this,0,1)) {
        thread.start();
      }
    }
case 1:
{
    break;
  }
default :
{
  return;
}
}
final BlockingQueue<ExtLogRecord> recordQueue=this.recordQueue;
record.copyAll();
if (overflowAction == OverflowAction.DISCARD) {
recordQueue.offer(record);
}
 else {
try {
recordQueue.put(record);
}
 catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
}
}
 

/** 
 * Set the configured work queue.  If the service is already started, the change will take effect upon next restart.
 * @param workQueue the work queue
 */
public synchronized void setWorkQueue(final BlockingQueue<Runnable> workQueue){
  if (workQueue == null) {
    setWorkQueue(new ArrayBlockingQueue<Runnable>(DEFAULT_QUEUE_LENGTH));
  }
  this.workQueue=workQueue;
}
 

public PyScriptExecutor(BlockingQueue taskQueue,String scriptsPath){
  super(taskQueue,scriptsPath);
  this.state=Py.getSystemState();
  this.state.path.insert(0,new PyString(scriptsPath));
  this.interpreter=new PythonInterpreter(null,state);
}
 

/** 
 * Register BroadcastReceivers programmaticaly because doing it from manifest would make K-9 auto-start. We don't want auto-start because the initialization sequence isn't safe while some events occur (SD card unmount).
 */
protected void registerReceivers(){
  final StorageGoneReceiver receiver=new StorageGoneReceiver();
  final IntentFilter filter=new IntentFilter();
  filter.addAction(Intent.ACTION_MEDIA_EJECT);
  filter.addAction(Intent.ACTION_MEDIA_UNMOUNTED);
  filter.addDataScheme("file");
  final BlockingQueue<Handler> queue=new SynchronousQueue<Handler>();
  new Thread(new Runnable(){
    @Override public void run(){
      Looper.prepare();
      try {
        queue.put(new Handler());
      }
 catch (      InterruptedException e) {
        Log.e(K9.LOG_TAG,"",e);
      }
      Looper.loop();
    }
  }
,"Unmount-thread").start();
  try {
    final Handler storageGoneHandler=queue.take();
    registerReceiver(receiver,filter,null,storageGoneHandler);
    Log.i(K9.LOG_TAG,"Registered: unmount receiver");
  }
 catch (  InterruptedException e) {
    Log.e(K9.LOG_TAG,"Unable to register unmount receiver",e);
  }
  registerReceiver(new ShutdownReceiver(),new IntentFilter(Intent.ACTION_SHUTDOWN));
  Log.i(K9.LOG_TAG,"Registered: shutdown receiver");
}
 

protected Object doExecute() throws Exception {
  final PrintStream out=System.out;
  Iterable<PaxLoggingEvent> le=this.logService.getEvents(entries == 0 ? Integer.MAX_VALUE : entries);
  for (  PaxLoggingEvent event : le) {
    printEvent(out,event);
  }
  final BlockingQueue<PaxLoggingEvent> queue=new LinkedBlockingQueue<PaxLoggingEvent>();
  PaxAppender appender=new PaxAppender(){
    public void doAppend(    PaxLoggingEvent event){
      queue.add(event);
    }
  }
;
  try {
    logService.addAppender(appender);
    for (; ; ) {
      PaxLoggingEvent event=queue.take();
      printEvent(out,event);
    }
  }
 catch (  InterruptedException e) {
  }
 finally {
    logService.removeAppender(appender);
  }
  out.println();
  return null;
}
 

/** 
 * @param target   Queue que se "wrappea". Es quien en realidad tiene implementada las operaciones
 * @param timeout  Como las {@link BlockingQueue} retornan null si no hay elementos, se le debe indicar un timeout de espera para volver a ver si hay elementos (y de esta forma "blockear"). Est dicho en milisegundos.  
 */
public BlockingQueueJobQueue(final BlockingQueue<T> target,final long timeout){
  Validate.notNull(target,"target is null");
  Validate.isTrue(timeout > 0,"timeout must be positive");
  this.target=target;
  this.timeout=timeout;
}
 

public AmqpRawConsumer(final AmqpRawConsumerConfiguration configuration,final BlockingQueue<AmqpRawMessage> buffer){
  super(configuration);
  this.exchange=Preconditions.checkNotNull(!Strings.isNullOrEmpty(configuration.exchange) ? configuration.exchange : AmqpRawConsumerConfiguration.defaultExchange);
  Preconditions.checkArgument(!this.exchange.isEmpty());
  this.queue=Preconditions.checkNotNull(!Strings.isNullOrEmpty(configuration.queue) ? configuration.queue : "");
  this.routingKey=Preconditions.checkNotNull(!Strings.isNullOrEmpty(configuration.routingKey) ? configuration.routingKey : AmqpRawConsumerConfiguration.defaultRoutingKey);
  Preconditions.checkArgument(!this.routingKey.isEmpty());
  this.buffer=(buffer != null) ? buffer : new LinkedBlockingDeque<AmqpRawMessage>();
}
 

@Test public void testAppendShutdownLoadRecover() throws Exception {
  this.recoverStorageManager.shutdown();
  final MetaClientConfig metaClientConfig=new MetaClientConfig();
  metaClientConfig.setRecoverMessageIntervalInMills(Integer.MAX_VALUE);
  this.recoverStorageManager=new RecoverStorageManager(metaClientConfig,this.subscribeInfoManager);
  this.recoverStorageManager.start(metaClientConfig);
  final String group="dennis";
  final BlockingQueue<Message> queue=new ArrayBlockingQueue<Message>(1024);
  this.subscribeInfoManager.subscribe("test",group,1024 * 1024,new MessageListener(){
    @Override public void recieveMessages(    final Message message){
      queue.offer(message);
    }
    @Override public Executor getExecutor(){
      return null;
    }
  }
);
  for (int i=0; i < 100; i++) {
    final Message msg2=new Message("test",("hello" + i).getBytes());
    MessageAccessor.setId(msg2,i);
    this.recoverStorageManager.append(group,msg2);
  }
  this.recoverStorageManager.shutdown();
  metaClientConfig.setRecoverMessageIntervalInMills(1000);
  this.recoverStorageManager=new RecoverStorageManager(metaClientConfig,this.subscribeInfoManager);
  this.recoverStorageManager.start(metaClientConfig);
  while (queue.size() < 100) {
    Thread.sleep(1000);
  }
  for (  final Message msg : queue) {
    assertEquals("hello" + msg.getId(),new String(msg.getData()));
  }
  assertEquals(0,this.recoverStorageManager.getOrCreateStore("test",group).size());
}
 

/** 
 * Creates the merger threads for the given tiles.
 * @param tiles the areas of the precompiled tiles
 * @param tilesCountdown the countdown that should be decreased after a tile is finished
 * @param saveQueue the queue the merged results should be added to
 * @return the preinitialized but not started mergers
 */
private List<PrecompSeaMerger> createMergers(Collection<uk.me.parabola.imgfmt.app.Area> tiles,CountDownLatch tilesCountdown,BlockingQueue<Entry<String,List<Way>>> saveQueue){
  List<PrecompSeaMerger> mergers=new ArrayList<PrecompSeaMerger>();
  for (  uk.me.parabola.imgfmt.app.Area bounds : tiles) {
    Rectangle mergeBounds=new Rectangle(bounds.getMinLong(),bounds.getMinLat(),bounds.getWidth(),bounds.getHeight());
    String tileKey=bounds.getMinLat() + "_" + bounds.getMinLong();
    PrecompSeaMerger merger=new PrecompSeaMerger(mergeBounds,tileKey,tilesCountdown,saveQueue);
    merger.setExecutorService(service);
    mergers.add(merger);
  }
  return mergers;
}
 

@State public void handleInvite(final IncomingCall call) throws Exception {
  call.answer();
  game=new DemoGameServer(call);
  final Prompt<Call> agePrompt=call.prompt("Welcome phone sweeper. " + "Press 1 if you are over 18, press 2 if you are under 18.","1,2",0);
  final Input<Call> input=agePrompt.getInput();
  if (!"1".equals(input.get().getConcept())) {
    final Output<Call> o=call.output("Sorry, you're too young");
    o.get();
    call.disconnect();
    return;
  }
  final MyListener listener=new MyListener();
  final BlockingQueue<InputCompleteEvent<Call>> queue=listener.getQueue();
  call.addObserver(listener);
  call.output("Ready Go");
  call.input(new DigitInputCommand());
  while (true) {
    final InputCompleteEvent<Call> event=queue.take();
    if (event.hasMatch() && event.getConcept() != null) {
      final int command=Integer.parseInt(event.getConcept());
switch (command) {
case 2:
        game.move(Direction.UP);
      break;
case 4:
    game.move(Direction.LEFT);
  break;
case 6:
game.move(Direction.RIGHT);
break;
case 8:
game.move(Direction.DOWN);
break;
}
}
}
}
 

private static int javaVersion0(){
  try {
    Class.forName("android.app.Application",false,ClassLoader.getSystemClassLoader());
    return 6;
  }
 catch (  Exception e) {
  }
  try {
    Class.forName("java.util.concurrent.LinkedTransferQueue",false,BlockingQueue.class.getClassLoader());
    return 7;
  }
 catch (  Exception e) {
  }
  return 6;
}
 

public MultiThreader(){
  int numThreads=Runtime.getRuntime().availableProcessors();
  BlockingQueue<Runnable> queue=new ArrayBlockingQueue<Runnable>(1);
  exec=new ThreadPoolExecutor(numThreads,numThreads,0,TimeUnit.SECONDS,queue,new ThreadFactory(){
    private final AtomicInteger seq=new AtomicInteger(1);
    @Override public Thread newThread(    Runnable runnable){
      return new Thread("MultiThreader-" + seq.getAndIncrement()){
        @Override public void run(){
          try {
            super.run();
          }
 catch (          Throwable t) {
            failure.compareAndSet(null,t);
          }
        }
      }
;
    }
  }
);
  failure=new AtomicReference<Throwable>();
}
 

@Override public ExecutorService initClusterTaskExecutor(RejectedExecutionHandler handler) throws PlatformException {
  BlockingQueue<Runnable> actionQueue=new ArrayBlockingQueue<Runnable>(getActionQueueSize());
  taskExec=new ThreadPoolExecutor(getActionMinThreads(),getActionMaxThreads(),getActionThreadTimeout(),TimeUnit.SECONDS,actionQueue,new ThreadFactory(){
    private final ThreadFactory factory=java.util.concurrent.Executors.defaultThreadFactory();
    private long id=0;
    @Override public Thread newThread(    Runnable r){
      Thread t=factory.newThread(r);
      t.setName("ODE-X Cluster Action Executor - " + ++id);
      t.setDaemon(true);
      return t;
    }
  }
,handler);
  taskExec.allowCoreThreadTimeOut(true);
  return taskExec;
}
 

/** 
 * Gets the single running instance of AudioPusher with or adding a new queue.
 * @param path filepath of the file to be used as input
 * @param id identifying value (jingle id, port, etc.) of who the queue belongs to
 * @param q queue of audio frames.
 * @return the new or currently active instance
 */
public static AudioPusher getInstance(String path,String id,BlockingQueue<short[]> q){
  if (audioPusher == null) {
    audioPusher=new AudioPusher(path,id,q);
  }
 else {
    audioPusher.addQueue(id,q);
  }
  return audioPusher;
}
 

/** 
 * Used in  {@link WebSocketServer} and {@link WebSocketClient}.
 * @param socketChannel The <tt>SocketChannel</tt> instance to read andwrite to. The channel should already be registered with a Selector before construction of this object.
 * @param bufferQueue The Queue that we should use to buffer data thathasn't been sent to the client yet.
 * @param listener The {@link WebSocketListener} to notify of events whenthey occur.
 */
WebSocket(SocketChannel socketChannel,BlockingQueue<ByteBuffer> bufferQueue,WebSocketListener listener){
  this.socketChannel=socketChannel;
  this.bufferQueue=bufferQueue;
  this.handshakeComplete=false;
  this.remoteHandshake=this.currentFrame=null;
  this.buffer=ByteBuffer.allocate(1);
  this.wsl=listener;
}
 

public SharedMemoryComputationEmitter(java.util.concurrent.Executor runnableExecutor,Enrichment enrichment,ContextEvent<ComputationInstanceContext> computationInstanceContextEvent,Context parentContext,AtomicInteger maxId,BlockingQueue<Future<L>> futures){
  this.runnableExecutor=runnableExecutor;
  this.enrichment=enrichment;
  this.computationInstanceContextEvent=computationInstanceContextEvent;
  this.maxId=maxId;
  this.parentContext=parentContext;
  this.futures=futures;
}
 

/** 
 * Creates a new cache triage dispatcher thread.  You must call  {@link #start()}in order to begin processing.
 * @param cacheQueue Queue of incoming requests for triage
 * @param networkQueue Queue to post requests that require network to
 * @param cache Cache interface to use for resolution
 * @param delivery Delivery interface to use for posting responses
 */
public CacheDispatcher(BlockingQueue<Request> cacheQueue,BlockingQueue<Request> networkQueue,Cache cache,ResponseDelivery delivery){
  mCacheQueue=cacheQueue;
  mNetworkQueue=networkQueue;
  mCache=cache;
  mDelivery=delivery;
}
 

/** 
 * @hide
 */
public static ExecutorService newConfiguredThreadPool(){
  int corePoolSize=0;
  int maximumPoolSize=Configuration.ASYNC_THREAD_COUNT;
  long keepAliveTime=60L;
  TimeUnit unit=TimeUnit.SECONDS;
  BlockingQueue<Runnable> workQueue=new LinkedBlockingQueue<Runnable>();
  RejectedExecutionHandler handler=new ThreadPoolExecutor.CallerRunsPolicy();
  return new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime,unit,workQueue,handler);
}
 

public void testConsumerCancellationNotification() throws IOException, InterruptedException {
  final BlockingQueue<Boolean> result=new ArrayBlockingQueue<Boolean>(1);
  channel.queueDeclare(queue,false,true,false,null);
  Consumer consumer=new QueueingConsumer(channel){
    @Override public void handleCancel(    String consumerTag) throws IOException {
      try {
        result.put(true);
      }
 catch (      InterruptedException e) {
        fail();
      }
    }
  }
;
  channel.basicConsume(queue,consumer);
  channel.queueDelete(queue);
  assertTrue(result.take());
}
 

private BlockingQueue<Runnable> threadQueue(int size){
switch (size) {
case -1:
    return new LinkedBlockingQueue<Runnable>();
case 0:
  return new SynchronousQueue<Runnable>();
default :
return new ArrayBlockingQueue<Runnable>(size);
}
}
 

/** 
 * Method creates a thread to process a queue
 */
private Thread createUpdateThread(final BlockingQueue<UpdateRequest> queue,final int priority,String name){
  final Thread thread=new Thread(name){
    @Override public void run(){
      while (!treeViewer.getControl().isDisposed() && queue != null) {
        try {
          final UpdateRequest req=queue.take();
          if (req.getElement() == null && req.getIndex() == -1)           return;
          if (req instanceof BlankUpdateRequest)           return;
          final boolean ok=req.process();
          if (!ok)           break;
        }
 catch (        InterruptedException ne) {
          break;
        }
catch (        org.eclipse.swt.SWTException swtE) {
          queue.clear();
          break;
        }
catch (        Exception ne) {
          queue.clear();
          continue;
        }
      }
    }
  }
;
  thread.setPriority(priority);
  thread.setDaemon(true);
  thread.start();
  return thread;
}
 

public FizzBuzzQueueConsumer(final FizzBuzzStep fizzBuzzStep,final BlockingQueue<Long> fizzInputQueue,final BlockingQueue<Long> buzzInputQueue,final BlockingQueue<Boolean> fizzOutputQueue,final BlockingQueue<Boolean> buzzOutputQueue){
  this.fizzBuzzStep=fizzBuzzStep;
  this.fizzInputQueue=fizzInputQueue;
  this.buzzInputQueue=buzzInputQueue;
  this.fizzOutputQueue=fizzOutputQueue;
  this.buzzOutputQueue=buzzOutputQueue;
}
 

/** 
 * Connects to Skype client.
 * @param timeout the maximum time in milliseconds to connect.
 * @return Status the status after connecting.
 * @throws ConnectorException when connection can not be established.
 */
protected Status connect(int timeout) throws ConnectorException {
  if (!SkypeFramework.isRunning()) {
    setStatus(Status.NOT_RUNNING);
    return getStatus();
  }
  try {
    final BlockingQueue<String> queue=new LinkedBlockingQueue<String>();
    SkypeFrameworkListener initListener=new SkypeFrameworkListener(){
      public void notificationReceived(      String notification){
        if ("OK".equals(notification) || "CONNSTATUS OFFLINE".equals(notification) || "ERROR 68".equals(notification)) {
          try {
            queue.put(notification);
          }
 catch (          InterruptedException e) {
            Thread.currentThread().interrupt();
          }
        }
      }
    }
;
    setStatus(Status.PENDING_AUTHORIZATION);
    SkypeFramework.addSkypeFrameworkListener(initListener);
    SkypeFramework.sendCommand("NAME " + getApplicationName());
    String result=queue.take();
    SkypeFramework.removeSkypeFrameworkListener(initListener);
    if ("OK".equals(result)) {
      setStatus(Status.ATTACHED);
    }
 else     if ("CONNSTATUS OFFLINE".equals(result)) {
      setStatus(Status.NOT_AVAILABLE);
    }
 else     if ("ERROR 68".equals(result)) {
      setStatus(Status.REFUSED);
    }
    return getStatus();
  }
 catch (  InterruptedException e) {
    throw new ConnectorException("Trying to connect was interrupted.",e);
  }
}
 

/** 
 * Connects to Skype client.
 * @param timeout the maximum time in milliseconds to connect.
 * @return Status the status after connecting.
 * @throws ConnectorException when connection can not be established.
 */
protected Status connect(int timeout) throws ConnectorException {
  if (!SkypeFramework.isRunning()) {
    setStatus(Status.NOT_RUNNING);
    return getStatus();
  }
  try {
    final BlockingQueue<String> queue=new LinkedBlockingQueue<String>();
    SkypeFrameworkListener initListener=new SkypeFrameworkListener(){
      public void notificationReceived(      String notification){
        if ("OK".equals(notification) || "CONNSTATUS OFFLINE".equals(notification) || "ERROR 68".equals(notification)) {
          try {
            queue.put(notification);
          }
 catch (          InterruptedException e) {
            Thread.currentThread().interrupt();
          }
        }
      }
    }
;
    setStatus(Status.PENDING_AUTHORIZATION);
    SkypeFramework.addSkypeFrameworkListener(initListener);
    SkypeFramework.sendCommand("NAME " + getApplicationName());
    String result=queue.take();
    SkypeFramework.removeSkypeFrameworkListener(initListener);
    if ("OK".equals(result)) {
      setStatus(Status.ATTACHED);
    }
 else     if ("CONNSTATUS OFFLINE".equals(result)) {
      setStatus(Status.NOT_AVAILABLE);
    }
 else     if ("ERROR 68".equals(result)) {
      setStatus(Status.REFUSED);
    }
    return getStatus();
  }
 catch (  InterruptedException e) {
    throw new ConnectorException("Trying to connect was interrupted.",e);
  }
}
 

public void putTable(HTableInterface table){
  String tableName=Bytes.toString(table.getTableName());
  BlockingQueue<HTableInterface> idleQueue=idleTables.get(tableName);
  BlockingQueue<HTableInterface> busyQueue=busyTables.get(tableName);
synchronized (idleQueue) {
    if (idleQueue.size() >= maxIdle)     return;
    if (busyQueue.remove(table)) {
      idleQueue.add(table);
    }
  }
}
 

public SrpConnection(String host,int port,BlockingQueue<SrpConnection> queue){
  try {
    this.client=new RedisClient(host,port);
    this.queue=queue;
  }
 catch (  IOException e) {
    throw new RedisConnectionFailureException("Could not connect",e);
  }
}
 

protected ExecutorService initializeExecutor(ThreadFactory threadFactory,RejectedExecutionHandler rejectedExecutionHandler){
  BlockingQueue<Runnable> queue=createQueue(this.queueCapacity);
  ThreadPoolExecutor executor=new ThreadPoolExecutor(this.corePoolSize,this.maxPoolSize,this.keepAliveSeconds,TimeUnit.SECONDS,queue,threadFactory,rejectedExecutionHandler);
  if (this.allowCoreThreadTimeOut) {
    executor.allowCoreThreadTimeOut(true);
  }
  this.exposedExecutor=(this.exposeUnconfigurableExecutor ? Executors.unconfigurableExecutorService(executor) : executor);
  return executor;
}
 

/** 
 * Creates a new custom thread pool
 * @param pattern                  pattern of the thread name
 * @param name                     ${name} in the pattern name
 * @param corePoolSize             the core size
 * @param maxPoolSize              the maximum pool size
 * @param keepAliveTime            keep alive time
 * @param timeUnit                 keep alive time unit
 * @param maxQueueSize             the maximum number of tasks in the queue, use <tt>Integer.MAX_VALUE</tt> or <tt>-1</tt> to indicate unbounded
 * @param rejectedExecutionHandler the handler for tasks which cannot be executed by the thread pool.If <tt>null</tt> is provided then  {@link java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy CallerRunsPolicy} is used.
 * @param daemon                   whether the threads is daemon or not
 * @return the created pool
 * @throws IllegalArgumentException if parameters is not valid
 */
public static ExecutorService newThreadPool(final String pattern,final String name,int corePoolSize,int maxPoolSize,long keepAliveTime,TimeUnit timeUnit,int maxQueueSize,RejectedExecutionHandler rejectedExecutionHandler,final boolean daemon){
  if (maxPoolSize < corePoolSize) {
    throw new IllegalArgumentException("MaxPoolSize must be >= corePoolSize, was " + maxPoolSize + " >= "+ corePoolSize);
  }
  BlockingQueue<Runnable> queue;
  if (corePoolSize == 0 && maxQueueSize <= 0) {
    queue=new SynchronousQueue<Runnable>();
    corePoolSize=1;
    maxPoolSize=1;
  }
 else   if (maxQueueSize <= 0) {
    queue=new LinkedBlockingQueue<Runnable>();
  }
 else {
    queue=new LinkedBlockingQueue<Runnable>(maxQueueSize);
  }
  ThreadPoolExecutor answer=new ThreadPoolExecutor(corePoolSize,maxPoolSize,keepAliveTime,timeUnit,queue);
  answer.setThreadFactory(new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread answer=new Thread(r,getThreadName(pattern,name));
      answer.setDaemon(daemon);
      return answer;
    }
  }
);
  if (rejectedExecutionHandler == null) {
    rejectedExecutionHandler=new ThreadPoolExecutor.CallerRunsPolicy();
  }
  answer.setRejectedExecutionHandler(rejectedExecutionHandler);
  return answer;
}