diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/pom.xml b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/pom.xml index de98778b3..91a204c04 100644 --- a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/pom.xml +++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/pom.xml @@ -53,5 +53,11 @@ ${google.cloud.functions.invoker.version} test + + com.github.stefanbirkner + system-rules + 1.19.0 + test + diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcloud/Context.java b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcloud/Context.java new file mode 100644 index 000000000..142ef82b7 --- /dev/null +++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcloud/Context.java @@ -0,0 +1,69 @@ +/* + * Copyright 2020-2020 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.adapter.gcloud; + +/** + * An immutable implementation of the Google Cloud Function + * {@link com.google.cloud.functions.Context} interface. + * + * @author Mike Eltsufin + * @since 3.0.5 + */ +public class Context implements com.google.cloud.functions.Context { + + private String eventId; + + private String timestamp; + + private String eventType; + + private String resource; + + public Context() { + } + + public Context(String eventId, String timestamp, String eventType, String resource) { + this.eventId = eventId; + this.timestamp = timestamp; + this.eventType = eventType; + this.resource = resource; + } + + @Override + public String eventId() { + return this.eventId; + + } + + @Override + public String timestamp() { + return this.timestamp; + + } + + @Override + public String eventType() { + return this.eventType; + + } + + @Override + public String resource() { + return this.resource; + } + +} diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcloud/FunctionInvoker.java b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcloud/FunctionInvoker.java index adeae7907..97d8646ba 100644 --- a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcloud/FunctionInvoker.java +++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcloud/FunctionInvoker.java @@ -21,9 +21,14 @@ import java.nio.charset.StandardCharsets; import java.util.Map.Entry; import java.util.function.Function; +import com.google.cloud.functions.Context; 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 com.google.gson.Gson; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.cloud.function.context.AbstractSpringFunctionAdapterInitializer; import org.springframework.messaging.Message; @@ -32,17 +37,23 @@ import org.springframework.util.Assert; import org.springframework.util.MimeTypeUtils; /** - * Implementation of {@link HttpFunction} for Google Cloud Function (GCF). - * This is the Spring Cloud Function adapter for GCF HTTP function. + * Implementation of {@link HttpFunction} and {@link RawBackgroundFunction} for Google + * Cloud Function (GCF). This is the Spring Cloud Function adapter for GCF HTTP and Raw + * Background function. * * @author Dmitry Solomakha * @author Mike Eltsufin * @author Oleg Zhurakousky - * * @since 3.0.4 */ -public class FunctionInvoker - extends AbstractSpringFunctionAdapterInitializer implements HttpFunction { +public class FunctionInvoker extends AbstractSpringFunctionAdapterInitializer + implements HttpFunction, RawBackgroundFunction { + + private static final Log log = LogFactory.getLog(FunctionInvoker.class); + + private String functionName = ""; + + private static final Gson gson = new Gson(); public FunctionInvoker() { super(); @@ -55,29 +66,33 @@ public class FunctionInvoker } private void init() { + if (System.getenv().containsKey("spring.cloud.function.definition")) { + this.functionName = System.getenv("spring.cloud.function.definition"); + } System.setProperty("spring.http.converters.preferred-json-mapper", "gson"); Thread.currentThread() // TODO: remove after upgrading to 1.0.0-alpha-2-rc5 .setContextClassLoader(FunctionInvoker.class.getClassLoader()); initialize(null); } + private Function, Message> lookupFunction() { + Function, Message> function = this.catalog.lookup(functionName, + MimeTypeUtils.APPLICATION_JSON.toString()); + Assert.notNull(function, "'function' with name '" + functionName + "' must not be null"); + return function; + } + /** - * The implementation of a GCF {@link HttpFunction} that will be used as the entry point from GCF. + * The implementation of a GCF {@link HttpFunction} that will be used as the entry + * point from GCF. */ @Override public void service(HttpRequest httpRequest, HttpResponse httpResponse) throws Exception { try { - String functionName = System.getenv().containsKey("spring.cloud.function.definition") - ? System.getenv("spring.cloud.function.definition") : ""; + Function, Message> function = lookupFunction(); - Function, Message> function = - this.catalog.lookup(functionName, MimeTypeUtils.APPLICATION_JSON.toString()); - Assert.notNull(function, "'function' with name '" + functionName + "' must not be null"); - - Message message = getInputType() == Void.class - ? null : MessageBuilder.withPayload(httpRequest.getReader()) - .copyHeaders(httpRequest.getHeaders()) - .build(); + Message message = getInputType() == Void.class ? null + : MessageBuilder.withPayload(httpRequest.getReader()).copyHeaders(httpRequest.getHeaders()).build(); Message result = function.apply(message); if (result != null) { @@ -91,4 +106,40 @@ public class FunctionInvoker httpResponse.getWriter().close(); } } + + /** + * The implementation of a GCF {@link RawBackgroundFunction} that will be used as the + * entry point from GCF. + * @param json the payload. + * @param context event context. + * @since 3.0.5 + */ + @Override + public void accept(String json, Context context) { + + Function, Message> function = lookupFunction(); + Message message; + + if ("google.pubsub.topic.publish".equals(context.eventType()) && getInputType() != PubSubMessage.class) { + // If the input type is not PubSubMessage, use the PubSubMessage.data field as + // payload, and put the full PubSubMessage and Context into message headers. + + PubSubMessage pubSubMessage = gson.fromJson(json, PubSubMessage.class); + + message = getInputType() == Void.class ? null : MessageBuilder.withPayload(pubSubMessage.getData()) + .setHeader("gcf_context", context).setHeader("gcf_message", pubSubMessage).build(); + } + else { + message = getInputType() == Void.class ? null + : MessageBuilder.withPayload(json).setHeader("context", context).build(); + } + + Message result = function.apply(message); + + if (result != null) { + log.info("Dropping background function result: " + new String(result.getPayload())); + } + + } + } diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcloud/PubSubMessage.java b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcloud/PubSubMessage.java new file mode 100644 index 000000000..120979875 --- /dev/null +++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/main/java/org/springframework/cloud/function/adapter/gcloud/PubSubMessage.java @@ -0,0 +1,70 @@ +/* + * Copyright 2020-2020 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.adapter.gcloud; + +import java.util.Map; + +/** + * A class that can be mapped to the GCF Pub/Sub Message event type. This is for use in + * the background functions. + * + * @author Mike Eltsufin + * @since 3.0.5 + */ +public class PubSubMessage { + + private String data; + + private Map attributes; + + private String messageId; + + private String publishTime; + + public String getData() { + return data; + } + + public void setData(String data) { + this.data = data; + } + + public Map getAttributes() { + return attributes; + } + + public void setAttributes(Map attributes) { + this.attributes = attributes; + } + + public String getMessageId() { + return messageId; + } + + public void setMessageId(String messageId) { + this.messageId = messageId; + } + + public String getPublishTime() { + return publishTime; + } + + public void setPublishTime(String publishTime) { + this.publishTime = publishTime; + } + +} diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcloud/FunctionInvokerTests.java b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcloud/FunctionInvokerHttpTests.java similarity index 83% rename from spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcloud/FunctionInvokerTests.java rename to spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcloud/FunctionInvokerHttpTests.java index 3ea8ed9fc..3ffdbe867 100644 --- a/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcloud/FunctionInvokerTests.java +++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/src/test/java/org/springframework/cloud/function/adapter/gcloud/FunctionInvokerHttpTests.java @@ -46,33 +46,33 @@ import static org.mockito.Mockito.when; * @author Dmitry Solomakha * @author Mike Eltsufin */ -public class FunctionInvokerTests { +public class FunctionInvokerHttpTests { private static final Gson gson = new Gson(); @Test public void testHelloWorldSupplier() throws Exception { - testFunction(HelloWorldSupplier.class, null, "Hello World!"); + testHttpFunction(HelloWorldSupplier.class, null, "Hello World!"); } @Test public void testJsonInputFunction() throws Exception { - testFunction(JsonInputFunction.class, new IncomingRequest("hello"), - "Thank you for sending the message: hello"); + testHttpFunction(JsonInputFunction.class, new IncomingRequest("hello"), + "Thank you for sending the message: hello"); } @Test public void testJsonInputOutputFunction() throws Exception { - testFunction(JsonInputOutputFunction.class, new IncomingRequest("hello"), - new OutgoingResponse("Thank you for sending the message: hello")); + testHttpFunction(JsonInputOutputFunction.class, new IncomingRequest("hello"), + new OutgoingResponse("Thank you for sending the message: hello")); } @Test - public void testJsonInputConsumer() throws Exception { - testFunction(JsonInputConsumer.class, new IncomingRequest("hello"), null); + public void testJsonInputConsumer_Background() throws Exception { + testHttpFunction(JsonInputConsumer.class, new IncomingRequest("hello"), null); } - private void testFunction(Class configurationClass, I input, O expectedOutput) throws Exception { + private void testHttpFunction(Class configurationClass, I input, O expectedOutput) throws Exception { try (FunctionInvoker handler = new FunctionInvoker(configurationClass);) { HttpRequest request = Mockito.mock(HttpRequest.class); @@ -95,55 +95,69 @@ public class FunctionInvokerTests { @Configuration @Import({ ContextFunctionCatalogAutoConfiguration.class }) protected static class HelloWorldSupplier { + @Bean public Supplier supplier() { return () -> "Hello World!"; } + } @Configuration @Import({ ContextFunctionCatalogAutoConfiguration.class }) protected static class JsonInputFunction { + @Bean public Function function() { return (in) -> "Thank you for sending the message: " + in.message; } + } @Configuration @Import({ ContextFunctionCatalogAutoConfiguration.class }) protected static class JsonInputOutputFunction { + @Bean public Function> function() { return (in) -> { - return MessageBuilder.withPayload(new OutgoingResponse("Thank you for sending the message: " + in.message)) + return MessageBuilder + .withPayload(new OutgoingResponse("Thank you for sending the message: " + in.message)) .setHeader("foo", "bar").build(); }; } + } @Configuration @Import({ ContextFunctionCatalogAutoConfiguration.class }) protected static class JsonInputConsumer { + @Bean public Consumer function() { return (in) -> System.out.println("Thank you for sending the message: " + in.message); } + } private static class IncomingRequest { + String message; IncomingRequest(String message) { this.message = message; } + } private static class OutgoingResponse { + String message; OutgoingResponse(String message) { this.message = message; } + } + }