Implement GCP Raw Background Function support

Add Pub/Sub payload vs. PubSumMessage support

Fixes: #491
Resolves #492

spring-cloud-function-adapters/spring-cloud-function-adapter-gcp/pom.xml

@@ -53,5 +53,11 @@
 			<version>${google.cloud.functions.invoker.version}</version>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>com.github.stefanbirkner</groupId>
+			<artifactId>system-rules</artifactId>
+			<version>1.19.0</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 </project>

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;
+	}
+
+}

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<HttpRequest> implements HttpFunction {
+public class FunctionInvoker extends AbstractSpringFunctionAdapterInitializer<HttpRequest>
+		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 <I> Function<Message<I>, Message<byte[]>> lookupFunction() {
+		Function<Message<I>, Message<byte[]>> 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<BufferedReader>, Message<byte[]>> function = lookupFunction();
 
-			Function<Message<BufferedReader>, Message<byte[]>> function =
-					this.catalog.lookup(functionName, MimeTypeUtils.APPLICATION_JSON.toString());
-			Assert.notNull(function, "'function' with name '" + functionName + "' must not be null");
-
-			Message<BufferedReader> message = getInputType() == Void.class
-					? null : MessageBuilder.withPayload(httpRequest.getReader())
-								.copyHeaders(httpRequest.getHeaders())
-								.build();
+			Message<BufferedReader> message = getInputType() == Void.class ? null
+					: MessageBuilder.withPayload(httpRequest.getReader()).copyHeaders(httpRequest.getHeaders()).build();
 			Message<byte[]> 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<String>, Message<byte[]>> function = lookupFunction();
+		Message<String> 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<byte[]> result = function.apply(message);
+
+		if (result != null) {
+			log.info("Dropping background function result: " + new String(result.getPayload()));
+		}
+
+	}
+
 }

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<String, String> attributes;
+
+	private String messageId;
+
+	private String publishTime;
+
+	public String getData() {
+		return data;
+	}
+
+	public void setData(String data) {
+		this.data = data;
+	}
+
+	public Map<String, String> getAttributes() {
+		return attributes;
+	}
+
+	public void setAttributes(Map<String, String> 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;
+	}
+
+}

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 <I, O> void testFunction(Class<?> configurationClass, I input, O expectedOutput) throws Exception {
+	private <I, O> 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<String> supplier() {
 			return () -> "Hello World!";
 		}
+
 	}
 
 	@Configuration
 	@Import({ ContextFunctionCatalogAutoConfiguration.class })
 	protected static class JsonInputFunction {
+
 		@Bean
 		public Function<IncomingRequest, String> function() {
 			return (in) -> "Thank you for sending the message: " + in.message;
 		}
+
 	}
 
 	@Configuration
 	@Import({ ContextFunctionCatalogAutoConfiguration.class })
 	protected static class JsonInputOutputFunction {
+
 		@Bean
 		public Function<IncomingRequest, Message<OutgoingResponse>> 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<IncomingRequest> 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;
 		}
+
 	}
+
 }