Wait for the successful channel re-subscription
@param oldChannelIdSymbolMap
@return
@throws BitfinexClientException
@throws InterruptedException
548606
private void waitForChannelResubscription(final Map<Integer, ChannelCallbackHandler> oldChannelIdSymbolMap)
			throws BitfinexClientException, InterruptedException {
		
		final Stopwatch stopwatch = Stopwatch.createStarted();
		final long MAX_WAIT_TIME_IN_MS = TimeUnit.MINUTES.toMillis(3);
		logger.info("Waiting for streams to resubscribe (max wait time {} msec)", MAX_WAIT_TIME_IN_MS);

		synchronized (channelIdToHandlerMap) {
			
			while(channelIdToHandlerMap.size() != oldChannelIdSymbolMap.size()) {
				
				if(stopwatch.elapsed(TimeUnit.MILLISECONDS) > MAX_WAIT_TIME_IN_MS) {
					// Raise BitfinexClientException
					handleResubscribeFailed(oldChannelIdSymbolMap);
				}
				
				channelIdToHandlerMap.wait(500);
			}
			
		}
	}
658215
public final void ensureOrder(TotalOrderRemoteTransactionState state, Collection<?> keysModified) throws InterruptedException {
      //the retries due to state transfer re-uses the same state. we need that the keys previous locked to be release
      //in order to insert it again in the keys locked.
      //NOTE: this method does not need to be synchronized because it is invoked by a one thread at the time, namely
      //the thread that is delivering the messages in total order.
      state.awaitUntilReset();
      TotalOrderLatch transactionSynchronizedBlock = new TotalOrderLatchImpl(state.getGlobalTransaction().globalId());
      state.setTransactionSynchronizedBlock(transactionSynchronizedBlock);
         //this will collect all the count down latch corresponding to the previous transactions in the queue
         for (Object key : keysModified) {
            TotalOrderLatch prevTx = keysLocked.put(key, transactionSynchronizedBlock);
            if (prevTx != null) {
               state.addSynchronizedBlock(prevTx);
            }
            state.addLockedKey(key);

      }

      TotalOrderLatch stateTransfer = stateTransferInProgress.get();
      if (stateTransfer != null) {
         state.addSynchronizedBlock(stateTransfer);
      }

      if (trace) {
         log.tracef("Transaction [%s] will wait for %s and locked %s", state.getGlobalTransaction().globalId(),
                    state.getConflictingTransactionBlocks(), state.getLockedKeys() == null ? "[ClearCommand]" :
               state.getLockedKeys());
      }
   }
369175
private static void waitForHBaseToComeOnline(MiniHBaseCluster hbaseCluster)
      throws IOException, InterruptedException {
    // Wait for the master to be initialized. This is required because even
    // before it's initialized, the regionserver can come online and the meta
    // table can be scannable. If the cluster is quickly shut down after all of
    // this before the master is initialized, it can cause the shutdown to hang
    // indefinitely as initialization tasks will block forever.
    //
    // Unfortunately, no method available to wait for master to come online like
    // regionservers, so we use a while loop with a sleep so we don't hammer the
    // isInitialized method.
    while (!hbaseCluster.getMaster().isInitialized()) {
      Thread.sleep(1000);
    }
    // Now wait for the regionserver to come online.
    hbaseCluster.getRegionServer(0).waitForServerOnline();
    // Don't leave here till we've done a successful scan of the hbase:meta
    // This validates that not only is the regionserver up, but that the
    // meta region is online so there are no race conditions where operations
    // requiring the meta region might run before it's available. Otherwise,
    // operations are susceptible to region not online errors.
    HTable t = new HTable(hbaseCluster.getConf(), HBASE_META_TABLE);
    ResultScanner s = t.getScanner(new Scan());
    while (s.next() != null) {
      continue;
    }
    s.close();
    t.close();
  }
653659
public void waitFor(long timeout) throws JSchException, ExitCode, InterruptedException {
        long deadline;

        try {
            deadline = System.currentTimeMillis() + timeout;
            while (!channel.isClosed()) {
                if (System.currentTimeMillis() >= deadline) {
                    throw new TimeoutException(this);
                }
                Thread.sleep(100); // throws InterruptedException
            }
            if (channel.getExitStatus() != 0) {
                throw new ExitCode(new Launcher(command), channel.getExitStatus());
            }
        } finally {
            channel.disconnect();
        }
    }
690482
@Override
	protected final AbstractMessage receiveFromDestination(long timeout, boolean duplicateRequired) throws JMSException
    {
        if (closed)
            return null; // [JMS SPEC]

        // Get something from preftech queue
        return getFromPrefetchQueue(timeout);
    }
53067
public long waitForZkTxId(long zkTxId, Object timeout)
    throws TimeoutException, InterruptedException
  {
    long endTime = ClockUtils.toEndTime(clock, timeout);

    synchronized(_lock)
    {
      while(_lastZkTxId < zkTxId)
      {
        ConcurrentUtils.awaitUntil(clock, _lock, endTime);
      }

      return _lastZkTxId;
    }
  }
334288
private void doSend(byte[] msg, boolean wait, KNXAddress dst)
		throws KNXAckTimeoutException, KNXLinkClosedException
	{
		if (closed)
			throw new KNXLinkClosedException("link closed");
		try {
			logger.info("send message to " + dst + (wait ? ", wait for ack" : ""));
			logger.trace("EMI " + DataUnitBuilder.toHex(msg, " "));
			conn.send(msg, wait);
			logger.trace("send to " + dst + " succeeded");
		}
		catch (final KNXPortClosedException e) {
			logger.error("send error, closing link", e);
			close();
			throw new KNXLinkClosedException("link closed, " + e.getMessage());
		}
	}
379334
def wait_multi(self, channels, allowed_methods=None, timeout=None):
        """"""
        chanmap = dict((chan.channel_id, chan) for chan in channels)
        chanid, method_sig, args, content = self._wait_multiple(
                chanmap.keys(), allowed_methods, timeout=timeout)

        channel = chanmap[chanid]

        if content \
        and channel.auto_decode \
        and hasattr(content, 'content_encoding'):
            try:
                content.body = content.body.decode(content.content_encoding)
            except Exception:
                pass

        amqp_method = channel._METHOD_MAP.get(method_sig, None)

        if amqp_method is None:
            raise Exception('Unknown AMQP method (%d, %d)' % method_sig)

        if content is None:
            return amqp_method(channel, args)
        else:
            return amqp_method(channel, args, content)
381773
@Override
    public void resumeConsumer(int suspendFlag)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "resumeConsumer", this);

        // Use this flag to identify whether a new thread must be spun off to look for more
        // messages
        boolean spinOffNewThread = false;
        try
        {
            // Take the lock
            this.lock();

            if (_consumerSuspended)
            {
                boolean setStillSuspended = false;

                // If we're resuming due to the ConsumerSet being resumed then we need to
                // re-check the set's state due to us not holding the set's activeMessageCount
                // lock at this time (due to the hierarchy). And if we find that the set is
                // actually suspended (due to someone else in the set getting in and adding
                // another message then we just ignore this resume as there'll be another one
                // along when the count drops back below the limit again.
                if (suspendFlag == DispatchableConsumerPoint.SUSPEND_FLAG_SET_ACTIVE_MSGS)
                    setStillSuspended = _consumerKey.isConsumerSetSuspended();

                if (!setStillSuspended)
                {
                    // clear the bit provided in the _suspendFlags
                    _suspendFlags &= ~suspendFlag;

                    if (_suspendFlags == 0) // No flags set so resume the consumer
                    {
                        _consumerSuspended = false;

                        // If the consumer is still active (started) we need
                        // to kickstart the consumer back into life to check for more
                        // messages
                        if (!_stopped && !_closing)
                        {
                            if (_asynchConsumerRegistered)
                            {
                                if (!_asynchConsumer.isAsynchConsumerRunning())
                                {
                                    // We'll have to take the full lock hierarchy and
                                    // possibly spin off another thread to start looking for more messages
                                    // (just like start())
                                    spinOffNewThread = true;
                                }
                            }
                            else
                            {
                                // Wake up any synchronous receivers
                                _waiter.signal();
                            }
                        }
                    }
                }
            }
        } finally
        {
            this.unlock();
        }

        // start a new thread to run the callback if we determined that it was necessary   
        if (spinOffNewThread)
        {
            try
            {
                _messageProcessor.startNewThread(new AsynchThread(this, false));
            } catch (InterruptedException e)
            {
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.JSLocalConsumerPoint.resumeConsumer",
                                            "1:4978:1.22.5.1",
                                            this);

                SibTr.exception(tc, e);

                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "resumeConsumer", e);
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "resumeConsumer");
    }
766276
private synchronized void asyncResponse(String command, SocketChannel sc) {
        if (closed) {
            Utils.tryToClose(sc);
        } else {
            Thread thread = new Thread(new ResponseThread(command, sc), "kernel-" + command + "-command-response");

            // We allow a maximum of one outstanding status start or stop command
            Thread oldThread = responseThread.getAndSet(thread);
            if (oldThread != null) {
                oldThread.interrupt();
            }

            thread.start();
        }
    }
181067
@Override
    public void waitForData() throws CouldNotPerformException, InterruptedException {
        try {
            if (isDataAvailable()) {
                return;
            }
            logger.debug("Wait for " + this.toString() + " data...");
            getDataFuture().get();
            dataObservable.waitForValue();
        } catch (ExecutionException | CancellationException ex) {
            if (shutdownInitiated) {
                throw new InterruptedException("Interrupt request because system shutdown was initiated!");
            }
            throw new CouldNotPerformException("Could not wait for data!", ex);
        }
    }
381044
private void sendAndRetry(AmqpTemplate template, Object message) throws InterruptedException
    {
        try
        {
            template.convertAndSend(message);
        }
        catch (AmqpException ex)
        {
            Thread.sleep(getRetryInterval());
            template.convertAndSend(message);
        }
    }
670587
protected function wait_channel($channel_id, $timeout = 0)
    {
        // Keeping the original timeout unchanged.
        $_timeout = $timeout;
        while (true) {
            $now = time();
            try {
                list($frame_type, $frame_channel, $payload) = $this->wait_frame($_timeout);
            } catch (AMQPTimeoutException $e) {
                if ( $this->heartbeat && microtime(true) - ($this->heartbeat*2) > $this->last_frame ) {
                    $this->debug->debug_msg("missed server heartbeat (at threshold * 2)");
                    $this->setIsConnected(false);
                    throw new AMQPHeartbeatMissedException("Missed server heartbeat");
                }

                throw $e;
            }

            $this->last_frame = microtime(true);

            if ($frame_channel === 0 && $frame_type === 8) {
                // skip heartbeat frames and reduce the timeout by the time passed
                $this->debug->debug_msg("received server heartbeat");
                if($_timeout > 0) {
                    $_timeout -= time() - $now;
                    if($_timeout <= 0) {
                        // If timeout has been reached, throw the exception without calling wait_frame
                        throw new AMQPTimeoutException("Timeout waiting on channel");
                    }
                }
                continue;

            } else {

                if ($frame_channel == $channel_id) {
                    return array($frame_type, $payload);
                }

                // Not the channel we were looking for.  Queue this frame
                //for later, when the other channel is looking for frames.
                // Make sure the channel still exists, it could have been
                // closed by a previous Exception.
                if (isset($this->channels[$frame_channel])) {
                    array_push($this->channels[$frame_channel]->frame_queue, array($frame_type, $payload));
                }

                // If we just queued up a method for channel 0 (the Connection
                // itself) it's probably a close method in reaction to some
                // error, so deal with it right away.
                if (($frame_type == 1) && ($frame_channel == 0)) {
                    $this->wait();
                }
            }
        }
    }
1066462
@Override
    public long dispose(long channelIdentifier) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "dispose", Long.valueOf(channelIdentifier));
        }
        long rc = 0L;
        synchronized (oneAtATime) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "have lock");
            }

            if (aioInitialized == AIO_INITIALIZED) {

                try {
                    rc = aio_dispose(channelIdentifier);
                } catch (AsyncException ae) {
                    // just log any errors
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "Error disposing channel: " + channelIdentifier + ", error: " + ae.getMessage());
                    }
                } catch (Throwable t) {
                    if (aioInitialized != AIO_SHUTDOWN) {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "caught throwable:" + t);
                        }
                        throw new RuntimeException("Throwable caught from aio dll: aio_dispose: " + t.getMessage());
                    }
                }
            }
        } // end-sync
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "dispose");
        }
        return rc;
    }
886078
private void authenticateAndWait(final CountDownLatch latch) throws InterruptedException, BitfinexClientException {
		if (authenticated) {
			return;
		}
		sendCommand(new AuthCommand(configuration.getAuthNonceProducer()));
		logger.debug("Waiting for connection ready events");
		latch.await(10, TimeUnit.SECONDS);

		if (!authenticated) {
			throw new BitfinexClientException("Unable to perform authentication, permissions are: " + permissions);
		}
	}
539000
private IGPSObject waitForResult() {
		synchronized (this.asyncWaitMutex) {
			if (this.asyncResult == null) {
				try {
					this.asyncWaitMutex.wait(1000);
				} catch (final InterruptedException e) {
					GPSdEndpoint.LOG.info("Interrupted while waiting for result", e);
				}
			}
			final IGPSObject result = this.asyncResult;
			this.asyncResult = null;
			return result;
		}
	}
747166
@Override
    public Message read() throws InterruptedException, IOException {
        Message message;

        message = super.read();

        if (message == ISOBlueBus.MESSAGE_NONE) {
            throw new ClosedChannelException();
        }

        return message;
    }