CustomExecutorListener.java

public class CustomExecutorListener<REQ> extends Listener<REQ> {
  private final Listener<REQ> delegate;
  private final ServerCall<REQ, ?> call;
  private final Executor executor;

  private CustomExecutorListener(
      Listener<REQ> delegate,
      ServerCall<REQ, ?> call,
      Executor executor
  ) {
    this.delegate = delegate;
    this.call = call;
    this.executor = Context.current().fixedContextExecutor(
        MoreExecutors.newSequentialExecutor(executor)
    );
  }

  public static <REQ> Listener<REQ> wrap(
      Listener<REQ> delegate,
      ServerCall<REQ, ?> call,
      Executor executor
  ) {
    return new CustomExecutorListener<>(delegate, call, executor);
  }

  @Override
  public void onMessage(REQ message) {
    executor.execute(() -> {
      try {
        delegate.onMessage(message);
      } catch (RuntimeException | Error e) {
        closeCall();
        throw e;
      }
    });
  }

  @Override
  public void onHalfClose() {
    executor.execute(() -> {
      try {
        delegate.onHalfClose();
      } catch (RuntimeException | Error e) {
        closeCall();
        throw e;
      }
    });
  }

  @Override
  public void onReady() {
    executor.execute(() -> {
      try {
        delegate.onReady();
      } catch (RuntimeException | Error e) {
        closeCall();
        throw e;
      }
    });
  }

  @Override
  public void onCancel() {
    executor.execute(delegate::onCancel);
  }

  @Override
  public void onComplete() {
    executor.execute(delegate::onComplete);
  }

  private void closeCall() {
    call.close(Status.UNKNOWN, new Metadata());
  }
}