}

  /**
   * Adds a listener that will be called using the given executor when events are later {@link
   * #enqueue enqueued} and {@link #dispatch dispatched}.
   */
  public void addListener(L listener, Executor executor) {
    checkNotNull(listener, "listener");
    checkNotNull(executor, "executor");
    listeners.add(new PerListenerQueue<>(listener, executor));
  }

  /**

          public String get(long time, TimeUnit unit) {
            throw new CancellationException(); // BAD!!
          }

          @Override
          public void addListener(Runnable runnable, Executor executor) {
            executor.execute(runnable);
          }
        };
    future.setFuture(badFuture);
    ExecutionException expected = getExpectingExecutionException(future);

  }

  public void testNoOpScheduledExecutorShutdown() {
    ListeningScheduledExecutorService executor = TestingExecutors.noOpScheduledExecutor();
    assertFalse(executor.isShutdown());
    assertFalse(executor.isTerminated());
    executor.shutdown();
    assertTrue(executor.isShutdown());
    assertTrue(executor.isTerminated());
  }

   */
  public <T extends @Nullable Object> ListenableFuture<T> submitAsync(
      AsyncCallable<T> callable, Executor executor) {
    checkNotNull(callable);
    checkNotNull(executor);
    TaskNonReentrantExecutor taskExecutor = new TaskNonReentrantExecutor(executor, this);
    AsyncCallable<T> task =
        new AsyncCallable<T>() {
          @Override

   *
   * @param listener the listener to run when the computation is complete
   * @param executor the executor to run the listener in
   * @throws RejectedExecutionException if we tried to execute the listener immediately but the
   *     executor rejected it.
   */
  void addListener(Runnable listener, Executor executor);

   *
   * @param listener the listener to run when the computation is complete
   * @param executor the executor to run the listener in
   * @throws RejectedExecutionException if we tried to execute the listener immediately but the
   *     executor rejected it.
   */
  void addListener(Runnable listener, Executor executor);

    )
  }

  fun connectionCount(): Int = connectionCount.get()

  fun shutdown() {
    serverSocket!!.close()
    executor.shutdown()
    if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
      throw IOException("Gave up waiting for executor to shut down")
    }
  }

  private fun service(from: Socket) {
    val name = "SocksProxy ${from.remoteSocketAddress}"
    threadName(name) {

    }


    @Override
    @Before
    public void setUp () throws Exception {
        super.setUp();
        this.executor = Executors.newCachedThreadPool();
    }


    @After
    @Override
    public void tearDown () throws Exception {
        this.executor.shutdown();
        this.executor.awaitTermination(10, TimeUnit.SECONDS);
        super.tearDown();
    }


    @Test

   *
   * @param listener the listener to run when the computation is complete
   * @param executor the executor to run the listener in
   * @throws RejectedExecutionException if we tried to execute the listener immediately but the
   *     executor rejected it.
   */
  void addListener(Runnable listener, Executor executor);

    executor.execute(task);
    return task;
  }

  /**
   * Executes {@code runnable} on the specified {@code executor}, returning a {@code Future} that
   * will complete after execution.
   *
   * @throws RejectedExecutionException if the task cannot be scheduled for execution
   * @since 28.2
   */
  public static ListenableFuture<@Nullable Void> submit(Runnable runnable, Executor executor) {