grpc-java async call in ServerInterceptor

Question

I have a grpc-java server and need to make an async call to an auth service, before processing a request. I think this should be done in an interceptor, but it requires to synchronously return a Listener from interceptCall()

class AuthInterceptor implements ServerInterceptor {

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
        ServerCall<ReqT, RespT> call,
        Metadata headers,
        ServerCallHandler<ReqT, RespT> next
    ) {
        String token = ""; //get token from headers
        authService.authorize(token).subscribe(
            ok -> // process the request
            error -> call.close(Status.UNAUTHENTICATED, headers)
        );
        // Here we need to return a Listener, but we haven't started a call yet
    }
}

So the questions are: how to make an async call from a ServerInterceptor, and if it can't be done, what is the right way to asynchronously authenticate requests it grpc? I know it can be done directly in grpc services with StreamObservers, but request authorization is a cross cutting concern and interceptors seem to be a perfect place for it.

Can you update your question showing what you ended up with using the DelayedListener? It's unclear to me how to connect the dots. — Abhijit Sarkar

Eric Anderson Eric Anderson · Accepted Answer · 2018-12-06T17:23:36

You do need to return a ServerCall.Listener. But since you don't know the Listener to delegate to, you can override each method in the Listener to add the callback to a queue. Once the authentication is complete, drain the queue.

class DelayedListener<ReqT> extends Listener<ReqT> {
  private Listener<ReqT> delegate;
  private List<Runnable> events = new ArrayList<Runnable>();

  @Override public synchronized void onMessage(ReqT message) {
    if (delegate == null) {
      events.add(() -> delegate.onMessage(message));
    } else {
      delegate.onMessage(message);
    }
  }
  ...
  public synchronized void setDelegate(Listener<ReqT> delegate) {
    this.delegate = delegate;
    for (Runnable runnable : events) {
      runnable.run();
    }
    events = null;
  }
}

grpc-java async call in ServerInterceptor

1 Answers