GH-10623: Add GrpcInboundGateway

build.gradle

@@ -62,6 +62,7 @@ ext {
     graalvmVersion = '25.0.1'
     greenmailVersion = '2.1.8'
     groovyVersion = '5.0.3'
+	grpcVersion = '1.77.0'
     hamcrestVersion = '3.0'
     hazelcastVersion = '5.6.0'
     hibernateVersion = '7.1.12.Final'
@@ -173,6 +174,7 @@ allprojects {
 		dependencyManagement(platform("org.springframework.security:spring-security-bom:$springSecurityVersion"))
 		dependencyManagement(platform("org.springframework.ws:spring-ws-bom:$springWsVersion"))
 		dependencyManagement(platform("org.mongodb:mongodb-driver-bom:$mongoDriverVersion"))
+		dependencyManagement(platform("io.grpc:grpc-bom:$grpcVersion"))
 	}
 }
 
@@ -627,6 +629,45 @@ project('spring-integration-groovy') {
     }
 }
 
+project('spring-integration-grpc') {
+	description = 'Spring Integration gRPC Support'
+
+	apply plugin: 'com.google.protobuf'
+
+	configurations {
+		[compileProtoPath, testCompileProtoPath].each {
+			it.extendsFrom(dependencyManagement)
+		}
+	}
+
+	dependencies {
+		api 'io.grpc:grpc-stub'
+
+		testImplementation 'io.grpc:grpc-protobuf'
+		testImplementation 'io.grpc:grpc-inprocess'
+		testImplementation "com.google.protobuf:protobuf-java:$protobufVersion"
+	}
+
+	protobuf {
+		protoc {
+			artifact = "com.google.protobuf:protoc:$protobufVersion"
+		}
+		plugins {
+			grpc {
+				artifact = "io.grpc:protoc-gen-grpc-java:$grpcVersion"
+			}
+		}
+		generateProtoTasks {
+			// Only generate for test source set, not main
+			ofSourceSet('test')*.plugins {
+				grpc {
+					option '@generated=omit'
+				}
+			}
+		}
+	}
+}
+
 project('spring-integration-hazelcast') {
     description = 'Spring Integration Hazelcast Support'
     dependencies {

spring-integration-grpc/src/main/java/org/springframework/integration/grpc/GrpcHeaders.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2025-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.grpc;
+
+/**
+ * Constants for gRPC-specific message headers.
+ *
+ * @author Artem Bilan
+ *
+ * @since 7.1
+ */
+public final class GrpcHeaders {
+
+	/**
+	 * The prefix for all gRPC-specific headers.
+	 */
+	public static final String PREFIX = "grpc_";
+
+	/**
+	 * The header containing the called gRPC service name.
+	 */
+	public static final String SERVICE = PREFIX + "service";
+
+	/**
+	 * The header containing the gRPC service method name.
+	 */
+	public static final String SERVICE_METHOD = PREFIX + "serviceMethod";
+
+	/**
+	 * The header containing the gRPC service method type.
+	 * One of the {@link io.grpc.MethodDescriptor.MethodType}
+	 */
+	public static final String METHOD_TYPE = PREFIX + "methodType";
+
+	/**
+	 * The header containing the gRPC service method schema descriptor.
+	 * A value from the {@link io.grpc.MethodDescriptor#getSchemaDescriptor()}
+	 */
+	public static final String SCHEMA_DESCRIPTOR = PREFIX + "schemaDescriptor";
+
+	private GrpcHeaders() {
+	}
+
+}

spring-integration-grpc/src/main/java/org/springframework/integration/grpc/inbound/GrpcInboundGateway.java

@@ -0,0 +1,250 @@
+/*
+ * Copyright 2025-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.grpc.inbound;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+
+import io.grpc.BindableService;
+import io.grpc.MethodDescriptor;
+import io.grpc.ServerServiceDefinition;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+import org.jspecify.annotations.Nullable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.core.log.LogMessage;
+import org.springframework.integration.gateway.DefaultMethodInvokingMethodInterceptor;
+import org.springframework.integration.gateway.MessagingGatewaySupport;
+import org.springframework.integration.grpc.GrpcHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ReflectionUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * The {@link MessagingGatewaySupport} implementation for gRPC {@link BindableService}.
+ * An instance of this class requires a {@link BindableService} class from the gRPC service definition.
+ * Only standard 'grpc' services are supported which implements a generated {@code AsyncService} interface.
+ * This gateway is a {@link BindableService} by itself to be registered with the gRPC server.
+ * An internal proxy is created to intercept gRPC method calls and convert them to Spring Integration messages.
+ * A reply from the downstream flow is produced back to the gRPC response payload.
+ * The request payload is a Proto message from gRPC request.
+ * The reply payload must be a Proto message for gRPC response.
+ * <p>
+ * This gateway supports all the gRPC {@link MethodDescriptor.MethodType} types.
+ * All the requests are produced to downstream flow in a reactive manner via {@link #sendAndReceiveMessageReactive(Object)}.
+ * The {@link MethodDescriptor.MethodType#UNARY} and {@link MethodDescriptor.MethodType#BIDI_STREAMING}
+ * are same from the downstream handling logic perspective.
+ * The {@link MethodDescriptor.MethodType#CLIENT_STREAMING} produces a {@link Flux} of gRPC request payloads.
+ * The {@link MethodDescriptor.MethodType#SERVER_STREAMING} reply can be a single entity or a {@link Flux} of them.
+ * <p>
+ * For convenience, the {@link GrpcHeaders} are populated into a request message.
+ * Such information can be used, for example, in downstream flow for routing.
+ *
+ * @author Artem Bilan
+ *
+ * @since 7.1
+ */
+public class GrpcInboundGateway extends MessagingGatewaySupport implements BindableService {
+
+	private final Class<? extends BindableService> grpcServiceClass;
+
+	@SuppressWarnings("NullAway.Init")
+	private Object asyncService;
+
+	@SuppressWarnings("NullAway.Init")
+	private ServerServiceDefinition serverServiceDefinition;
+
+	public GrpcInboundGateway(Class<? extends BindableService> grpcServiceClass) {
+		this.grpcServiceClass = grpcServiceClass;
+	}
+
+	@Override
+	protected void onInit() {
+		super.onInit();
+		Class<?>[] serviceInterfaces =
+				ClassUtils.getAllInterfacesForClass(this.grpcServiceClass, getApplicationContext().getClassLoader());
+
+		for (Class<?> serviceInterface : serviceInterfaces) {
+			if ("AsyncService".equals(serviceInterface.getSimpleName())) {
+				createServiceProxyAndServerDefinition(serviceInterface);
+				break;
+			}
+		}
+
+		Assert.state(this.asyncService != null,
+				"Only standard 'grpc' service are supported providing an 'AsyncService' contract.");
+	}
+
+	@SuppressWarnings("NullAway")
+	private void createServiceProxyAndServerDefinition(Class<?> serviceInterface) {
+		ProxyFactory proxyFactory = new ProxyFactory(serviceInterface, (MethodInterceptor) this::interceptGrpc);
+		proxyFactory.addAdvice(new DefaultMethodInvokingMethodInterceptor());
+		this.asyncService = proxyFactory.getProxy(getApplicationContext().getClassLoader());
+		Method bindServiceMethod =
+				ClassUtils.getStaticMethod(this.grpcServiceClass.getEnclosingClass(), "bindService", serviceInterface);
+
+		this.serverServiceDefinition =
+				(ServerServiceDefinition) ReflectionUtils.invokeMethod(bindServiceMethod, null, this.asyncService);
+	}
+
+	@Override
+	public ServerServiceDefinition bindService() {
+		return this.serverServiceDefinition;
+	}
+
+	@SuppressWarnings({"unchecked", "NullAway"})
+	private @Nullable Object interceptGrpc(MethodInvocation invocation) {
+		Object[] arguments = invocation.getArguments();
+
+		String fullMethodName =
+				this.serverServiceDefinition.getServiceDescriptor().getName() +
+						'/' +
+						StringUtils.capitalize(invocation.getMethod().getName());
+
+		MethodDescriptor<?, ?> serviceMethod =
+				this.serverServiceDefinition.getMethod(fullMethodName)
+						.getMethodDescriptor();
+
+		logger.debug(LogMessage.format("gRPC request for [%s] with arguments %s",
+				fullMethodName, Arrays.toString(arguments)));
+
+		switch (serviceMethod.getType()) {
+			case UNARY -> {
+				unary(serviceMethod, arguments[0], (StreamObserver<Object>) arguments[1]);
+				return null;
+			}
+			case SERVER_STREAMING -> {
+				serverStreaming(serviceMethod, arguments[0], (StreamObserver<Object>) arguments[1]);
+				return null;
+			}
+			case CLIENT_STREAMING -> {
+				return clientStreaming(serviceMethod, (StreamObserver<Object>) arguments[0]);
+			}
+			case BIDI_STREAMING -> {
+				return bidiStreaming(serviceMethod, (StreamObserver<Object>) arguments[0]);
+			}
+			default -> throw new IllegalStateException("Unknown gRPC method type: " + serviceMethod.getType());
+		}
+	}
+
+	private void unary(MethodDescriptor<?, ?> methodDescriptor, Object requestPayload,
+			StreamObserver<Object> responseObserver) {
+
+		sendRequestAndProduceReply(methodDescriptor, requestPayload)
+				.subscribe(responseObserver::onNext,
+						t -> responseObserver.onError(toGrpcStatusException(t)),
+						responseObserver::onCompleted);
+	}
+
+	private void serverStreaming(MethodDescriptor<?, ?> methodDescriptor, Object requestPayload,
+			StreamObserver<Object> responseObserver) {
+
+		sendRequestAndProduceReply(methodDescriptor, requestPayload)
+				.flatMapMany(payload -> payload instanceof Flux<?> flux ? flux : Flux.just(payload))
+				.subscribe(responseObserver::onNext,
+						t -> responseObserver.onError(toGrpcStatusException(t)),
+						responseObserver::onCompleted);
+	}
+
+	private StreamObserver<?> clientStreaming(MethodDescriptor<?, ?> methodDescriptor,
+			StreamObserver<Object> responseObserver) {
+
+		Sinks.Many<Object> requestPayload = Sinks.many().unicast().onBackpressureBuffer();
+
+		return new StreamObserver<>() {
+
+			@Override
+			public void onNext(Object value) {
+				requestPayload.tryEmitNext(value);
+			}
+
+			@Override
+			public void onError(Throwable t) {
+				throw toGrpcStatusException(t, "gRPC request [" + methodDescriptor.getFullMethodName() + "] has failed");
+			}
+
+			@Override
+			public void onCompleted() {
+				requestPayload.tryEmitComplete();
+				sendRequestAndProduceReply(methodDescriptor, requestPayload.asFlux())
+						.subscribe(responseObserver::onNext,
+								t -> responseObserver.onError(toGrpcStatusException(t)),
+								responseObserver::onCompleted);
+			}
+
+		};
+	}
+
+	private StreamObserver<?> bidiStreaming(MethodDescriptor<?, ?> methodDescriptor,
+			StreamObserver<Object> responseObserver) {
+
+		return new StreamObserver<>() {
+
+			@Override
+			public void onNext(Object value) {
+				sendRequestAndProduceReply(methodDescriptor, value)
+						.subscribe(responseObserver::onNext,
+								t -> responseObserver.onError(toGrpcStatusException(t)));
+			}
+
+			@Override
+			public void onError(Throwable t) {
+				throw toGrpcStatusException(t, "gRPC request [" + methodDescriptor.getFullMethodName() + "] has failed");
+			}
+
+			@Override
+			public void onCompleted() {
+				responseObserver.onCompleted();
+			}
+
+		};
+	}
+
+	private Mono<?> sendRequestAndProduceReply(MethodDescriptor<?, ?> serviceMethod, Object requestPayload) {
+		Message<?> requestMessage =
+				getMessageBuilderFactory()
+						.withPayload(requestPayload)
+						.setHeader(GrpcHeaders.SERVICE, serviceMethod.getServiceName())
+						.setHeader(GrpcHeaders.SERVICE_METHOD, serviceMethod.getBareMethodName())
+						.setHeader(GrpcHeaders.METHOD_TYPE, serviceMethod.getType())
+						.setHeader(GrpcHeaders.SCHEMA_DESCRIPTOR, serviceMethod.getSchemaDescriptor())
+						.build();
+
+		return sendAndReceiveMessageReactive(requestMessage)
+				.map(Message::getPayload);
+	}
+
+	private static StatusRuntimeException toGrpcStatusException(Throwable throwable) {
+		return toGrpcStatusException(throwable, throwable.getMessage());
+	}
+
+	private static StatusRuntimeException toGrpcStatusException(Throwable throwable, @Nullable String description) {
+		return Status.fromThrowable(throwable)
+				.withDescription(description)
+				.asRuntimeException();
+	}
+
+}

spring-integration-grpc/src/main/java/org/springframework/integration/grpc/inbound/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Components for server-side gRPC support.
+ */
+@org.jspecify.annotations.NullMarked
+package org.springframework.integration.grpc.inbound;

spring-integration-grpc/src/main/java/org/springframework/integration/grpc/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Base package for gRPC support.
+ */
+@org.jspecify.annotations.NullMarked
+package org.springframework.integration.grpc;