gRPC (C#) に追加されたインターセプターの使用方法

「試験的」と銘打たれてはいますが、gRPC C# 1.10 でインターセプターが追加されました。どのように使うのかを確認してみました。

クライアントサイド

Channel または CallInvoker に対してインターセプターを設定する拡張メソッドが追加されています。拡張メソッドは Grpc.Core.Interceptors 名前空間に定義されています。サービスクライアントを生成するとき、Channel または CallInvoker のどちらかに対してインターセプターを設定します。Channel と CallInvoker の両方にインターセプターを設定した場合は両方が呼び出されます。

Channel を使用する場合

var channel = new Channel("localhost:50000"), ChannelCredentials.Insecure);

// インターセプターを設定
// インターセプターを呼び出すように実装された Channel が返される
var interceptor = new SampleInterceptor();
channel = channel.Intercept(interceptor);

var client = new SampleService.SampleClient(channel);

CallInvoker を使用する場合

var channel = new Channel("localhost:50000"), ChannelCredentials.Insecure);

var invoker = new DefaultCallInvoker(channel);

// インターセプターを設定
// インターセプターを呼び出すように実装された CallInvoker が返される
var interceptor = new SampleInterceptor();
invoker = invoker.Intercept(interceptor);

var client = new SampleService.SampleClient(invoker);

インターセプターは Grpc.Core.Interceptors.Interceptor クラスを継承して実装します。クライアントサイドで使用するインターセプターでは、次の 5 つのメソッドをオーバライドします。RPC メソッドの呼び出しの前後に割り込ませたい処理を実装します。

private class SampleInterceptor : Interceptor
{

  public override AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(
    ClientInterceptorContext<TRequest, TResponse> context
    , AsyncClientStreamingCallContinuation<TRequest, TResponse> continuation
    )
  {
      return base.AsyncClientStreamingCall(context, continuation);
  }

  public override AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(
    ClientInterceptorContext<TRequest, TResponse> context
    , AsyncDuplexStreamingCallContinuation<TRequest, TResponse> continuation
    )
  {
    return base.AsyncDuplexStreamingCall(context, continuation);
  }

  public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(
    TRequest request
    , ClientInterceptorContext<TRequest, TResponse> context
    , AsyncServerStreamingCallContinuation<TRequest, TResponse> continuation
    )
  {
    return base.AsyncServerStreamingCall(request, context, continuation);
  }

  public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
    TRequest request
    , ClientInterceptorContext<TRequest, TResponse> context
    , AsyncUnaryCallContinuation<TRequest, TResponse> continuation
    )
  {
    return base.AsyncUnaryCall(request, context, continuation);
  }

  public override TResponse BlockingUnaryCall<TRequest, TResponse>(
    TRequest request
    , ClientInterceptorContext<TRequest, TResponse> context
    , BlockingUnaryCallContinuation<TRequest, TResponse> continuation
    )
  {
    return base.BlockingUnaryCall(request, context, continuation);
  }

}

 

サーバーサイド

ServerServiceDefinition に対してインターセプターを設定する拡張メソッドが追加されています。この拡張メソッドも Grpc.Core.Interceptors 名前空間に定義されています。サービスインスタンスを生成してサーバーに登録するとき、インターセプターを設定します。

var service = SampleService.BuildService(new SampleServiceImpl());

// インターセプターを設定
// インターセプターを呼び出すように実装された ServerServiceDefinition が返される
service = service.Intercept(new SampleInterceptor());

Server server = new Server();
server.Services.Add(service);

サーバーサイドの場合もインターセプターは Grpc.Core.Interceptors.Interceptor クラスを継承して実装します。次の 4 つのメソッドをオーバライドします。RPC メソッドの呼び出しの前後に割り込ませたい処理を実装します。下の ClientStreamingServerHandler メソッドのように context.Status にエラーステータスを設定すれば RPC メソッドの呼び出しを止められます。独自の認証を実装するような場合に使えます。

private class SampleInterceptor : Interceptor
{

  public override Task<TResponse> ClientStreamingServerHandler<TRequest, TResponse>(
    IAsyncStreamReader<TRequest> requestStream
    , ServerCallContext context
    , ClientStreamingServerMethod<TRequest, TResponse> continuation
    )
  {
    // エラーステータスを設定
    context.Status = new Status(StatusCode.Aborted, "aborted by interceptor.");
    return base.ClientStreamingServerHandler(requestStream, context, continuation);
  }

  public override Task DuplexStreamingServerHandler<TRequest, TResponse>(
    IAsyncStreamReader<TRequest> requestStream
    , IServerStreamWriter<TResponse> responseStream
    , ServerCallContext context
    , DuplexStreamingServerMethod<TRequest, TResponse> continuation
    )
  {
    return base.DuplexStreamingServerHandler(requestStream, responseStream, context, continuation);
  }

  public override Task ServerStreamingServerHandler<TRequest, TResponse>(
    TRequest request
    , IServerStreamWriter<TResponse> responseStream
    , ServerCallContext context
    , ServerStreamingServerMethod<TRequest, TResponse> continuation
    )
  {
    return base.ServerStreamingServerHandler(request, responseStream, context, continuation);
  }

  public override Task<TResponse> UnaryServerHandler<TRequest, TResponse>(
    TRequest request
    , ServerCallContext context
    , UnaryServerMethod<TRequest, TResponse> continuation
    )
  {
    return base.UnaryServerHandler(request, context, continuation);
  }

}