}

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

  /**

   *
   * @param executor The executor to decorate
   * @param nameSupplier The source of names for each task
   */
  @J2ktIncompatible
  @GwtIncompatible // concurrency
  static Executor renamingDecorator(Executor executor, Supplier<String> nameSupplier) {
    checkNotNull(executor);
    checkNotNull(nameSupplier);
    return command -> executor.execute(threadRenaming(command, nameSupplier));
  }

                  + runnable
                  + " with executor "
                  + executor,
              e);
    }
  }

  private static final class RunnableExecutorPair {
    final Runnable runnable;
    final Executor executor;
    @Nullable RunnableExecutorPair next;

    RunnableExecutorPair(
        Runnable runnable, Executor executor, @Nullable RunnableExecutorPair next) {

    barrier.await(1, SECONDS);
    assertFalse(executor.isShutdown());
    assertFalse(executor.isTerminated());

    executor.shutdown();
    assertTrue(executor.isShutdown());
    assertThrows(RejectedExecutionException.class, () -> executor.submit(() -> {}));
    assertFalse(executor.isTerminated());

    // WAIT #2
    barrier.await(1, SECONDS);
    assertFalse(executor.awaitTermination(20, MILLISECONDS));

      Function<? super Throwable, ? extends V> fallback,
      Executor executor) {
    return (FluentFuture<V>) Futures.catching(this, exceptionType, fallback, executor);
  }

  public final FluentFuture<V> catchingAsync(
      Class<Throwable> exceptionType,
      AsyncFunction<? super Throwable, ? extends V> fallback,
      Executor executor) {

      Executor executor) {
    return AbstractCatchingFuture.create(input, exceptionType, fallback, executor);
  }

  public static <V extends @Nullable Object> ListenableFuture<V> catchingAsync(
      ListenableFuture<? extends V> input,
      Class<Throwable> exceptionType,
      AsyncFunction<? super Throwable, ? extends V> fallback,
      Executor executor) {

    SettableFuture<?> future = SettableFuture.create();
    Executor delegate =
        new Executor() {
          @Override
          public void execute(Runnable task) {
            if (future.set(null)) {
              awaitUninterruptibly(latch);
            }
            throw new RejectedExecutionException();
          }
        };
    SequentialExecutor executor = new SequentialExecutor(delegate);

      assertEquals(Service.State.STOPPING, state());
      shutdownCalled++;
    }

    @Override
    protected Executor executor() {
      return executor;
    }

    @Override
    public void tearDown() throws Exception {
      executor.shutdown();
    }
  }

      final Runnable runnable;
      final Executor executor;

      RunnableExecutorPair(Runnable runnable, Executor executor) {
        this.runnable = runnable;
        this.executor = executor;
      }

      void execute() {
        try {
          executor.execute(runnable);
        } catch (RuntimeException e) {
          log.log(
              Level.SEVERE,

    )

  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) {