unblock local development with the pubsub emulator

Currently, the functions frameworks are dependent on some private
dataplane event marshalling logic in order to correctly pass PubSub
events to background functions. This commit implements the same
marshalling logic in the FF in order to enable local development using
the PubSub emulator.

We have already made this change in other languages:
GoogleCloudPlatform/functions-framework-nodejs#272
GoogleCloudPlatform/functions-framework-ruby#100
GoogleCloudPlatform/functions-framework-python#121
GoogleCloudPlatform/functions-framework-go#70

Author: matthewrobertson

invoker/core/src/main/java/com/google/cloud/functions/invoker/Event.java

@@ -21,6 +21,8 @@
 import com.google.gson.JsonObject;
 import com.google.gson.JsonParseException;
 import java.lang.reflect.Type;
+import java.time.OffsetDateTime;
+import java.time.format.DateTimeFormatter;
 
 /**
  * Represents an event that should be handled by a background function. This is an internal format
@@ -53,6 +55,31 @@ public Event deserialize(
         context =
             jsonDeserializationContext.deserialize(
                 adjustContextResource(contextCopy), CloudFunctionsContext.class);
+      } else if (isPubSubEmulatorPayload(root)) {
+        JsonObject message = root.getAsJsonObject("message");
+
+        String timestampString = message.has("publishTime")
+          ? message.get("publishTime").getAsString()
+          : DateTimeFormatter.ISO_INSTANT.format(OffsetDateTime.now());
+
+        context = CloudFunctionsContext.builder()
+          .setEventType("google.pubsub.topic.publish")
+          .setTimestamp(timestampString)
+          .setEventId(message.get("messageId").getAsString())
+          .setResource("{" +
+              "\"name\":null," +
+              "\"service\":\"pubsub.googleapis.com\"," +
+              "\"type\":\"type.googleapis.com/google.pubsub.v1.PubsubMessage\""+
+            "}")
+          .build();
+
+        JsonObject marshalledData = new JsonObject();
+        marshalledData.addProperty("@type", "type.googleapis.com/google.pubsub.v1.PubsubMessage");
+        marshalledData.add("data", message.get("data"));
+        if (message.has("attributes")) {
+          marshalledData.add("attributes", message.get("attributes"));
+        }
+        data = marshalledData;
       } else {
         JsonObject rootCopy = root.deepCopy();
         rootCopy.remove("data");
@@ -63,6 +90,14 @@ public Event deserialize(
       return Event.of(data, context);
     }
 
+    private boolean isPubSubEmulatorPayload(JsonObject root) {
+      if (root.has("subscription") && root.has("message") && root.get("message").isJsonObject()) {
+        JsonObject message = root.getAsJsonObject("message");
+        return message.has("data") && message.has("messageId");
+      }
+      return false;
+    }
+
     /**
      * Replaces 'resource' member from context JSON with its string equivalent. The original
      * 'resource' member can be a JSON object itself while {@link CloudFunctionsContext} requires it

invoker/core/src/test/java/com/google/cloud/functions/invoker/BackgroundFunctionExecutorTest.java

@@ -2,14 +2,22 @@
 
 import static com.google.cloud.functions.invoker.BackgroundFunctionExecutor.backgroundFunctionTypeArgument;
 import static com.google.common.truth.Truth8.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 
 import com.google.cloud.functions.BackgroundFunction;
 import com.google.cloud.functions.Context;
+import com.google.gson.JsonObject;
+
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.Reader;
 import java.util.Map;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
+
 @RunWith(JUnit4.class)
 public class BackgroundFunctionExecutorTest {
   private static class PubSubMessage {
@@ -62,4 +70,38 @@ public void backgroundFunctionTypeArgument_raw() {
         (Class<? extends BackgroundFunction<?>>) (Class<?>) ForgotTypeParameter.class;
     assertThat(backgroundFunctionTypeArgument(c)).isEmpty();
   }
+
+  @Test
+  public void parseLegacyEventPubSub() throws IOException {
+    try (Reader reader = new InputStreamReader(getClass().getResourceAsStream("/pubsub_background.json"))) {
+      Event event = BackgroundFunctionExecutor.parseLegacyEvent(reader);
+
+      Context context = event.getContext();
+      assertEquals("google.pubsub.topic.publish", context.eventType());
+      assertEquals("1", context.eventId());
+      assertEquals("2021-06-28T05:46:32.390Z", context.timestamp());
+
+      JsonObject data = event.getData().getAsJsonObject();
+      assertEquals(data.get("data").getAsString(), "eyJmb28iOiJiYXIifQ==");
+      String attr = data.get("attributes").getAsJsonObject().get("test").getAsString();
+      assertEquals(attr, "123");
+    }
+  }
+
+  @Test
+  public void parseLegacyEventPubSubEmulator() throws IOException {
+    try (Reader reader = new InputStreamReader(getClass().getResourceAsStream("/pubsub_emulator.json"))) {
+      Event event = BackgroundFunctionExecutor.parseLegacyEvent(reader);
+
+      Context context = event.getContext();
+      assertEquals("google.pubsub.topic.publish", context.eventType());
+      assertEquals("1", context.eventId());
+      assertNotNull(context.timestamp());
+
+      JsonObject data = event.getData().getAsJsonObject();
+      assertEquals(data.get("data").getAsString(), "eyJmb28iOiJiYXIifQ==");
+      String attr = data.get("attributes").getAsJsonObject().get("test").getAsString();
+      assertEquals(attr, "123");
+    }
+  }
 }

invoker/core/src/test/resources/pubsub_background.json

@@ -0,0 +1,19 @@
+{
+  "data": {
+    "@type": "type.googleapis.com/google.pubsub.v1.PubsubMessage",
+    "data": "eyJmb28iOiJiYXIifQ==",
+    "attributes": {
+      "test": "123"
+    }
+  },
+  "context": {
+    "eventId": "1",
+    "eventType": "google.pubsub.topic.publish",
+    "resource": {
+      "name": "projects/FOO/topics/BAR_TOPIC",
+      "service": "pubsub.googleapis.com",
+      "type": "type.googleapis.com/google.pubsub.v1.PubsubMessage"
+    },
+    "timestamp": "2021-06-28T05:46:32.390Z"
+  }
+}

invoker/core/src/test/resources/pubsub_emulator.json

@@ -0,0 +1,10 @@
+{
+  "subscription": "projects/FOO/subscriptions/BAR_SUB",
+  "message": {
+    "data": "eyJmb28iOiJiYXIifQ==",
+    "messageId": "1",
+    "attributes": {
+      "test": "123"
+    }
+  }
+}