gRPC サポート

Starting with version 7.1, Spring Integration provides inbound and outbound gateways to communicate via gRPC (英語) protocol.

この依存関係はプロジェクトに必要です:

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-grpc</artifactId>
    <version>7.1.1</version>
</dependency>
implementation "org.springframework.integration:spring-integration-grpc:7.1.1"

gRPC 用の Spring Integration コンポーネントはプロトコルバッファから生成されず、一般的な gRPC サービスやスタブ実装のように型安全ではありません。これは主に、Spring Integration フレームワーク自体の汎用性に起因しています。Spring Integration フレームワークでは、作業単位が Message 抽象化であり、このメッセージのペイロード型は通常、統合コンポーネントの内部ロジックのスコープ外にあります。サービス呼び出し用の gRPC メッセージは、変換の仮定なしにそのまま送受信されます。例: gRPC サービスメソッドが次のようになっている場合:

service TestHelloWorld {

  // Sends a greeting
  rpc SayHello(HelloRequest) returns (HelloReply) {}

  // Sends a greeting and something else
  rpc StreamSayHello(HelloRequest) returns (stream HelloReply) {}

  // Sends a greeting to everyone present
  rpc HelloToEveryOne(stream HelloRequest) returns (HelloReply) {}

  // Streams requests and replies
  rpc BidiStreamHello(stream HelloRequest) returns (stream HelloReply) {}

}

HelloRequest は、受信ゲートウェイ（サーバー）側ではリクエストメッセージペイロードとなり、送信ゲートウェイ（クライアント）側ではリクエストメッセージペイロードとなります。HelloReply は受信ゲートウェイでは応答メッセージペイロードとなり、送信ゲートウェイで受信されます。

GrpcHeaders クラスには、gRPC ゲートウェイの前後のメッセージで使用される（および設定される）ヘッダー名を表す便利な定数が含まれています。例: GrpcHeaders.METHOD_TYPE ヘッダーには、サーバー側（受信ゲートウェイ）の io.grpc.MethodDescriptor.MethodType 列挙値が含まれており、下流ルーティングを容易にします。もう 1 つの便利なヘッダーは GrpcHeaders.SERVICE_METHOD で、サーバー側で呼び出された gRPC サービスメソッド、またはクライアントスタブから呼び出す gRPC サービスメソッドを示します。

受信ゲートウェイの GrpcHeaders.SERVICE_METHOD ヘッダーには、Protobuf で宣言されているとおりの gRPC サービスメソッド名（上記の .proto の例を参照）と、サービス定義の io.grpc.MethodDescriptor に格納されている値が正確に格納されます。

gRPC 用受信ゲートウェイ

GrpcInboundGateway は、gRPC リクエストを受信し、ダウンストリームフローにメッセージを送信し、gRPC レスポンスを生成する MessagingGatewaySupport の実装です。初期化には、このゲートウェイのインスタンスは、通常 Protobuf から生成され、*ImplBase クラス名を持つ、BindableService を実装する抽象 gRPC サービスクラスのみを必要とします。

標準の gRPC サービスのみがサポートされています。GrpcInboundGateway ロジックは、生成された AsyncService 契約に基づいています。Reactor および Kotlin ベースのサービス生成は、ゲートウェイ定義からこれらの型が公開されていないため、Spring Integration ロジックでは意味を成しません。

ゲートウェイは、前述の AsyncService インターフェースを使用してプロキシを作成し、gRPC サービスメソッドをインターセプトします。

以下の例は、GrpcInboundGateway の設定方法を示しています。

@Bean
GrpcInboundGateway helloWorldService() {
    return new GrpcInboundGateway(TestHelloWorldGrpc.TestHelloWorldImplBase.class);
}

GrpcInboundGateway は BindableService を実装し、gRPC サービスの AsyncService 契約用のプロキシに基づいて ServerServiceDefinition を公開します。このゲートウェイのインスタンスを ServerBuilder に登録する必要があり、アプリケーション内で他の *ImplBase 実装は必要ありません。

Spring gRPC と BindableService 実装の自動検出機能を使用する場合、GrpcInboundGateway はトップレベルの Bean として宣言する必要があります。そのため、IntegrationFlow.from(new GrpcInboundGateway(TestHelloWorldGrpc.TestHelloWorldImplBase.class)) のような Java DSL API は推奨されません。なぜなら、そのような BindableService 実装では、対応する Spring gRPC インフラストラクチャから認識されないからです。

GrpcInboundGateway は、sendAndReceiveMessageReactive() API を使用してダウンストリームフローとやり取りし、Mono のレスポンスを gRPC StreamObserver に適合させます。前述のとおり、リクエストメッセージのペイロードはまさに gRPC リクエストメッセージであり、gRPC レスポンスメッセージの形式でのレスポンスを期待します。ダウンストリームロジックは型安全であり、*ImplBase を手動で実装する場合と同様の方法で gRPC メッセージを処理できます。

MethodDescriptor.MethodType.UNARY と MethodDescriptor.MethodType.BIDI_STREAMING は、ダウンストリーム処理ロジックの観点からは同じです。つまり、BIDI_STREAMING はリクエストアイテムのループとして処理され、ゲートウェイはレスポンスアイテムをレスポンス StreamObserver に即座に生成します。異なる BIDI_STREAMING ロジックについては、通常の gRPC サービス実装が推奨されます。

MethodDescriptor.MethodType.CLIENT_STREAMING モードでは、gRPC リクエスト項目のペイロードとして Flux を含むメッセージが生成されます。

MethodDescriptor.MethodType.SERVER_STREAMING モードの場合、レスポンスペイロードは単一の gRPC レスポンスメッセージ、またはそれらの Flux となる。

以下の例は、前述の TestHelloWorldGrpc.TestHelloWorldImplBase サービスに対する IntegrationFlow の実装例を示しています。

@Bean
IntegrationFlow grpcIntegrationFlow(GrpcInboundGateway helloWorldService) {
    return IntegrationFlow.from(helloWorldService)
            .route(Message.class, message ->
                    		message.getHeaders().get(GrpcHeaders.SERVICE_METHOD, String.class),
                    router -> router

                            .subFlowMapping("SayHello", flow -> flow
                                    .transform(this::requestReply))

                            .subFlowMapping("StreamSayHello", flow -> flow
                                    .transform(this::streamReply))

                            .subFlowMapping("HelloToEveryOne", flow -> flow
                                    .transformWith(transformSpec -> transformSpec
                                          .transformer(this::streamRequest)
                                          .async(true)))

                            .subFlowMapping("BidiStreamHello", flow -> flow
                                    .transform(this::requestReply))
            )
            .get();
}

private HelloReply requestReply(HelloRequest helloRequest) {
    return newHelloReply("Hello " + helloRequest.getName());
}

private Flux<HelloReply> streamReply(HelloRequest helloRequest) {
    return Flux.just(
           newHelloReply("Hello " + helloRequest.getName()),
           newHelloReply("Hello again!"));
}

private Mono<HelloReply> streamRequest(Flux<HelloRequest> request) {
    return request
            .map(HelloRequest::getName)
            .collectList()
            .map(names -> StringUtils.collectionToDelimitedString(names, ", "))
            .map("Hello "::concat)
            .map(TestConfig::newHelloReply);
}

private static HelloReply newHelloReply(String message) {
    return HelloReply.newBuilder().setMessage(message).build();
}

ルーティングは、GrpcInboundGateway によって設定された GrpcHeaders.SERVICE_METHOD ヘッダー上で行われます。ダウンストリームのトランスフォーマーのすべてのビジネスメソッドは、TestHelloWorldGrpc.TestHelloWorldImplBase サービス用の gRPC メッセージに関して型安全です。

エラー処理

ダウンストリーム処理中にエラーが発生した場合、GrpcInboundGateway は結果として得られた Throwable を gRPC StatusException にマッピングしようとします。その際、Status.fromThrowable(throwable) を利用して特定の gRPC ステータスコード (Status.UNAVAILABLE、Status.INVALID_ARGUMENT など) を保持します。ステータスコードが UNKNOWN と評価された場合、ゲートウェイは "Internal Server Error" の汎用的な説明にフォールバックします。これにより、クライアントは元の gRPC ステータスが提供された場合に、それを確実に受け取ることができます。

DSL を使用した設定

Grpc ファクトリを使用して、Java DSL を使用するフローに GrpcInboundGateway を追加します。

@Bean
IntegrationFlow grpcInboundFlow() {
    return IntegrationFlow.from(
                    Grpc.inboundGateway(TestSingleHelloWorldGrpc.TestSingleHelloWorldImplBase.class)
                        .requestTimeout(3000L))
            .transform(this::requestReply)
            .get();
}

private HelloReply requestReply(HelloRequest helloRequest) {
    return HelloReply.newBuilder().setMessage("Hello " + helloRequest.getName()).build();
}

gRPC 用送信ゲートウェイ

GrpcOutboundGateway は、gRPC リクエストをリモート gRPC サーバーに送信し、gRPC スタブとして動作するレスポンスを受信する AbstractReplyProducingMessageHandler の実装です。初期化には、このゲートウェイのインスタンスに gRPC Channel と gRPC サービスクラス（例: TestHelloWorldGrpc.class）が必要です。

ゲートウェイは、サービスの ServiceDescriptor から取得した gRPC メソッドを動的に呼び出します。以下の gRPC 通信パターンをサポートしています。

  • Unary : 単一リクエスト→ async が true の場合は Mono が返され、そうでない場合はレスポンスオブジェクトが返されます

  • Server streaming : 単一のリクエスト→ 複数のレスポンスの Flux 

  • Client streaming : 複数のリクエスト→ 単一のレスポンスによる Mono 

  • Bidirectional streaming : 複数のリクエスト→ 複数のレスポンスの Flux 

GrpcOutboundGateway はデフォルトで非同期モードです。ゲートウェイ設定で setAsync(false) から非同期モードを無効にすることができます。詳細は非同期サービスアクティベーターを参照してください。

メソッド名構成

呼び出すメソッド名は、以下の 4 つの方法で設定できます。

  1. 単一メソッドのサービス向け Auto-detection: 

    @Bean
public GrpcOutboundGateway grpcOutboundGateway(ManagedChannel channel) {
    // When TestSingleMethodGrpc has only one method, it will be auto-detected
    return new GrpcOutboundGateway(channel, TestSingleMethodGrpc.class);
}

  2. Explicit method name が setMethodName() を使用する:

    @Bean
public GrpcOutboundGateway grpcOutboundGateway(ManagedChannel channel) {
    GrpcOutboundGateway gateway = new GrpcOutboundGateway(channel, TestHelloWorldGrpc.class);
    gateway.setMethodName("SayHello");
    return gateway;
}

  3. Dynamic resolution 経由 setMethodNameExpression():

    @Bean
public GrpcOutboundGateway dynamicMethodGateway(ManagedChannel channel) {
    GrpcOutboundGateway gateway = new GrpcOutboundGateway(channel, TestHelloWorldGrpc.class);
    gateway.setMethodNameExpression(new SpelExpressionParser().parseExpression("payload.class.simpleName"));
    return gateway;
}

  4. メソッド名もメソッド名式も設定されておらず、サービスが複数のメソッドを提供している場合、ゲートウェイは入力メッセージ内の GrpcHeaders.SERVICE_METHOD ヘッダーを検索して、呼び出すメソッドを決定します。GrpcHeaders.SERVICE_METHOD ヘッダーが見つからない場合は、IllegalStateException 例外がスローされます。

    @Bean
public GrpcOutboundGateway dynamicMethodGateway(ManagedChannel channel) {
	 // Looks for GrpcHeaders.SERVICE_METHOD header in the input message
    return new GrpcOutboundGateway(channel, TestHelloWorldGrpc.class);
}

リクエストペイロード処理

GrpcOutboundGateway は MethodDescriptor からメソッド型を自動的に検出し、呼び出しを適切に処理します。

  • Unary methods accept a single gRPC request message, returning a Mono<ResponseType> in the async mode (by default). If async is set to false, then the response object is returned into a reply message payload as is.

  • Server streaming methods accept a single gRPC request message and return a Flux<ResponseType>.

  • Client streaming and Bidirectional streaming methods accept flexible input types:

    • Flux<RequestType>

    • Mono<RequestType>

    • Stream<RequestType>

    • Collection<RequestType>

    • RequestTypes[]

    • Single RequestType object

Client streaming methods return a Mono<ResponseType>, while bidirectional streaming methods return a Flux<ResponseType> containing the response.

DSL を使用した設定

Use the Grpc factory to add the GrpcOutboundGateway to flows using the Java DSL . The simplest configuration for a service with a single method:

@Bean
IntegrationFlow grpcOutboundFlow(ManagedChannel channel) {
	return f -> f
            .handle(Grpc.outboundGateway(channel, TestSingleHelloWorldGrpc.class))
            .transform(this::upperCase);
}

private HelloReply upperCase(HelloReply helloReply) {
    return HelloReply.newBuilder().setMessage(helloReply.getMessage().toUpperCase()).build();
}

When the gRPC service has only one method, it will be auto-detected.

For services with multiple methods, use the DSL’s .methodName(),.methodNameExpression(), or .methodNameFunction() methods. See メソッド名構成。