fix local development with the Pub/Sub emulator (GoogleCloudPlatform#98)

* 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

* fix code style issues

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,34 @@ 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 +93,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

@@ -1,10 +1,15 @@
 package com.google.cloud.functions.invoker;
 
 import static com.google.cloud.functions.invoker.BackgroundFunctionExecutor.backgroundFunctionTypeArgument;
+import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth8.assertThat;
 
 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;
@@ -20,7 +25,8 @@ private static class PubSubMessage {
   }
 
   private static class PubSubFunction implements BackgroundFunction<PubSubMessage> {
-    @Override public void accept(PubSubMessage payload, Context context) {}
+    @Override
+    public void accept(PubSubMessage payload, Context context) {}
   }
 
   @Test
@@ -31,7 +37,8 @@ public void backgroundFunctionTypeArgument_simple() {
   private abstract static class Parent implements BackgroundFunction<PubSubMessage> {}
 
   private static class Child extends Parent {
-    @Override public void accept(PubSubMessage payload, Context context) {}
+    @Override
+    public void accept(PubSubMessage payload, Context context) {}
   }
 
   @Test
@@ -42,7 +49,8 @@ public void backgroundFunctionTypeArgument_superclass() {
   private interface GenericParent<T> extends BackgroundFunction<T> {}
 
   private static class GenericChild implements GenericParent<PubSubMessage> {
-    @Override public void accept(PubSubMessage payload, Context context) {}
+    @Override
+    public void accept(PubSubMessage payload, Context context) {}
   }
 
   @Test
@@ -52,7 +60,8 @@ public void backgroundFunctionTypeArgument_genericInterface() {
 
   @SuppressWarnings("rawtypes")
   private static class ForgotTypeParameter implements BackgroundFunction {
-    @Override public void accept(Object payload, Context context) {}
+    @Override
+    public void accept(Object payload, Context context) {}
   }
 
   @Test
@@ -62,4 +71,41 @@ 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();
+      assertThat(context.eventType()).isEqualTo("google.pubsub.topic.publish");
+      assertThat(context.eventId()).isEqualTo("1");
+      assertThat(context.timestamp()).isEqualTo("2021-06-28T05:46:32.390Z");
+
+      JsonObject data = event.getData().getAsJsonObject();
+      assertThat(data.get("data").getAsString()).isEqualTo("eyJmb28iOiJiYXIifQ==");
+      String attr = data.get("attributes").getAsJsonObject().get("test").getAsString();
+      assertThat(attr).isEqualTo("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();
+      assertThat(context.eventType()).isEqualTo("google.pubsub.topic.publish");
+      assertThat(context.eventId()).isEqualTo("1");
+      assertThat(context.timestamp()).isNotNull();
+      ;
+
+      JsonObject data = event.getData().getAsJsonObject();
+      assertThat(data.get("data").getAsString()).isEqualTo("eyJmb28iOiJiYXIifQ==");
+      String attr = data.get("attributes").getAsJsonObject().get("test").getAsString();
+      assertThat(attr).isEqualTo("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"
+    }
+  }
+}