From ef9c2462c4b948d15bc28c3c1713ef7d58e5c5c2 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 11 Mar 2025 16:49:34 +0100 Subject: [PATCH] GH-1245 Add initial proxy support This feature woudl allow request to be proxied to additional targets such as grpc, rsocket etc. Resolves #1245 --- .../grpc/GrpcFunctionAutoConfiguration.java | 60 +++++++++++++++++++ .../grpc/GrpcServerMessageHandler.java | 31 +--------- ...ot.autoconfigure.AutoConfiguration.imports | 1 + .../function/grpc/GrpcInteractionTests.java | 3 +- .../function/context/FunctionProperties.java | 5 ++ 5 files changed, 69 insertions(+), 31 deletions(-) create mode 100644 spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcFunctionAutoConfiguration.java diff --git a/spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcFunctionAutoConfiguration.java b/spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcFunctionAutoConfiguration.java new file mode 100644 index 000000000..96e4fa874 --- /dev/null +++ b/spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcFunctionAutoConfiguration.java @@ -0,0 +1,60 @@ +/* + * Copyright 2021-2021 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.cloud.function.grpc; + +import java.util.function.Function; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.function.context.FunctionProperties; +import org.springframework.cloud.function.context.MessageRoutingCallback; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.Message; + +/** + * + * @author Oleg Zhurakousky + * @since 3.2 + */ +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties(FunctionGrpcProperties.class) +public class GrpcFunctionAutoConfiguration { + + public static String GRPC_INVOKER_FUNCTION = "grpcInvokerFunction"; + + public static String GRPC = "grpc"; + + @Bean + public Function, Message> grpcInvokerFunction() { + return input -> { + return GrpcUtils.requestReply(input); + }; + } + + @Bean + public MessageRoutingCallback routingCallback() { + return new MessageRoutingCallback() { + public String routingResult(Message message) { + if (message.getHeaders().containsKey(FunctionProperties.PROXY) + && message.getHeaders().get(FunctionProperties.PROXY).equals(GRPC)) { + return GRPC_INVOKER_FUNCTION; + } + return null; + } + }; + } +} diff --git a/spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServerMessageHandler.java b/spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServerMessageHandler.java index f04405846..74207d4d2 100644 --- a/spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServerMessageHandler.java +++ b/spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServerMessageHandler.java @@ -32,36 +32,11 @@ package org.springframework.cloud.function.grpc; -import java.util.Map; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -// -import io.grpc.Status; -import io.grpc.stub.ServerCallStreamObserver; -import io.grpc.stub.StreamObserver; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.reactivestreams.Publisher; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Sinks; -import reactor.core.publisher.Sinks.Many; -// -import org.springframework.cloud.function.context.FunctionCatalog; -import org.springframework.cloud.function.context.FunctionProperties; -import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.function.grpc.MessagingServiceGrpc.MessagingServiceImplBase; -import org.springframework.context.SmartLifecycle; -import org.springframework.messaging.Message; -import org.springframework.util.Assert; -import org.springframework.util.CollectionUtils; import com.google.protobuf.GeneratedMessageV3; -// -//import com.google.protobuf.GeneratedMessage; +import io.grpc.stub.StreamObserver; /** * @@ -72,12 +47,8 @@ import com.google.protobuf.GeneratedMessageV3; @SuppressWarnings("rawtypes") public class GrpcServerMessageHandler extends MessagingServiceImplBase { - private Log logger = LogFactory.getLog(GrpcServerMessageHandler.class); - private final MessageHandlingHelper helper; - private boolean running; - public GrpcServerMessageHandler(MessageHandlingHelper helper) { this.helper = helper; } diff --git a/spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index 1aee89ea4..258fc4773 100644 --- a/spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1 +1,2 @@ org.springframework.cloud.function.grpc.GrpcAutoConfiguration +org.springframework.cloud.function.grpc.GrpcFunctionAutoConfiguration diff --git a/spring-cloud-function-adapters/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java b/spring-cloud-function-adapters/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java index be9411558..4d429c0ca 100644 --- a/spring-cloud-function-adapters/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java +++ b/spring-cloud-function-adapters/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java @@ -50,7 +50,7 @@ import static org.junit.jupiter.api.Assertions.fail; * @author Oleg Zhurakousky * @author Chris Bono */ -@Disabled +//@Disabled public class GrpcInteractionTests { @BeforeEach @@ -211,6 +211,7 @@ public class GrpcInteractionTests { } @Test + @Disabled public void testClientStreaming() throws Exception { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( SampleConfiguration.class).web(WebApplicationType.NONE).run( diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionProperties.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionProperties.java index efad20fb4..a9e05dbb3 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionProperties.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionProperties.java @@ -60,6 +60,11 @@ public class FunctionProperties implements EnvironmentAware, ApplicationContextA */ public final static String FUNCTION_DEFINITION = PREFIX + ".definition"; + /** + * Key for the proxy name. + */ + public final static String PROXY = "proxy"; + /** * Definition of the function to be used. This could be function name (e.g., 'myFunction') * or function composition definition (e.g., 'myFunction|yourFunction')