From 4b0636f48f623c893f62cc371fd222240bd2439f Mon Sep 17 00:00:00 2001 From: geezylucas Date: Mon, 14 Apr 2025 23:58:16 -0600 Subject: [PATCH] GH-1259 feat: add reactor publisher support Resolves #1259 Resolves #1264 Signed-off-by: geezylucas --- .../function/adapter/gcp/FunctionInvoker.java | 155 +++++++++++++----- .../gcp/FunctionInvokerBackgroundTests.java | 38 +++++ .../adapter/gcp/FunctionInvokerHttpTests.java | 56 +++++++ 3 files changed, 211 insertions(+), 38 deletions(-) diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcp/FunctionInvoker.java b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcp/FunctionInvoker.java index 7ccc4de93..72b819557 100644 --- a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcp/FunctionInvoker.java +++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcp/FunctionInvoker.java @@ -17,8 +17,11 @@ package org.springframework.cloud.function.adapter.gcp; import java.io.BufferedReader; +import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.Collection; +import java.util.List; import java.util.Map.Entry; import java.util.function.Consumer; import java.util.function.Function; @@ -30,9 +33,12 @@ import com.google.cloud.functions.HttpFunction; import com.google.cloud.functions.HttpRequest; import com.google.cloud.functions.HttpResponse; import com.google.cloud.functions.RawBackgroundFunction; + +import reactor.core.publisher.Flux; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - +import org.reactivestreams.Publisher; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; import org.springframework.cloud.function.context.FunctionCatalog; @@ -40,6 +46,7 @@ import org.springframework.cloud.function.context.FunctionalSpringApplication; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration; import org.springframework.cloud.function.context.config.RoutingFunction; +import org.springframework.cloud.function.json.JsonMapper; import org.springframework.cloud.function.utils.FunctionClassUtils; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.messaging.Message; @@ -76,6 +83,8 @@ public class FunctionInvoker implements HttpFunction, RawBackgroundFunction { private ConfigurableApplicationContext context; + private JsonMapper jsonMapper; + public FunctionInvoker() { this(FunctionClassUtils.getStartClass()); } @@ -90,18 +99,19 @@ public class FunctionInvoker implements HttpFunction, RawBackgroundFunction { System.setProperty(ContextFunctionCatalogAutoConfiguration.JSON_MAPPER_PROPERTY, "gson"); } Thread.currentThread() // TODO: remove after upgrading to 1.0.0-alpha-2-rc5 - .setContextClassLoader(FunctionInvoker.class.getClassLoader()); + .setContextClassLoader(FunctionInvoker.class.getClassLoader()); log.info("Initializing: " + configurationClass); SpringApplication springApplication = springApplication(configurationClass); this.context = springApplication.run(); this.catalog = this.context.getBean(FunctionCatalog.class); + this.jsonMapper = this.context.getBean(JsonMapper.class); initFunctionConsumerOrSupplierFromCatalog(); } private Function, Message> lookupFunction() { Function, Message> function = this.catalog.lookup(functionName, - MimeTypeUtils.APPLICATION_JSON.toString()); + MimeTypeUtils.APPLICATION_JSON.toString()); Assert.notNull(function, "'function' with name '" + functionName + "' must not be null"); return function; } @@ -114,43 +124,22 @@ public class FunctionInvoker implements HttpFunction, RawBackgroundFunction { public void service(HttpRequest httpRequest, HttpResponse httpResponse) throws Exception { Function, Message> function = lookupFunction(); - Message message = this.functionWrapped.getInputType() == Void.class || this.functionWrapped.getInputType() == null ? null - : MessageBuilder.withPayload(httpRequest.getReader()).copyHeaders(httpRequest.getHeaders()).build(); + Message message = this.functionWrapped.getInputType() == Void.class + || this.functionWrapped.getInputType() == null ? null + : MessageBuilder.withPayload(httpRequest.getReader()).copyHeaders(httpRequest.getHeaders()) + .build(); - Message result = function.apply(message); + Object resultObject = function.apply(message); - if (result != null) { - MessageHeaders headers = result.getHeaders(); - if (result.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) { - httpResponse.setContentType(result.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()); - } - else if (result.getHeaders().containsKey("Content-Type")) { - httpResponse.setContentType(result.getHeaders().get("Content-Type").toString()); - } - else { - httpRequest.getContentType().ifPresent(contentType -> httpResponse.setContentType(contentType)); - } - String content = result.getPayload() instanceof String strPayload ? strPayload : new String((byte[]) result.getPayload(), StandardCharsets.UTF_8); - httpResponse.getWriter().write(content); - for (Entry header : headers.entrySet()) { - Object values = header.getValue(); - if (values instanceof Collection) { - String headerValue = ((Collection) values).stream().map(item -> item.toString()).collect(Collectors.joining(",")); - httpResponse.appendHeader(header.getKey(), headerValue); - } - else { - httpResponse.appendHeader(header.getKey(), header.getValue().toString()); - } + if (resultObject != null) { + Message result = null; + if (resultObject instanceof Publisher) { + result = getResultFromPublisher(resultObject); + } else { + result = (Message) resultObject; } - if (headers.containsKey(HTTP_STATUS_CODE)) { - if (headers.get(HTTP_STATUS_CODE) instanceof Integer) { - httpResponse.setStatusCode((int) headers.get(HTTP_STATUS_CODE)); - } - else { - log.warn("The statusCode should be an Integer value"); - } - } + buildHttpResponse(httpRequest, httpResponse, result); } } @@ -162,19 +151,109 @@ public class FunctionInvoker implements HttpFunction, RawBackgroundFunction { * @param context event context. * @since 3.0.5 */ + @SuppressWarnings("unchecked") @Override public void accept(String json, Context context) { Function, Message> function = lookupFunction(); Message message = this.functionWrapped.getInputType() == Void.class ? null - : MessageBuilder.withPayload(json).setHeader("gcf_context", context).build(); + : MessageBuilder.withPayload(json).setHeader("gcf_context", context).build(); - Message result = function.apply(message); + Object resultObject = function.apply(message); + + Message result = null; + if (resultObject instanceof Publisher) { + result = getResultFromPublisher(resultObject); + } else { + result = (Message) resultObject; + } if (result != null) { log.info("Dropping background function result: " + new String(result.getPayload())); } } + /** + * This method build the http response from service + * + * @throws IOException + */ + private void buildHttpResponse(HttpRequest httpRequest, HttpResponse httpResponse, Message result) + throws IOException { + MessageHeaders headers = result.getHeaders(); + if (result.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) { + httpResponse.setContentType(result.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()); + } else if (result.getHeaders().containsKey("Content-Type")) { + httpResponse.setContentType(result.getHeaders().get("Content-Type").toString()); + } else { + httpRequest.getContentType().ifPresent(contentType -> httpResponse.setContentType(contentType)); + } + String content = result.getPayload() instanceof String strPayload ? strPayload + : new String((byte[]) result.getPayload(), StandardCharsets.UTF_8); + httpResponse.getWriter().write(content); + for (Entry header : headers.entrySet()) { + Object values = header.getValue(); + if (values instanceof Collection) { + String headerValue = ((Collection) values).stream().map(item -> item.toString()) + .collect(Collectors.joining(",")); + httpResponse.appendHeader(header.getKey(), headerValue); + } else { + httpResponse.appendHeader(header.getKey(), header.getValue().toString()); + } + } + + if (headers.containsKey(HTTP_STATUS_CODE)) { + if (headers.get(HTTP_STATUS_CODE) instanceof Integer) { + httpResponse.setStatusCode((int) headers.get(HTTP_STATUS_CODE)); + } else { + log.warn("The statusCode should be an Integer value"); + } + } + } + + /** + * This methd get the result from reactor's publisher + * + * For reference: https://github.com/spring-cloud/spring-cloud-function/blob/main/spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/AWSLambdaUtils.java + */ + private Message getResultFromPublisher(Object resultObject) { + List results = new ArrayList<>(); + Message lastMessage = null; + for (Object item : Flux.from((Publisher) resultObject).toIterable()) { + log.info("Response value: " + item); + if (item instanceof Message messageItem) { + results.add(convertFromJsonIfNecessary(messageItem.getPayload())); + lastMessage = messageItem; + } else { + results.add(convertFromJsonIfNecessary(item)); + } + } + + byte[] resultsPayload; + if (results.size() == 1) { + resultsPayload = jsonMapper.toJson(results.get(0)); + } else if (results.size() > 1) { + resultsPayload = jsonMapper.toJson(results); + } else { + resultsPayload = null; + } + + Assert.notNull(resultsPayload, "Couldn't resolve payload result"); + + MessageBuilder messageBuilder = MessageBuilder.withPayload(resultsPayload); + if (lastMessage != null) { + messageBuilder.copyHeaders(lastMessage.getHeaders()); + } + return messageBuilder.build(); + } + + private Object convertFromJsonIfNecessary(Object value) { + if (JsonMapper.isJsonString(value)) { + return jsonMapper.fromJson(value, Object.class); + } + + return value; + } + private void initFunctionConsumerOrSupplierFromCatalog() { String name = resolveName(Function.class); this.functionWrapped = this.catalog.lookup(Function.class, name); diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcp/FunctionInvokerBackgroundTests.java b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcp/FunctionInvokerBackgroundTests.java index 29951e594..901a21c88 100644 --- a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcp/FunctionInvokerBackgroundTests.java +++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcp/FunctionInvokerBackgroundTests.java @@ -22,6 +22,10 @@ import java.util.function.Supplier; import com.github.blindpirate.extensions.CaptureSystemOutput; import com.google.gson.Gson; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + import org.hamcrest.Matchers; import org.junit.jupiter.api.Test; @@ -58,6 +62,18 @@ public class FunctionInvokerBackgroundTests { "Thank you for sending the message: hello", null, null); } + @Test + public void testJsonInputFunction_BackgroundMono(CaptureSystemOutput.OutputCapture outputCapture) { + testBackgroundFunction(outputCapture, JsonInputFunctionMono.class, new IncomingRequest("hello"), + "Thank you for sending the message: hello", null, null); + } + + @Test + public void testJsonInputFunction_BackgroundFlux(CaptureSystemOutput.OutputCapture outputCapture) { + testBackgroundFunction(outputCapture, JsonInputFunctionFlux.class, new IncomingRequest("hello"), + "Thank you for sending the message: hello", null, null); + } + @Test public void testJsonInputOutputFunction_Background(CaptureSystemOutput.OutputCapture outputCapture) { testBackgroundFunction(outputCapture, JsonInputOutputFunction.class, new IncomingRequest("hello"), @@ -149,6 +165,28 @@ public class FunctionInvokerBackgroundTests { } + @Configuration + @Import({ ContextFunctionCatalogAutoConfiguration.class }) + protected static class JsonInputFunctionMono { + + @Bean + public Function, Mono> function() { + return (in) -> in.map(o -> "Thank you for sending the message: " + o.message); + } + + } + + @Configuration + @Import({ ContextFunctionCatalogAutoConfiguration.class }) + protected static class JsonInputFunctionFlux { + + @Bean + public Function, Flux> function() { + return (in) -> in.map(o -> "Thank you for sending the message: " + o.message); + } + + } + @Configuration @Import({ ContextFunctionCatalogAutoConfiguration.class }) protected static class JsonInputOutputFunction { diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcp/FunctionInvokerHttpTests.java b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcp/FunctionInvokerHttpTests.java index d02d322eb..a1b6dc41b 100644 --- a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcp/FunctionInvokerHttpTests.java +++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcp/FunctionInvokerHttpTests.java @@ -30,6 +30,10 @@ import java.util.function.Supplier; import com.google.cloud.functions.HttpRequest; import com.google.cloud.functions.HttpResponse; import com.google.gson.Gson; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -90,6 +94,36 @@ public class FunctionInvokerHttpTests { } + @Test + public void testJsonInputFunctionMono() throws Exception { + + FunctionInvoker handler = new FunctionInvoker(JsonInputFunctionMono.class); + + String expectedOutput = "Thank you for sending the message: hello"; + IncomingRequest input = new IncomingRequest("hello"); + + when(request.getReader()).thenReturn(new BufferedReader(new StringReader(gson.toJson(input)))); + handler.service(request, response); + bufferedWriter.close(); + + + assertThat(writer.toString()).isEqualTo(gson.toJson(expectedOutput)); + } + + @Test + public void testJsonInputFunctionFlux() throws Exception { + + FunctionInvoker handler = new FunctionInvoker(JsonInputFunctionFlux.class); + + String expectedOutput = "hello!!!"; + + when(request.getReader()).thenReturn(new BufferedReader(new StringReader("hello"))); + handler.service(request, response); + bufferedWriter.close(); + + + assertThat(writer.toString()).isEqualTo(gson.toJson(expectedOutput)); + } @Test public void testJsonInputFunction() throws Exception { @@ -238,6 +272,28 @@ public class FunctionInvokerHttpTests { } + @Configuration + @Import({ ContextFunctionCatalogAutoConfiguration.class }) + protected static class JsonInputFunctionMono { + + @Bean + public Function, Mono> function() { + return (in) -> in.map(o -> "Thank you for sending the message: " + o.message); + } + + } + + @Configuration + @Import({ ContextFunctionCatalogAutoConfiguration.class }) + protected static class JsonInputFunctionFlux { + + @Bean + public Function, Flux> function() { + return (in) -> in.map(word -> word + "!!!"); + } + + } + @Configuration @Import({ ContextFunctionCatalogAutoConfiguration.class }) protected static class JsonInputOutputFunction {