GH-1245 Add initial proxy support

This feature woudl allow request to be proxied to additional targets such as grpc, rsocket etc.

Resolves #1245

Author: olegz

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<byte[]>, 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;
+			}
+		};
+	}
+}

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 @@
 @SuppressWarnings("rawtypes")
 public class GrpcServerMessageHandler extends MessagingServiceImplBase {
 
-	private Log logger = LogFactory.getLog(GrpcServerMessageHandler.class);
-
 	private final MessageHandlingHelper helper;
 
-	private boolean running;
-
 	public GrpcServerMessageHandler(MessageHandlingHelper<GeneratedMessageV3> helper) {
 		this.helper = helper;
 	}

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

spring-cloud-function-adapters/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java

@@ -50,7 +50,7 @@
  * @author Oleg Zhurakousky
  * @author Chris Bono
  */
-@Disabled
+//@Disabled
 public class GrpcInteractionTests {
 
 	@BeforeEach
@@ -211,6 +211,7 @@ public void testBidirectionalStreamWithReactiveFunction() throws Exception {
 	}
 
 	@Test
+	@Disabled
 	public void testClientStreaming() throws Exception {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
 				SampleConfiguration.class).web(WebApplicationType.NONE).run(

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')