if (logger.isDebugEnabled()) {
                    logger.debug("Waiting for crawler termination: thread={}, timeout={}ms", parentThread.getName(), millis);
                }
                parentThread.join(millis);
            } catch (final InterruptedException e) {
                logger.warn("Crawler thread was interrupted while waiting for termination: thread={}, sessionId={}", parentThread.getName(),

    latch = new CountDownLatch(1);
    future = createListenableFuture(Boolean.TRUE, null, latch);
  }

  @Override
  protected void tearDown() throws Exception {

    // Make sure we have no waiting threads.
    latch.countDown();
  }

  /** Constructs a listenable future with a value available after the latch has counted down. */
  protected abstract <V> ListenableFuture<V> createListenableFuture(

    TimeoutException e =
        assertThrows(
            TimeoutException.class, () -> service.startAsync().awaitRunning(1, MILLISECONDS));
    assertThat(e)
        .hasMessageThat()
        .isEqualTo("Timed out waiting for Foo [STARTING] to reach the RUNNING state.");
  }

  private static class TestService extends AbstractIdleService {
    int startUpCalled = 0;
    int shutDownCalled = 0;

    long timeoutNanos = unit.toNanos(timeout);
    if (timeoutNanos <= OverflowAvoidingLockSupport.MAX_NANOSECONDS_THRESHOLD) {
      return super.get(timeout, unit);
    }
    // Waiting 68 years should be enough for any program.
    return super.get(
        min(timeoutNanos, OverflowAvoidingLockSupport.MAX_NANOSECONDS_THRESHOLD), NANOSECONDS);
  }

    } else {
      events.add(Failure(t, response))
    }
  }

  private fun nextEvent(): Any =
    events.poll(10, TimeUnit.SECONDS)
      ?: throw AssertionError("Timed out waiting for event.")

  fun assertTextMessage(payload: String?) {
    assertThat(nextEvent()).isEqualTo(Message(string = payload))
  }

  fun assertBinaryMessage(payload: ByteString?) {

     * @param ctx The content indexing context
     * @param reader The supplier for DocumentReader
     * @param docPerReq The number of documents to process per request
     * @param waitController The runnable to control waiting between requests
     * @return A Promise that will be resolved with the SuggestIndexResponse
     */

   * found in any textbook.
   */

  /** Monitor guarding all access */
  final Monitor monitor;

  /** Guard for waiting takes */
  private final Monitor.Guard notEmpty;

  /** Guard for waiting puts */
  private final Monitor.Guard notFull;

  // Internal helper methods

  /** Circularly increment i. */
  final int inc(int i) {

    }
  }

  private fun readAndWriteFrames() {
    check(socket == null)
    val socket = serverSocket!!.accept()!!
    this.socket = socket

    // Bail out now if this instance was closed while waiting for the socket to accept.
    synchronized(this) {
      if (executor.isShutdown) {
        socket.close()
        return
      }
    }
    val outputStream = socket.getOutputStream()

        .that(input)
        .isNotInstanceOf(ListenableFuture.class);
    ListenableFuture<String> listenable = listenInPoolThread(input);
    /*
     * This will occur before the waiting get() in the
     * listenInPoolThread-spawned thread completes:
     */
    RecordingRunnable earlyListener = new RecordingRunnable();
    listenable.addListener(earlyListener, directExecutor());

   * Creates a TimeLimiter instance using the given executor service to execute method calls.
   *
   * <p><b>Warning:</b> using a bounded executor may be counterproductive! If the thread pool fills
   * up, any time callers spend waiting for a thread may count toward their time limit, and in this
   * case the call may even time out before the target method is ever invoked.
   *