Java Code Examples for java.util.concurrent.LinkedBlockingQueue

public ScriptExecutionPool(Class clazz,String scriptsPath,int numberOfThreads,int maxNumberOfTasks){
  taskQueue=new LinkedBlockingQueue(maxNumberOfTasks);
  Constructor executorsConstructor=null;
  try {
    executorsConstructor=clazz.getConstructor(BlockingQueue.class,String.class);
    executorsConstructor.setAccessible(true);
  }
 catch (  NoSuchMethodException e) {
    log.error(e.toString());
  }
  if (executorsConstructor != null) {
    for (int i=0; i < numberOfThreads; i++) {
      try {
        executors.add((ScriptExecutor)executorsConstructor.newInstance(taskQueue,scriptsPath));
      }
 catch (      Exception e) {
        log.error(e.toString());
      }
    }
    for (    ScriptExecutor executor : executors) {
      executor.start();
    }
  }
}
 

@Test public void testFIFO(){
  Queue q=new LinkedBlockingDeque();
  q.add("test");
  q.add("pest");
  assertEquals("test",q.poll());
  q=new LinkedBlockingQueue();
  q.add("test");
  q.add("pest");
  assertEquals("test",q.poll());
  Stack v=new Stack();
  v.add("test");
  v.add("pest");
  assertEquals("pest",v.pop());
}
 

protected void initExecutor(Map<String,Object> config){
  if (executor == null) {
    int threads=ConfigUtils.getInteger(config,MAX_POOL_SIZE,CFG_PREFIX + ".threads");
    if (threads <= 1) {
      executor=new Executor(){
        public void execute(        Runnable command){
          command.run();
        }
      }
;
    }
 else {
      ThreadFactory threadFactory=new RepositoryConnectorThreadFactory(getClass().getSimpleName());
      executor=new ThreadPoolExecutor(threads,threads,3,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory);
    }
  }
}
 

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);
}
 

EchoMachine(WriteCallback callback,boolean dumpWhenFull){
  this.callback=callback;
  this.dumpWhenFull=dumpWhenFull;
  dataQueue=new LinkedBlockingQueue<byte[]>(QUEUE_SIZE);
  handler=new Handler(this);
}
 

private KeyedBucket(String key){
  this.guard=new AtomicBoolean();
  this.queue=new LinkedBlockingQueue<Runnable>();
  this.key=key;
  this.waitTimeStart=new AtomicLong(0L);
  this.waitTimeEnd=new AtomicLong(0L);
}
 

/** 
 * Creates a new FastAGIConnectionHandler to handle the given FastAGI socket connection.
 * @param mappingStrategy    the strategy to use to determine which script to run.
 * @param asyncAgiStartEvent the AsyncAgiEvent that started this connection, must be a start sub event.
 * @param agiChannelFactory  The factory to use for creating new AgiChannel instances.
 * @throws IllegalArgumentException if asyncAgiStartEvent is not a start sub type".
 */
public AsyncAgiConnectionHandler(MappingStrategy mappingStrategy,AsyncAgiEvent asyncAgiStartEvent,AgiChannelFactory agiChannelFactory) throws IllegalArgumentException {
  super(mappingStrategy,agiChannelFactory);
  if (!asyncAgiStartEvent.isStart()) {
    throw new IllegalArgumentException("AsyncAgiEvent passed to AsyncAgiConnectionHandler is not a start sub event");
  }
  connection=(ManagerConnection)asyncAgiStartEvent.getSource();
  channelName=asyncAgiStartEvent.getChannel();
  environment=asyncAgiStartEvent.decodeEnv();
  asyncAgiEvents=new LinkedBlockingQueue<AsyncAgiEvent>();
  setIgnoreMissingScripts(true);
}
 

public SimpleHostConnectionPool(Host host,ConnectionFactory<CL> factory,ConnectionPoolMonitor monitor,ConnectionPoolConfiguration config,Listener<CL> listener){
  this.host=host;
  this.config=config;
  this.factory=factory;
  this.listener=listener;
  this.retryContext=config.getRetryBackoffStrategy().createInstance();
  this.latencyStrategy=config.getLatencyScoreStrategy().createInstance();
  this.badHostDetector=config.getBadHostDetector().createInstance();
  this.monitor=monitor;
  this.availableConnections=new LinkedBlockingQueue<Connection<CL>>();
}
 

@SuppressWarnings("unchecked") public JobExecutorManager(FlowManager allKnownFlows,JobManager jobManager,Mailman mailman,String jobSuccessEmail,String jobFailureEmail,int maxThreads){
  this.jobManager=jobManager;
  this.mailman=mailman;
  this.jobSuccessEmail=jobSuccessEmail;
  this.jobFailureEmail=jobFailureEmail;
  this.allKnownFlows=allKnownFlows;
  Multimap<String,JobExecution> typedMultiMap=HashMultimap.create();
  this.completed=Multimaps.synchronizedMultimap(typedMultiMap);
  this.executing=new ConcurrentHashMap<String,ExecutingJobAndInstance>();
  this.executor=new ThreadPoolExecutor(0,maxThreads,10,TimeUnit.SECONDS,new LinkedBlockingQueue(),new ExecutorThreadFactory());
}
 

public EnhancementConductor(){
  log.debug("Starting up Services Conductor");
  pidQueue=new LinkedBlockingQueue<EnhancementMessage>();
  collisionList=Collections.synchronizedList(new ArrayList<EnhancementMessage>());
  lockedPids=Collections.synchronizedSet(new HashSet<String>());
  activeMessages=Collections.synchronizedList(new ArrayList<EnhancementMessage>());
  finishedMessages=Collections.synchronizedList(new LimitedWindowList<EnhancementMessage>(maxFinishedMessages));
}
 

@Test public void simulatedMultithreadedProcessing() throws Exception {
  final Window<Integer,String,String> window=new Window<Integer,String,String>(5);
  final int requestThreadCount=8;
  final int requestsPerThread=10000;
  final BlockingQueue<Integer> requestQueue=new LinkedBlockingQueue<Integer>();
  RequestThread[] requestThreads=new RequestThread[requestThreadCount];
  for (int i=0; i < requestThreadCount; i++) {
    requestThreads[i]=new RequestThread(window,requestQueue,i,requestsPerThread);
  }
  ResponseThread responseThread=new ResponseThread(window,requestQueue,requestThreadCount * requestsPerThread);
  for (  RequestThread requestThread : requestThreads) {
    requestThread.start();
  }
  responseThread.start();
  for (  RequestThread requestThread : requestThreads) {
    requestThread.join();
  }
  responseThread.join();
  for (int i=0; i < requestThreadCount; i++) {
    if (requestThreads[i].throwable != null) {
      logger.error("",requestThreads[i].throwable);
    }
    Assert.assertNull("RequestThread " + i + " throwable wasn't null: "+ requestThreads[i].throwable,requestThreads[i].throwable);
  }
  if (responseThread.throwable != null) {
    logger.error("",responseThread.throwable);
  }
  Assert.assertNull("ResponseThread throwable wasn't null",responseThread.throwable);
  Assert.assertEquals(0,window.getSize());
}
 

public PollableSmppSessionHandler(){
  this.receivedPduRequests=new LinkedBlockingQueue<PduRequest>();
  this.receivedExpectedPduResponses=new LinkedBlockingQueue<PduAsyncResponse>();
  this.receivedUnexpectedPduResponses=new LinkedBlockingQueue<PduResponse>();
  this.throwables=new LinkedBlockingQueue<Throwable>();
  this.closedCount=new AtomicInteger();
}
 

protected void setUp() throws Exception {
  eventQueue=new LinkedBlockingQueue<String>();
  randomInt=(int)(Math.random() * 100000d);
  randomByte=(byte)(Math.random() * 127d);
  randomLong=randomLong();
  randomString=randomString(32);
}
 

public LocalQueueAndWriter(final CollectorConfig config,final String path,final EventWriter eventWriter,final WriterStats stats){
  this.queue=new LinkedBlockingQueue<Event>(config.getMaxQueueSize());
  this.eventWriter=eventWriter;
  this.stats=stats;
  this.executor=new FailsafeScheduledExecutor(1,path + "-HDFS-dequeuer");
  executor.submit(new LocalQueueWorker(queue,eventWriter,stats));
}
 

private void sendAndAssertOneMessage() throws Exception, InterruptedException {
  final String payload="dummy payload";
  _testKit.removeService("SimpleCamelService");
  final MockHandler simpleCamelService=_testKit.registerInOnlyService("SimpleCamelService");
  sendTextToQueue(payload,"testQueue");
  Thread.sleep(3000);
  final LinkedBlockingQueue<Exchange> recievedMessages=simpleCamelService.getMessages();
  assertThat(recievedMessages,is(notNullValue()));
  final Exchange recievedExchange=recievedMessages.iterator().next();
  assertThat(recievedExchange.getMessage().getContent(String.class),is(equalTo(payload)));
}
 

HBaseWriter(@Nonnull final HBaseWriterConfig hbaseWriterConfig,@Nonnull final Configuration hadoopConfig,@Nonnull final SpillController spillController){
  super(hbaseWriterConfig,hadoopConfig);
  Preconditions.checkNotNull(spillController,"spill controller not be null!");
  this.hbaseWriterConfig=hbaseWriterConfig;
  this.writeQueue=new LinkedBlockingQueue<Callable<Put>>(hbaseWriterConfig.getQueueLength());
  this.enqueueTimeout=hbaseWriterConfig.getEnqueueTimeout();
  this.spillController=spillController;
}
 

MongoWriter(final MongoWriterConfig mongoWriterConfig){
  this.mongoWriterConfig=mongoWriterConfig;
  this.collectionName=mongoWriterConfig.getCollectionName();
  this.writeQueue=new LinkedBlockingQueue<Callable<DBObject>>(mongoWriterConfig.getQueueLength());
  this.enqueueTimeout=mongoWriterConfig.getEnqueueTimeout();
}
 

/** 
 * Wait for a number of messages.
 * @param eventQueue event queue
 * @param numMessages number of messages
 */
private void waitFor(final LinkedBlockingQueue<Exchange> eventQueue,final int numMessages){
  long start=System.currentTimeMillis();
  while (System.currentTimeMillis() < start + _waitTimeout) {
    if (eventQueue.size() >= numMessages) {
      return;
    }
    sleep();
  }
  TestCase.fail("Timed out waiting on event queue length to be " + numMessages + " or greater.");
}
 

public void init(){
  workQueue=new LinkedBlockingQueue<EventWrapper>(maxQueueSize);
  for (  PrototypeWrapper pw : prototypeWrappers) {
    adviceLists.add(pw.advise());
  }
  Thread t=new Thread(this,"PEContainer");
  t.start();
  t=new Thread(new Watcher());
  t.start();
}
 

public TestTermIO() throws IOException {
  this.inner=new LinkedBlockingQueue<Integer>();
  this.outter=new LinkedBlockingQueue<String>();
  this.width=32;
  this.properties=new HashMap<String,String>();
}
 

@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);
  }
}
 

public BatchWriter(EmbeddedSolrServer solr,int batchSize,TaskID tid,int writerThreads,int queueSize){
  this.solr=solr;
  this.writerThreads=writerThreads;
  this.queueSize=queueSize;
  taskId=tid;
  batchPool=new ThreadPoolExecutor(writerThreads,writerThreads,5,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(queueSize),new ThreadPoolExecutor.CallerRunsPolicy());
  this.batchToWrite=new ArrayList<SolrInputDocument>(batchSize);
}
 

/** 
 * Creates a new PortHandler object.
 * @param ioPort
 * @param listener an object interested in receiving messages from the handlerin push mode
 */
public PortHandler(IOPort ioPort,PortListener listener){
  this.ioPort=ioPort;
  this.listener=listener;
  channelCount=getWidth();
  queue=new LinkedBlockingQueue<Token>();
  Nameable actor=ioPort.getContainer();
  if (actor != null) {
    actorInfo=((NamedObj)actor).getFullName();
  }
}
 

public void start(){
  ThreadFactory tf=new ThreadFactoryBuilder().setNameFormat("ContainerLauncher #%d").setDaemon(true).build();
  launcherPool=new ThreadPoolExecutor(INITIAL_POOL_SIZE,Integer.MAX_VALUE,1,TimeUnit.HOURS,new LinkedBlockingQueue<Runnable>(),tf);
  eventHandlingThread=new Thread(){
    @Override public void run(){
      ContainerLauncherEvent event=null;
      while (!Thread.currentThread().isInterrupted()) {
        try {
          event=eventQueue.take();
        }
 catch (        InterruptedException e) {
          LOG.error("Returning, interrupted : " + e);
          return;
        }
        int poolSize=launcherPool.getCorePoolSize();
        if (poolSize != limitOnPoolSize) {
          int numNodes=allNodes.size();
          int idealPoolSize=Math.min(limitOnPoolSize,numNodes);
          if (poolSize < idealPoolSize) {
            int newPoolSize=Math.min(limitOnPoolSize,idealPoolSize + INITIAL_POOL_SIZE);
            LOG.info("Setting ContainerLauncher pool size to " + newPoolSize + " as number-of-nodes to talk to is "+ numNodes);
            launcherPool.setCorePoolSize(newPoolSize);
          }
        }
        launcherPool.execute(createEventProcessor(event));
      }
    }
  }
;
  eventHandlingThread.setName("ContainerLauncher Event Handler");
  eventHandlingThread.start();
  super.start();
}
 

public ConnectivityAwareExecutor(Context ctx,int slowMobileThreads,int fastMobileThreads,int wifiThreads){
  super(1,1,0,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(),new ExecutorThreadFactory());
  this.ctx=ctx.getApplicationContext();
  this.slowMobileThreads=slowMobileThreads;
  this.fastMobileThreads=fastMobileThreads;
  this.wifiThreads=wifiThreads;
  connectivityManager=(ConnectivityManager)ctx.getSystemService(CONNECTIVITY_SERVICE);
  ctx.registerReceiver(connectivityReceiver,new IntentFilter(CONNECTIVITY_ACTION));
}
 

public DebuggerMessage take(int sessionId) throws InterruptedException {
  LinkedBlockingQueue<DebuggerMessage> consumer=null;
synchronized (queueLock) {
    consumer=consumers.get(sessionId);
    LinkedBlockingQueue<DebuggerMessage> newConsumer=new LinkedBlockingQueue<DebuggerMessage>(1024);
    if (consumer != null) {
      consumer.drainTo(newConsumer);
      consumer.offer(poison());
    }
    consumer=newConsumer;
    consumers.put(sessionId,consumer);
    takeTimestamps.put(sessionId,System.currentTimeMillis());
  }
  DebuggerMessage result=consumer.take();
synchronized (queueLock) {
  }
  if (result.type == PING) {
    pendingPings.remove(sessionId);
  }
  if (result != null && result.type == POISON) {
    throw new InterruptedException();
  }
  if (CoreMoSyncPlugin.getDefault().isDebugging()) {
    CoreMoSyncPlugin.trace("TAKE: Session id {0}: {1}",sessionId,result);
  }
  return result;
}
 

private ThreadPoolManager(){
  _effectsScheduledThreadPool=new ScheduledThreadPoolExecutor(Config.THREAD_P_EFFECTS,new PriorityThreadFactory("EffectsSTPool",Thread.MIN_PRIORITY));
  _generalScheduledThreadPool=new ScheduledThreadPoolExecutor(Config.THREAD_P_GENERAL,new PriorityThreadFactory("GerenalSTPool",Thread.NORM_PRIORITY));
  _ioPacketsThreadPool=new ThreadPoolExecutor(2,Integer.MAX_VALUE,5L,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),new PriorityThreadFactory("I/O Packet Pool",Thread.NORM_PRIORITY + 1));
  _generalPacketsThreadPool=new ThreadPoolExecutor(4,6,15L,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),new PriorityThreadFactory("Normal Packet Pool",Thread.NORM_PRIORITY + 1));
  _generalThreadPool=new ThreadPoolExecutor(2,4,5L,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),new PriorityThreadFactory("General Pool",Thread.NORM_PRIORITY));
  _aiThreadPool=new ThreadPoolExecutor(1,Config.AI_MAX_THREAD,10L,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
  _aiScheduledThreadPool=new ScheduledThreadPoolExecutor(Config.AI_MAX_THREAD,new PriorityThreadFactory("AISTPool",Thread.NORM_PRIORITY));
}
 

public LinkedBlockingQueue<WalrusDataMessage> getQueue(String key1,String key2){
  ConcurrentHashMap<String,LinkedBlockingQueue<WalrusDataMessage>> queues=queueMap.putIfAbsent(key1,new ConcurrentHashMap<String,LinkedBlockingQueue<WalrusDataMessage>>());
  if (queues == null) {
    queues=queueMap.get(key1);
  }
  LinkedBlockingQueue<WalrusDataMessage> queue=queues.putIfAbsent(key2,new LinkedBlockingQueue<WalrusDataMessage>(DATA_QUEUE_SIZE));
  if (queue == null) {
    queue=queues.get(key2);
  }
  return queue;
}
 

/** 
 * Initialize internal data structures
 */
public void init(Map<String,String> configParams){
  this.messageListeners=new ConcurrentHashMap<OFType,ListenerDispatcher<OFType,IOFMessageListener>>();
  this.switchListeners=new CopyOnWriteArraySet<IOFSwitchListener>();
  this.haListeners=new CopyOnWriteArraySet<IHAListener>();
  this.activeSwitches=new ConcurrentHashMap<Long,IOFSwitch>();
  this.connectedSwitches=new HashSet<OFSwitchImpl>();
  this.controllerNodeIPsCache=new HashMap<String,String>();
  this.updates=new LinkedBlockingQueue<IUpdate>();
  this.factory=new BasicFactory();
  this.providerMap=new HashMap<String,List<IInfoProvider>>();
  setConfigParams(configParams);
  this.role=getInitialRole(configParams);
  this.roleChanger=new RoleChanger();
  initVendorMessages();
  this.systemStartTime=System.currentTimeMillis();
}
 

/** 
 * This is private and not thread safe.
 */
private LinkedBlockingQueue<String> getQueue(State state){
  Preconditions.checkNotNull(state,"Attempted to get queue for invalid null state");
switch (state) {
case WRITING:
    return writingQ;
case LOGGED:
  return loggedQ;
case SENDING:
return sendingQ;
case IMPORT:
default :
return null;
}
}
 

/** 
 * This is private and not thread safe.
 */
private LinkedBlockingQueue<String> getQueue(State state){
  Preconditions.checkNotNull(state,"Attempted to get queue for invalid null state");
switch (state) {
case WRITING:
    return writingQ;
case LOGGED:
  return loggedQ;
case SENDING:
return sendingQ;
case IMPORT:
default :
return null;
}
}
 

public MultiLane(int width){
  if (width <= 0)   throw new IllegalArgumentException("width must be positive but is " + width);
  if ((width & (width - 1)) != 0)   throw new IllegalArgumentException("width must be a power of 2 but is " + width);
  lanes=(LinkedBlockingQueue<T>[])new LinkedBlockingQueue[width];
  for (int i=0; i < width; i++)   lanes[i]=new LinkedBlockingQueue<T>();
}
 

public void init(){
  workQueue=new LinkedBlockingQueue<EventWrapper>(maxQueueSize);
  for (  PrototypeWrapper pw : prototypeWrappers) {
    adviceLists.add(pw.advise());
  }
  Thread t=new Thread(this,"PEContainer");
  t.start();
  t=new Thread(new Watcher());
  t.start();
}
 

/** 
 * Creates a GerritHandler with the specified values.
 * @param gerritHostName        the hostName for gerrit.
 * @param gerritSshPort         the ssh port that the gerrit server listens to.
 * @param authentication        the authentication credentials.
 * @param numberOfWorkerThreads the number of eventthreads.
 */
public GerritHandler(String gerritHostName,int gerritSshPort,Authentication authentication,int numberOfWorkerThreads){
  super("Gerrit Events Reader");
  this.gerritHostName=gerritHostName;
  this.gerritSshPort=gerritSshPort;
  this.authentication=authentication;
  this.numberOfWorkerThreads=numberOfWorkerThreads;
  workQueue=new LinkedBlockingQueue<Work>();
  workers=new ArrayList<EventThread>(numberOfWorkerThreads);
  for (int i=0; i < numberOfWorkerThreads; i++) {
    workers.add(new EventThread(this,"Gerrit Worker EventThread_" + i));
  }
}
 

public Dispatcher(int queryTimeoutMs,int bulkQueryTimeoutMs,int queryMaxNumTries){
  getTasks=new LinkedBlockingQueue<GetTask>();
  this.queryTimeoutNano=queryTimeoutMs * 1000000;
  this.bulkQueryTimeoutNano=bulkQueryTimeoutMs * 1000000;
  this.queryMaxNumTries=queryMaxNumTries;
}
 

public void connect(BluetoothDevice device) throws IOException {
  if (D)   Log.d(TAG,"" + System.currentTimeMillis() + " connect");
  BluetoothAdapter.getDefaultAdapter().cancelDiscovery();
  try {
    Method m=device.getClass().getMethod("createRfcommSocket",new Class[]{int.class});
    mSock=(BluetoothSocket)m.invoke(device,1);
  }
 catch (  Exception e) {
    Log.e(TAG,"create bluetooth socket: " + e);
    throw new IOException("createRfcommSocket() failed");
  }
  try {
    mSock.connect();
    mIn=new BufferedReader(new InputStreamReader(mSock.getInputStream()),128);
    mOut=mSock.getOutputStream();
    queue=new LinkedBlockingQueue<String>();
    start();
  }
 catch (  IOException e) {
    Log.e(TAG,"connect() failed",e);
    mSock=null;
    throw e;
  }
}
 

/** 
 * 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));
}
 

protected BlockingQueue<RPCBuffer> getOutputQueue(String name){
synchronized (mOutputQueueMap) {
    if (!mOutputQueueMap.containsKey(name)) {
      mOutputQueueMap.put(name,new LinkedBlockingQueue<RPCBuffer>(1000));
    }
    return mOutputQueueMap.get(name);
  }
}
 

public void testNoninterruptible() throws InterruptedException {
  BlockingQueue<Object> q=new LinkedBlockingQueue<Object>();
  Thread t=tryMatchInThread(INPUT,BACKTRACKER,q);
  Thread.sleep(1000);
  t.interrupt();
  Object result=q.take();
  assertTrue("mismatch uncompleted",Boolean.FALSE.equals(result));
}
 

/** 
 * Initializes threading parameters for the HTTPClient's  {@link ThreadSafeClientConnManager}, and this class' ThreadPoolExecutor. 
 */
protected void initThreadPools(final int poolSize,final ExecutorService threadPool){
  if (poolSize < 1)   throw new IllegalArgumentException("poolSize may not be < 1");
  HttpParams params=client != null ? client.getParams() : new BasicHttpParams();
  ConnManagerParams.setMaxTotalConnections(params,poolSize);
  ConnManagerParams.setMaxConnectionsPerRoute(params,new ConnPerRouteBean(poolSize));
  HttpProtocolParams.setVersion(params,HttpVersion.HTTP_1_1);
  SchemeRegistry schemeRegistry=new SchemeRegistry();
  schemeRegistry.register(new Scheme("http",PlainSocketFactory.getSocketFactory(),80));
  schemeRegistry.register(new Scheme("https",SSLSocketFactory.getSocketFactory(),443));
  ClientConnectionManager cm=new ThreadSafeClientConnManager(params,schemeRegistry);
  super.client=new DefaultHttpClient(cm,params);
  this.threadPool=threadPool != null ? threadPool : new ThreadPoolExecutor(poolSize,poolSize,120,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
}
 

public synchronized ThreadPoolExecutor startExecutor(){
  if (_executor == null) {
    LinkedBlockingQueue<Runnable> queue=new LinkedBlockingQueue<Runnable>(_maxExecQueueCapacity);
    _executor=new ThreadPoolExecutor(_maxThreads,_maxThreads,30,TimeUnit.SECONDS,queue);
    _executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  }
  return _executor;
}
 

private <T>Map<String,List<MessageStream<T>>> consume(Map<String,Integer> topicCountMap,Decoder<T> decoder){
  if (topicCountMap == null) {
    throw new IllegalArgumentException("topicCountMap is null");
  }
  ZkGroupDirs dirs=new ZkGroupDirs(config.getGroupId());
  Map<String,List<MessageStream<T>>> ret=new HashMap<String,List<MessageStream<T>>>();
  String consumerUuid=config.getConsumerId();
  if (consumerUuid == null) {
    consumerUuid=generateConsumerId();
  }
  logger.info(format("create message stream by consumerid [%s] with groupid [%s]",consumerUuid,config.getGroupId()));
  final String consumerIdString=config.getGroupId() + "_" + consumerUuid;
  final TopicCount topicCount=new TopicCount(consumerIdString,topicCountMap);
  for (  Map.Entry<String,Set<String>> e : topicCount.getConsumerThreadIdsPerTopic().entrySet()) {
    final String topic=e.getKey();
    final Set<String> threadIdSet=e.getValue();
    final List<MessageStream<T>> streamList=new ArrayList<MessageStream<T>>();
    for (    String threadId : threadIdSet) {
      LinkedBlockingQueue<FetchedDataChunk> stream=new LinkedBlockingQueue<FetchedDataChunk>(config.getMaxQueuedChunks());
      queues.put(new StringTuple(topic,threadId),stream);
      streamList.add(new MessageStream<T>(topic,stream,config.getConsumerTimeoutMs(),decoder));
    }
    ret.put(topic,streamList);
    logger.debug("adding topic " + topic + " and stream to map.");
  }
  ZKRebalancerListener<T> loadBalancerListener=new ZKRebalancerListener<T>(config.getGroupId(),consumerIdString,ret);
  this.rebalancerListeners.add(loadBalancerListener);
  loadBalancerListener.start();
  registerConsumerInZK(dirs,consumerIdString,topicCount);
  zkClient.subscribeStateChanges(new ZKSessionExpireListener<T>(dirs,consumerIdString,topicCount,loadBalancerListener));
  zkClient.subscribeChildChanges(dirs.consumerRegistryDir,loadBalancerListener);
  for (  String topic : ret.keySet()) {
    final String partitionPath=ZkUtils.BrokerTopicsPath + "/" + topic;
    zkClient.subscribeChildChanges(partitionPath,loadBalancerListener);
  }
  loadBalancerListener.syncedRebalance();
  return ret;
}
 

ContainerExecutor(final int corePoolSize,final int maximumPoolSize,final long keepAliveTime,final TimeUnit unit){
  super(corePoolSize,maximumPoolSize,keepAliveTime,unit,new LinkedBlockingQueue<Runnable>(),new ThreadFactory(){
    private final int id=executorSeq.getAndIncrement();
    private final AtomicInteger threadSeq=new AtomicInteger(1);
    public Thread newThread(    final Runnable r){
      Thread thread=new ServiceThread(r,ServiceContainerImpl.this);
      thread.setName(String.format("MSC service thread %d-%d",Integer.valueOf(id),Integer.valueOf(threadSeq.getAndIncrement())));
      thread.setUncaughtExceptionHandler(HANDLER);
      return thread;
    }
  }
,POLICY);
}
 

/** 
 * calcul du gradient en chaque beta 
 */
private double[] computeGrad(GeneralizedDoubleGaussL2 kernel){
  debug.print(3,"++++++ g : ");
  final double grad[]=new double[dim];
  int nbcpu=Runtime.getRuntime().availableProcessors();
  ThreadPoolExecutor threadPool=new ThreadPoolExecutor(nbcpu,nbcpu,10,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
  Queue<Future<?>> futures=new LinkedList<Future<?>>();
class GradRunnable implements Runnable {
    GeneralizedDoubleGaussL2 kernel;
    int i;
    public GradRunnable(    GeneralizedDoubleGaussL2 kernel,    int i){
      this.kernel=kernel;
      this.i=i;
    }
    public void run(){
      double[][] matrix=kernel.distanceMatrixUnthreaded(listOfExamples,i);
      double sum=0;
      for (int x=0; x < matrix.length; x++)       if (lambda_matrix[x] != null)       for (int y=0; y < matrix.length; y++)       sum+=matrix[x][y] * lambda_matrix[x][y];
      grad[i]+=0.5 * sum;
    }
  }
  for (int i=0; i < grad.length; i++) {
    Runnable r=new GradRunnable(kernel,i);
    futures.add(threadPool.submit(r));
  }
  while (!futures.isEmpty()) {
    try {
      futures.remove().get();
    }
 catch (    Exception e) {
      System.err.println("error with grad :");
      e.printStackTrace();
    }
  }
  threadPool.shutdownNow();
  for (int i=0; i < grad.length; i++)   if (Math.abs(grad[i]) < num_cleaning)   grad[i]=0.0;
  debug.println(3,Arrays.toString(grad));
  return grad;
}
 

/** 
 * Starts the client thread and connects to the given server address. This method is asynchronous; use a ClientListener to check for connection events.
 * @param address the address to connect to
 * @param port the port to connect on
 */
public void connect(InetAddress address,int port){
  this.port=port;
  if (connected) {
    throw new IllegalStateException("Currently connected to a server.");
  }
  this.address=address;
  running=true;
  eventQueue=new LinkedBlockingQueue<Runnable>();
  clientThread=new Thread(new Runnable(){
    @Override public void run(){
      runThread();
    }
  }
,"Client Thread");
  clientThread.setDaemon(true);
  clientThread.start();
  eventThread=new Thread(new Runnable(){
    public void run(){
      while (true) {
        Runnable run=null;
        try {
          run=eventQueue.take();
          run.run();
        }
 catch (        InterruptedException ie) {
          break;
        }
      }
    }
  }
,"Client Event Thread");
  eventThread.setDaemon(true);
  eventThread.start();
}
 

/** 
 */
protected void initializeComponents(){
  super.initializeComponents();
  pendingQueue=new LinkedBlockingQueue<PendingRequest>();
  processor=new RequestProcessor();
  processerThread=new Thread(processor,"request-processor");
  processerThread.start();
}
 

/** 
 * Constructor to create a new, empty <code>AbsSession</code> object with a maximum number of allowed connections to a given iSCSI Target. This is the abstract definition for Session implementations
 * @param linkFactory The LinkFactory which called the Constructor
 * @param initConfiguration The configuration to use within this session.
 * @param initTargetName The name of the iSCSI Target.
 * @param inetAddress The <code>InetSocketAddress</code> of the leading connection of this session.
 * @param initExecutor The <code>ExecutorService</code> for the Connection Threads
 * @throws Exception if anything happens
 */
public Session(final LinkFactory linkFactory,final Configuration initConfiguration,final String initTargetName,final InetSocketAddress inetAddress,final ExecutorService initExecutor) throws Exception {
  maxConnections=Integer.parseInt(initConfiguration.getSessionSetting(initTargetName,OperationalTextKey.MAX_CONNECTIONS));
  factory=linkFactory;
  configuration=initConfiguration;
  commandSequenceNumber=new SerialArithmeticNumber();
  maximumCommandSequenceNumber=new SerialArithmeticNumber(1);
  nextFreeConnectionID=1;
  inetSocketAddress=inetAddress;
  initiatorTaskTag=new SerialArithmeticNumber(1);
  targetName=initTargetName;
  phase=new SecurityNegotiationPhase();
  capacityInformations=new TargetCapacityInformations();
  connections=new LinkedBlockingQueue<Connection>(maxConnections);
  executor=initExecutor;
  taskBalancer=new SimpleTaskBalancer(connections);
  outstandingTasks=new ConcurrentHashMap<ITask,Connection>();
  addNewConnection();
  maxConnections=Integer.parseInt(configuration.getSessionSetting(targetName,OperationalTextKey.MAX_CONNECTIONS));
  int targetMaxC=connections.peek().getSettingAsInt(OperationalTextKey.MAX_CONNECTIONS);
  if (targetMaxC < maxConnections) {
    maxConnections=targetMaxC;
  }
  addConnections(maxConnections - 1);
}
 

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;
}
 

private void init(WebSocketListener listener,Draft draft,SocketChannel sockchannel){
  this.sockchannel=sockchannel;
  this.bufferQueue=new LinkedBlockingQueue<ByteBuffer>(10);
  this.socketBuffer=ByteBuffer.allocate(65558);
  socketBuffer.flip();
  this.wsl=listener;
  this.role=Role.CLIENT;
  this.draft=draft;
}
 

@Override public RateLimitJobQueueBuilder withRateLimitQueue(){
  return new RateLimitJobQueueBuilder(){
    private BlockingQueue<Job> target;
    private long throwlingInMs=1000;
    private long poolingTimeout=500;
    @Override public RateLimitJobQueueBuilder withTargetQueue(    final BlockingQueue<Job> target){
      this.target=target;
      return this;
    }
    @Override public RateLimitJobQueueBuilder withThrowling(    final long throwlingInMs){
      this.throwlingInMs=throwlingInMs;
      return this;
    }
    @Override public RateLimitJobQueueBuilder withPoolingTimeout(    final long poolingTimeout){
      this.poolingTimeout=poolingTimeout;
      return this;
    }
    @Override public SchedulerBuilder doneRateLimitQueue(){
      if (target == null) {
        target=new LinkedBlockingQueue<Job>();
      }
      queue=new RateLimitBlockingQueueJobQueue<Job>(target,poolingTimeout,throwlingInMs);
      return DefaultSchedulerBuilder.this;
    }
  }
;
}
 

private static ExecutorService getBitmapExecutor(BitmapSource bitmapSource,BitmapSource.Type type){
  if (BitmapCache.getInstance().existOnDisk(bitmapSource.getAbsoluteFileName(type))) {
    if (sSingleThreadExecutor == null) {
      sSingleThreadExecutor=Executors.newSingleThreadExecutor(new BitmapLoaderThreadFactory("single thread"));
    }
    return sSingleThreadExecutor;
  }
 else {
    if (sBitmapExecutor == null) {
      sBitmapExecutor=new ThreadPoolExecutor(DEFAULT_CORE_POOL_SIZE,MAXIMUM_POOL_SIZE,KEEP_ALIVE,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(),new BitmapLoaderThreadFactory("multiple threads"));
    }
    return sBitmapExecutor;
  }
}
 

public AmqpPublisher(final AmqpRawPublisher accessor,final AmqpPublisherConfiguration configuration){
  super((accessor != null) ? accessor : new AmqpRawPublisher(configuration,null),configuration);
  this.router=configuration.router;
  this.rawBuffer=this.accessor.buffer;
  this.buffer=new LinkedBlockingQueue<ILoggingEvent>();
}
 

public Receiver(final XBee xbee){
  thisXBee=xbee;
  thisXBee.addPacketListener(this);
  queue=new LinkedBlockingQueue<RxResponse16>();
  messageListeners=new HashMap<Integer,MessageListener>();
}
 

public void testLBQPut() throws Exception {
  LinkedBlockingQueue<String> twoThreadBlockingQueue=new LinkedBlockingQueue<String>();
  String[] items=generate(num);
  for (  String item : items) {
    twoThreadBlockingQueue.put(item);
  }
  System.gc();
}
 

public AsyncNotifier(){
  this.SYNC=new Object();
  this.eventQueue=new LinkedBlockingQueue<Event>();
  setPriority(MIN_PRIORITY);
  setName("SMSLib-AsyncNotifier : " + getGateway().getGatewayId());
  setDaemon(true);
  start();
  getGateway().getService().getLogger().logDebug("AsyncNotifier thread started.",null,getGateway().getGatewayId());
}
 

/** 
 * Initializes a new thread pool with a given name, number of threads, and queue size.
 * @param name identifying name
 * @param numThreads the number of threads allowed in the pool
 * @param maxQueueSize the size of the pool entry queue
 */
public ThreadPool(String name,int numThreads,int maxQueueSize){
  _name=name;
  _maxThreads=numThreads;
  _queue=new LinkedBlockingQueue<T>(maxQueueSize);
  _myThreadGroup=new MyThreadGroup();
  _threads.add(new MyThread());
}
 

public Stage(String name,int capacity,int maxThreads,boolean blockOnAdd){
  this.name=name;
  this.capacity=capacity;
  this.blockOnAdd=blockOnAdd;
  this.maxThreads=maxThreads;
  queue=new LinkedBlockingQueue<V>(capacity);
  threads=new ArrayList<Thread>();
  for (int i=0; i < maxThreads; i++) {
    addThread();
  }
}
 

public MemoryNotifyDispatcher(Executor executor,int cap){
  _executor=executor;
  _cap=cap;
  _remainingThreshold=(int)(_cap * 0.25);
  _queue=new LinkedBlockingQueue<NotifyRequest>(_cap);
}
 

/** 
 * Initializes a new thread pool with a given name, number of threads, and queue size.
 * @param name identifying name
 * @param numThreads the number of threads allowed in the pool
 * @param maxQueueSize the size of the pool entry queue
 */
public ThreadPool(String name,int numThreads,int maxQueueSize){
  _name=name;
  _maxThreads=numThreads;
  _queue=new LinkedBlockingQueue<T>(maxQueueSize);
  _myThreadGroup=new MyThreadGroup();
  _threads.add(new MyThread());
}
 

/** 
 * Create a PeerGroup
 */
public PeerGroup(BlockStore blockStore,NetworkParameters params,BlockChain chain){
  this.blockStore=blockStore;
  this.params=params;
  this.chain=chain;
  inactives=new LinkedBlockingQueue<PeerAddress>();
  peers=Collections.synchronizedSet(new HashSet<Peer>());
  peerEventListeners=Collections.synchronizedSet(new HashSet<PeerEventListener>());
  pendingTransactionListeners=Collections.synchronizedList(new ArrayList<PendingTransactionListener>());
  peerPool=new ThreadPoolExecutor(CORE_THREADS,DEFAULT_CONNECTIONS,THREAD_KEEP_ALIVE_SECONDS,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(1),new PeerGroupThreadFactory());
}
 

/** 
 * Create a new message sender without starting it.
 * @param channel Channel to send messages to.
 */
MessageSender(IChannelMessageSerializer serializer,Messenger messenger,ILogger logger){
  setName("Probe Message Sending Thread");
  setDaemon(true);
  this.serializer=serializer;
  this.messenger=messenger;
  this.logger=logger;
  sendQueue=new LinkedBlockingQueue<ProbeMessage>();
}
 

@Test public void scheduleTaskAtFixedRate() throws Exception {
  final Queue<Long> timestamps=new LinkedBlockingQueue<Long>();
  ScheduledFuture<?> f=loop.scheduleAtFixedRate(new Runnable(){
    @Override public void run(){
      timestamps.add(System.nanoTime());
      try {
        Thread.sleep(50);
      }
 catch (      InterruptedException e) {
      }
    }
  }
,100,100,TimeUnit.MILLISECONDS);
  Thread.sleep(550);
  assertTrue(f.cancel(true));
  assertEquals(5,timestamps.size());
  Long previousTimestamp=null;
  for (  Long t : timestamps) {
    if (previousTimestamp == null) {
      previousTimestamp=t;
      continue;
    }
    assertTrue(t.longValue() - previousTimestamp.longValue() >= TimeUnit.MILLISECONDS.toNanos(90));
    previousTimestamp=t;
  }
}
 

public IndexRunner(String logname,String u,String i,String r,String uid,String pswd,String cpswd){
  foundWords=new TreeSet<String>();
  logger=new ApplicationLogger(logname);
  conn=new DatabaseConnection(logger,u,i,r,uid,pswd,cpswd,500);
  indexType=SCAN;
  guid=null;
  keepRunning=true;
  doc=new QDomDocument();
  workQueue=new LinkedBlockingQueue<String>(MAX_QUEUED_WAITING);
}
 

public AsyncEventExecutor(int poolSize,int maxPoolSize,int keepAliveTime,int queueSize){
  queue=new LinkedBlockingQueue<Runnable>(queueSize);
  mono_queue=new LinkedBlockingQueue<Runnable>(queueSize);
  NamedThreadFactory threadFactory=new NamedThreadFactory("Nuxeo Async Events");
  executor=new ThreadPoolExecutor(poolSize,maxPoolSize,keepAliveTime,TimeUnit.SECONDS,queue,threadFactory);
  mono_executor=new ThreadPoolExecutor(1,1,keepAliveTime,TimeUnit.SECONDS,mono_queue,threadFactory);
}
 

/** 
 * Called when the activity is first created. 
 */
@Override public void onCreate(Bundle savedInstanceState){
  super.onCreate(savedInstanceState);
  TextView tv=new TextView(this);
  tv.setText("how do i make anything work");
  setContentView(tv);
  tv.setText("i have file");
  try {
    int sample_rate=8000;
    int frame_size=160;
    int frame_rate=sample_rate / frame_size;
    SipdroidSocket socket=new SipdroidSocket(5004);
    SipdroidSocket recv_socket=new SipdroidSocket(6004);
    boolean do_sync=true;
    BlockingQueue<short[]> queue=new LinkedBlockingQueue<short[]>();
    SenderThread sender=new SenderThread(do_sync,frame_rate,frame_size,socket,"10.0.2.2",33333,queue);
    ReceiverThread receiver=new ReceiverThread(recv_socket);
    AudioPusher pusher=new AudioPusher("/test3.wav",queue);
    sender.start();
    pusher.start();
    receiver.start();
    boolean running=true;
    long time=System.currentTimeMillis();
    while (running) {
      Thread.sleep(frame_size);
    }
    sender.halt();
    receiver.halt();
  }
 catch (  Throwable t) {
    Log.e("AudioTrack","Playback Failed");
    tv.setText(t.getMessage());
  }
}
 

public ConnectionBase(final ConnectionInformation connectionInformation){
  super();
  this.connectionInformation=connectionInformation;
  this.lookupExecutor=new ThreadPoolExecutor(0,1,1,TimeUnit.MINUTES,new LinkedBlockingQueue<Runnable>(),new NamedThreadFactory("ConnectionBaseExecutor/" + connectionInformation.toMaskedString()));
  this.messenger=new Messenger(getMessageTimeout());
  this.pingService=new PingService(this.messenger);
  this.connector=createConnector();
}
 

public StorageImpl() throws IOException {
  this.taskQueue=new LinkedBlockingQueue<Runnable>();
  this.executorService=new ThreadPoolExecutor(1,1,0L,TimeUnit.MILLISECONDS,this.taskQueue,new NamedThreadFactory(StorageImpl.class.getName()));
  this.rootFolder=new File(System.getProperty("org.openscada.ds.storage.file.root",System.getProperty("user.home") + File.separator + ".openscadaDS"));
  if (!this.rootFolder.exists()) {
    this.rootFolder.mkdirs();
  }
  if (!this.rootFolder.exists() || !this.rootFolder.isDirectory()) {
    throw new IOException(String.format("Unable to use directory: %s",this.rootFolder));
  }
}
 

public BatchWriter(EmbeddedSolrServer solr,int batchSize,TaskID tid,int writerThreads,int queueSize){
  this.solr=solr;
  this.writerThreads=writerThreads;
  this.queueSize=queueSize;
  taskId=tid;
  batchPool=new ThreadPoolExecutor(writerThreads,writerThreads,5,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(queueSize),new ThreadPoolExecutor.CallerRunsPolicy());
  this.batchToWrite=new ArrayList<SolrInputDocument>(batchSize);
}
 

protected <R extends Mergeable<R>>Execution<R> createSharedMemoryExecution(Computation<R> computation){
  Collection<ComputationId> ids=new ArrayList<>();
  for (int i=0; i < MAX_THREADS; i++) {
    final int currentId=i;
    ids.add(new ComputationId(){
      @Override public int currentId(){
        return currentId;
      }
      @Override public int maxId(){
        return MAX_THREADS - 1;
      }
    }
);
  }
  BlockingQueue<Future<R>> futures=new LinkedBlockingQueue<>();
  return SharedMemoryExecution.of(ids,getManager().resolve(java.util.concurrent.Executor.class,Default.class,getManager().getRootContext()),computation,getManager().resolve(Enrichment.class,Default.class,getManager().getRootContext()),new ContextEvent<ComputationInstanceContext>(){
    @Override public void initialize(    ComputationInstanceContext context){
      context.setParent(getManager().getRootContext());
      getManager().initializeContext(context);
    }
    @Override public void finalize(    ComputationInstanceContext context){
      getManager().finalizeContext(context);
    }
  }
,getManager().getRootContext(),futures);
}
 

public void start(){
  localDebugger("|");
  localLogger("Entering: start()");
  receiveQueue=new LinkedBlockingQueue<PCEPMessage>();
  sendingQueue=new LinkedBlockingQueue<PCEPMessage>();
  sendingThreadIsActive=false;
  initSendingThread();
}
 

public AsyncStatsContainer(StatsContainer container){
  this.container=container;
  this.queue=new LinkedBlockingQueue<StatsCommand>();
  this.executor=Executors.newSingleThreadExecutor();
  executor.execute(new AsyncStatsListener());
}
 

@Override public void connect() throws ImException {
  if (mConnected) {
    throw new ImException("Already connected");
  }
  mStopped=false;
  mStopRetry=false;
  mSendQueue=new LinkedBlockingQueue<Primitive>();
  mReceiveQueue=new LinkedBlockingQueue<Primitive>();
  mSendThread=new Thread(this,"HttpDataChannel");
  mSendThread.setDaemon(true);
  mSendThread.start();
  mConnected=true;
}
 

BackgroundLoaderManager(Context context){
  mPendingTaskUris=new HashSet<Uri>();
  mCallbacks=new HashMap<Uri,Set<ItemLoadedCallback>>();
  final LinkedBlockingQueue<Runnable> queue=new LinkedBlockingQueue<Runnable>();
  final int poolSize=MAX_THREADS;
  mExecutor=new ThreadPoolExecutor(poolSize,poolSize,5,TimeUnit.SECONDS,queue,new BackgroundLoaderThreadFactory(getTag()));
  mCallbackHandler=new Handler();
}
 

/** 
 * Constructor
 * @param surfaceHolder The surface holder
 */
public PreviewThread(SurfaceHolder surfaceHolder){
  mMainHandler=new Handler(Looper.getMainLooper());
  mQueue=new LinkedBlockingQueue<Runnable>();
  mSurfaceHolder=surfaceHolder;
  mPreviewState=PREVIEW_STATE_STOPPED;
  mOverlayDataQueue=new LinkedBlockingQueue<VideoEditor.OverlayData>();
  for (int i=0; i < OVERLAY_DATA_COUNT; i++) {
    mOverlayDataQueue.add(new VideoEditor.OverlayData());
  }
  start();
}
 

/** 
 * @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);
}
 

@Test public void receiveFile() throws Exception {
  final String payload="dummy payload";
  _testKit.removeService("GreetingService");
  final MockHandler greetingService=_testKit.registerInOnlyService("GreetingService");
  createFile(payload,FILE_NAME);
  Thread.sleep(3000);
  final LinkedBlockingQueue<Exchange> recievedMessages=greetingService.getMessages();
  assertThat(recievedMessages,is(notNullValue()));
  final Exchange recievedExchange=recievedMessages.iterator().next();
  assertThat(recievedExchange.getMessage().getContent(String.class),is(equalTo(payload)));
}
 

public static ThreadPoolExecutor coreThreadsTimoutExecutor(final int numberOfThreads,final int threadPriority,final String threadNamePrefix){
  final ThreadFactory factory=new ThreadFactoryBuilder().setPriority(threadPriority).setNameFormat(threadNamePrefix + "%d").setDaemon(true).build();
  final ThreadPoolExecutor pool=new ThreadPoolExecutor(numberOfThreads,numberOfThreads,100L,MILLISECONDS,new LinkedBlockingQueue<Runnable>(),factory);
  pool.allowCoreThreadTimeOut(true);
  return pool;
}
 

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);
}
}
 

public void init(){
  workQueue=new LinkedBlockingQueue<EventWrapper>(maxQueueSize);
  for (  PrototypeWrapper pw : prototypeWrappers) {
    adviceLists.add(pw.advise());
  }
  Thread t=new Thread(this,"PEContainer");
  t.start();
  t=new Thread(new Watcher());
  t.start();
}
 

@Override public StreamDataProvider<JSONObject> buildDataProvider(DataSourceFilter<String> dataFilter,String oldSinceKey,ShardingStrategy shardingStrategy,Set<Integer> partitions) throws Exception {
  String path=config.get("file.path");
  if (path == null) {
    path="data/cars.json";
  }
  long offset=oldSinceKey == null ? 0L : Long.parseLong(oldSinceKey);
  PerfFileDataProvider provider=new PerfFileDataProvider(_versionComparator,new File(path),0L,new LinkedBlockingQueue<JSONObject>(30000));
  if (dataFilter != null) {
    provider.setFilter(dataFilter);
  }
  return provider;
}
 

/** 
 * 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);
  }
}
 

Collection<Queue<Boolean>> queues(){
  List<Queue<Boolean>> queues=new ArrayList<Queue<Boolean>>();
  queues.add(new ConcurrentLinkedQueue<Boolean>());
  queues.add(new ArrayBlockingQueue<Boolean>(count,false));
  queues.add(new ArrayBlockingQueue<Boolean>(count,true));
  queues.add(new LinkedBlockingQueue<Boolean>());
  queues.add(new LinkedBlockingDeque<Boolean>());
  queues.add(new PriorityBlockingQueue<Boolean>());
  queues.add(new PriorityQueue<Boolean>());
  queues.add(new LinkedList<Boolean>());
  Collections.shuffle(queues);
  return queues;
}
 

/** 
 * Default constructor called by sub-classes
 */
public Protocol(){
  currentCommand=new Command();
  waitingForAck=null;
  tempAckProcessor=null;
  pendingInstructions=new LinkedBlockingQueue<ProtocolInstruction>();
}
 

protected void initExecutor(Map<String,Object> config){
  if (executor == null) {
    int threads=ConfigUtils.getInteger(config,MAX_POOL_SIZE,CFG_PREFIX + ".threads");
    if (threads <= 1) {
      executor=new Executor(){
        public void execute(        Runnable command){
          command.run();
        }
      }
;
    }
 else {
      ThreadFactory threadFactory=new RepositoryConnectorThreadFactory(getClass().getSimpleName());
      executor=new ThreadPoolExecutor(threads,threads,3,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory);
    }
  }
}
 

/** 
 * Create a robot stream writer for a specific Bluetooth connection
 * @param btc The Bluetooth connection to send to
 */
protected RobotSendingQueue(BluetoothConnection btc){
  this.btc=btc;
  this.sendingQueue=new LinkedBlockingQueue<Pair<CommandMessage,Boolean>>();
  this.w=new Robot.RobotSendingQueue.Writer();
  this.startWriter();
}
 

protected Message doSendAndReceiveWithFixed(final String exchange,final String routingKey,final Message message){
  Message replyMessage=this.execute(new ChannelCallback<Message>(){
    public Message doInRabbit(    Channel channel) throws Exception {
      final LinkedBlockingQueue<Message> replyHandoff=new LinkedBlockingQueue<Message>();
      String messageTag=UUID.randomUUID().toString();
      RabbitTemplate.this.replyHolder.put(messageTag,replyHandoff);
      String replyTo=message.getMessageProperties().getReplyTo();
      if (StringUtils.hasLength(replyTo) && logger.isDebugEnabled()) {
        logger.debug("Dropping replyTo header:" + replyTo + " in favor of template's configured reply-queue:"+ RabbitTemplate.this.replyQueue.getName());
      }
      String springReplyTo=(String)message.getMessageProperties().getHeaders().get(STACKED_REPLY_TO_HEADER);
      message.getMessageProperties().setHeader(STACKED_REPLY_TO_HEADER,pushHeaderValue(replyTo,springReplyTo));
      message.getMessageProperties().setReplyTo(RabbitTemplate.this.replyQueue.getName());
      String correlation=(String)message.getMessageProperties().getHeaders().get(STACKED_CORRELATION_HEADER);
      if (StringUtils.hasLength(correlation)) {
        message.getMessageProperties().setHeader(STACKED_CORRELATION_HEADER,pushHeaderValue(messageTag,correlation));
      }
 else {
        message.getMessageProperties().setHeader("spring_reply_correlation",messageTag);
      }
      if (logger.isDebugEnabled()) {
        logger.debug("Sending message with tag " + messageTag);
      }
      doSend(channel,exchange,routingKey,message,null);
      Message reply=(replyTimeout < 0) ? replyHandoff.take() : replyHandoff.poll(replyTimeout,TimeUnit.MILLISECONDS);
      RabbitTemplate.this.replyHolder.remove(messageTag);
      return reply;
    }
  }
);
  return replyMessage;
}
 

@Test public void testThreadPoolExecutor() throws InterruptedException {
  Executor executor=new ThreadPoolExecutor(5,5,5L,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(5));
  SignallingRunnable runner=new SignallingRunnable("testThreadPoolExecutor");
  executor.execute(runner);
{
    Operation op=assertLastExecutionOperation(runner);
    List<Operation> opsList=TEST_COLLECTOR.getCollectedOperations();
    assertEquals("Mismatched number of operations generated",2,opsList.size());
    SourceCodeLocation scl=op.getSourceCodeLocation();
    assertEquals("Mismatched class name",SignallingRunnable.class.getName(),scl.getClassName());
    assertEquals("Mismatched method name","run",scl.getMethodName());
  }
{
    Operation op=assertCurrentThreadExecution();
    SourceCodeLocation scl=op.getSourceCodeLocation();
    assertEquals("Mismatched class name",getClass().getName(),scl.getClassName());
    assertEquals("Mismatched method name","execute",scl.getMethodName());
  }
}
 

/** 
 * Create the BlockingQueue to use for the ThreadPoolExecutor. <p>A LinkedBlockingQueue instance will be created for a positive capacity value; a SynchronousQueue else.
 * @param queueCapacity the specified queue capacity
 * @return the BlockingQueue instance
 * @see java.util.concurrent.LinkedBlockingQueue
 * @see java.util.concurrent.SynchronousQueue
 */
protected BlockingQueue<Runnable> createQueue(int queueCapacity){
  if (queueCapacity > 0) {
    return new LinkedBlockingQueue<Runnable>(queueCapacity);
  }
 else {
    return new SynchronousQueue<Runnable>();
  }
}
 

private void setQueryString(String query){
synchronized (mLoaderThread) {
    mQueryVersion++;
    mLoadedPages=new CopyOnWriteArraySet<Integer>();
    mPageLoadQueue=new LinkedBlockingQueue<PageLoadCommand>();
  }
  mQueryString=query;
  if (query != null && query.length() > 0)   mSearchQueryPattern=Pattern.compile(query,Pattern.LITERAL | Pattern.CASE_INSENSITIVE);
 else   mSearchQueryPattern=null;
  mPageLoadQueue.add(new PageLoadCommand(mQueryVersion,0));
  mLoaderThread.interrupt();
}
 

/** 
 * 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;
}
 

/** 
 * Records and then clears all pending counts if we have crossed a window boundary or have a bunch of data accumulated or if forced.
 * @param force  If true, then windows and such are ignored and the data is pushed out regardless
 */
private void recordCounts(boolean force){
  long currentRecordWindowStart=(now() / reportingInterval) * reportingInterval;
  if (lastRecordOutput == 0) {
    lastRecordOutput=currentRecordWindowStart;
  }
  final int bufferedTuples=tupleLog.get().size();
  if (force || currentRecordWindowStart > lastRecordOutput || bufferedTuples > maxBufferedTuples) {
    if (force) {
      logger.info("Forced recording");
    }
 else     if (bufferedTuples > maxBufferedTuples) {
      logger.info("Recording due to max tuples");
    }
 else {
      logger.info("Recording due to time");
    }
    Queue<Tuple> oldLog=tupleLog.getAndSet(new LinkedBlockingQueue<Tuple>());
    Multiset<String> counts=HashMultiset.create();
    for (    Tuple tuple : oldLog) {
      counts.add(tuple.getString(0) + "\t" + tuple.getString(1));
    }
    for (    String keyValue : counts.elementSet()) {
      final int n=counts.count(keyValue);
      outputCollector.emit(oldLog,new Values(keyValue,n));
      count.addAndGet(n);
    }
    logger.info(String.format("Logged %d events",count.get()));
    for (    Tuple tuple : oldLog) {
      outputCollector.ack(tuple);
    }
    lastRecordOutput=currentRecordWindowStart;
  }
}
 

public void testConcurrency() throws Exception {
  Persister persister=new Persister();
  CountDownLatch latch=new CountDownLatch(20);
  BlockingQueue<Status> status=new LinkedBlockingQueue<Status>();
  Example example=new Example();
  example.name="Eample Name";
  example.value="Some Value";
  example.number=10;
  example.date=new Date();
  example.locale=Locale.UK;
  for (int i=0; i < 20; i++) {
    Worker worker=new Worker(latch,persister,status,example);
    worker.start();
  }
  for (int i=0; i < 20; i++) {
    assertEquals("Serialization fails when used concurrently",status.take(),Status.SUCCESS);
  }
}
 

public ImageLoader(Context context){
  queue=new LinkedBlockingQueue<Task>(500);
  imageSizeDefault=context.getResources().getDrawable(R.drawable.unknown_album).getIntrinsicHeight();
  DisplayMetrics metrics=context.getResources().getDisplayMetrics();
  imageSizeLarge=(int)Math.round(Math.min(metrics.widthPixels,metrics.heightPixels) * 0.6);
  for (int i=0; i < CONCURRENCY; i++) {
    new Thread(this,"ImageLoader").start();
  }
  createLargeUnknownImage(context);
}
 

public ImageLoader(Context context){
  queue=new LinkedBlockingQueue<Task>(500);
  imageSizeDefault=context.getResources().getDrawable(R.drawable.unknown_album).getIntrinsicHeight();
  DisplayMetrics metrics=context.getResources().getDisplayMetrics();
  imageSizeLarge=(int)Math.round(Math.min(metrics.widthPixels,metrics.heightPixels) * 0.6);
  for (int i=0; i < CONCURRENCY; i++) {
    new Thread(this,"ImageLoader").start();
  }
  createLargeUnknownImage(context);
}
 

public ImageLoader(Context context){
  queue=new LinkedBlockingQueue<Task>(500);
  imageSizeDefault=context.getResources().getDrawable(R.drawable.unknown_album).getIntrinsicHeight();
  DisplayMetrics metrics=context.getResources().getDisplayMetrics();
  imageSizeLarge=(int)Math.round(Math.min(metrics.widthPixels,metrics.heightPixels) * 0.6);
  for (int i=0; i < CONCURRENCY; i++) {
    new Thread(this,"ImageLoader").start();
  }
  createLargeUnknownImage(context);
}
 

public ImageLoader(Context context){
  queue=new LinkedBlockingQueue<Task>(500);
  imageSizeDefault=context.getResources().getDrawable(R.drawable.unknown_album).getIntrinsicHeight();
  DisplayMetrics metrics=context.getResources().getDisplayMetrics();
  imageSizeLarge=(int)Math.round(Math.min(metrics.widthPixels,metrics.heightPixels) * 0.6);
  for (int i=0; i < CONCURRENCY; i++) {
    new Thread(this,"ImageLoader").start();
  }
  createLargeUnknownImage(context);
}
 

public ImageLoader(Context context){
  queue=new LinkedBlockingQueue<Task>(500);
  imageSizeDefault=context.getResources().getDrawable(R.drawable.unknown_album).getIntrinsicHeight();
  DisplayMetrics metrics=context.getResources().getDisplayMetrics();
  imageSizeLarge=(int)Math.round(Math.min(metrics.widthPixels,metrics.heightPixels) * 0.6);
  for (int i=0; i < CONCURRENCY; i++) {
    new Thread(this,"ImageLoader").start();
  }
  createLargeUnknownImage(context);
}
 

public Indexer(){
  if (logger.isLoggable(Level.INFO)) {
    logger.info("Creating indexer ...");
  }
  this.db=DatabaseHelper.getInstance();
  this.queue=new LinkedBlockingQueue<IndexRequest>();
  this.worker=null;
}
 

public UnorderedThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,ThreadFactory threadFactory,IoEventQueueHandler queueHandler){
  super(0,1,keepAliveTime,unit,new LinkedBlockingQueue<Runnable>(),threadFactory,new AbortPolicy());
  if (corePoolSize < 0) {
    throw new IllegalArgumentException("corePoolSize: " + corePoolSize);
  }
  if (maximumPoolSize == 0 || maximumPoolSize < corePoolSize) {
    throw new IllegalArgumentException("maximumPoolSize: " + maximumPoolSize);
  }
  if (queueHandler == null) {
    queueHandler=IoEventQueueHandler.NOOP;
  }
  this.corePoolSize=corePoolSize;
  this.maximumPoolSize=maximumPoolSize;
  this.queueHandler=queueHandler;
}
 

@Test public void main() throws InterruptedException {
  BlockingQueue<Runnable> queue=new LinkedBlockingQueue<Runnable>();
  ThreadPoolExecutor executor=new ThreadPoolExecutor(SIZE,SIZE,1000,TimeUnit.MILLISECONDS,queue);
  long time=System.currentTimeMillis();
  for (int i=0; i < SIZE; i++) {
    executor.submit(new Runnable(){
      final Tedis tedis=new Tedis(ip1);
      public void run(){
        for (int j=0; j <= 2000; j++) {
          tedis.set("foo".getBytes(),"bar".getBytes());
          if (j % 1000 == 0) {
            System.out.println("finished:" + j);
          }
        }
      }
    }
);
  }
  executor.shutdown();
  while (!executor.isTerminated()) {
  }
  System.out.println("qps:" + (SIZE * 2000) / ((System.currentTimeMillis() - time) / 1000));
}
 

protected Queue<StrategyParams> getTasks(final StrategyParams params){
  for (  StrategyParam param : params.getAll()) {
    param.setValue(param.getMin());
  }
  Queue<StrategyParams> tasks=new LinkedBlockingQueue<StrategyParams>();
  boolean allTasksAssigned=false;
  while (!allTasksAssigned && !cancelled) {
    StrategyParams strategyParamsCopy=new StrategyParams(params);
    tasks.add(strategyParamsCopy);
    StrategyParam lastParam=params.get(params.size() - 1);
    lastParam.setValue(lastParam.getValue() + lastParam.getStep());
    for (int paramNumber=params.size() - 1; paramNumber >= 0; paramNumber--) {
      StrategyParam param=params.get(paramNumber);
      if (param.getValue() > param.getMax()) {
        param.setValue(param.getMin());
        if (paramNumber == 0) {
          allTasksAssigned=true;
          break;
        }
 else {
          int prevParamNumber=paramNumber - 1;
          StrategyParam prevParam=params.get(prevParamNumber);
          prevParam.setValue(prevParam.getValue() + prevParam.getStep());
        }
      }
    }
  }
  return tasks;
}
 

/** 
 * Start accepting new connections in a background thread.
 */
public void start(){
  if (!this.socket.isBound()) {
    throw new IllegalStateException("Can't start ConnectionHandler " + "without a bound socket!");
  }
  this.stop=false;
  if (this.executor == null || this.executor.isShutdown()) {
    this.executor=new ThreadPoolExecutor(OUTBOUND_CONNECTIONS_POOL_SIZE,OUTBOUND_CONNECTIONS_POOL_SIZE,OUTBOUND_CONNECTIONS_THREAD_KEEP_ALIVE_SECS,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),new ConnectorThreadFactory());
  }
  if (this.thread == null || !this.thread.isAlive()) {
    this.thread=new Thread(this);
    this.thread.setName("bt-serve");
    this.thread.start();
  }
}
 

/** 
 * Initialize internal data structures
 */
public void init(Map<String,String> configParams){
  this.messageListeners=new ConcurrentHashMap<OFType,ListenerDispatcher<OFType,IOFMessageListener>>();
  this.switchListeners=new CopyOnWriteArraySet<IOFSwitchListener>();
  this.haListeners=new CopyOnWriteArraySet<IHAListener>();
  this.activeSwitches=new ConcurrentHashMap<Long,IOFSwitch>();
  this.connectedSwitches=new HashSet<OFSwitchImpl>();
  this.controllerNodeIPsCache=new HashMap<String,String>();
  this.updates=new LinkedBlockingQueue<IUpdate>();
  this.factory=new BasicFactory();
  this.providerMap=new HashMap<String,List<IInfoProvider>>();
  setConfigParams(configParams);
  this.role=getInitialRole(configParams);
  this.roleChanger=new RoleChanger();
  initVendorMessages();
  this.systemStartTime=System.currentTimeMillis();
}
 

/** 
 * 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);
  }
}
 

/** 
 * Send a request and wait for the response.
 * @param request
 * @return
 * @throws InterruptedException
 */
public Packet sendSync(Packet request) throws InterruptedException {
  LinkedBlockingQueue<Packet> queue=new LinkedBlockingQueue<Packet>();
  PacketListener listener=new SyncPacketListener(queue);
  PacketFilter filter=new IdPacketFilter(request.getPacketID());
  addPacketListener(listener,filter);
  sendPacket(request);
  Packet response=queue.poll(10000,TimeUnit.MILLISECONDS);
  removePacketListener(listener);
  return response;
}
 

public BlockingQueue<String> createLineQueue(){
  LinkedBlockingQueue<String> queue=new LinkedBlockingQueue<String>();
synchronized (textWaiters) {
    textWaiters.add(queue);
  }
  return queue;
}
 

/** 
 * Gets a value, blocks until a value is available.
 * @param thiz This.
 * @return The value.
 */
@SuppressWarnings("unchecked") @WeelMethod public final static Value take(final ValueMap thiz){
  try {
    return ((LinkedBlockingQueue<Value>)thiz.get("#INSTANCE#").getObject()).take();
  }
 catch (  InterruptedException e) {
    return new Value();
  }
}