Import of Cloud Functions JVM from Git-on-Borg.

  - 098bb04cc54e68dd0d37ef0b03a38730191c36d1 Add support for CloudEvents to the Java Functions Framewo... by Éamonn McManus <[email protected]>

PiperOrigin-RevId: 291057836

Author: eamonnmcmanus

invoker-core/core/pom.xml

@@ -32,6 +32,11 @@
         <artifactId>javax.servlet-api</artifactId>
         <version>3.1.0</version>
     </dependency>
+    <dependency>
+      <groupId>io.cloudevents</groupId>
+      <artifactId>cloudevents-api</artifactId>
+      <version>1.2.0</version>
+    </dependency>
     <dependency>
         <groupId>javax.annotation</groupId>
         <artifactId>javax.annotation-api</artifactId>

invoker-core/core/src/main/java/com/google/cloud/functions/invoker/CloudFunctionsContext.java

@@ -24,4 +24,17 @@ abstract class CloudFunctionsContext implements Context {
   public static TypeAdapter<CloudFunctionsContext> typeAdapter(Gson gson) {
     return new AutoValue_CloudFunctionsContext.GsonTypeAdapter(gson);
   }
+
+  static Builder builder() {
+    return new AutoValue_CloudFunctionsContext.Builder();
+  }
+
+  @AutoValue.Builder
+  abstract static class Builder {
+    abstract Builder setEventId(String x);
+    abstract Builder setTimestamp(String x);
+    abstract Builder setEventType(String x);
+    abstract Builder setResource(String x);
+    abstract CloudFunctionsContext build();
+  }
 }

invoker-core/core/src/main/java/com/google/cloud/functions/invoker/NewBackgroundFunctionExecutor.java

@@ -1,16 +1,26 @@
 package com.google.cloud.functions.invoker;
 
+import static java.util.stream.Collectors.joining;
+import static java.util.stream.Collectors.toMap;
+
 import com.google.cloud.functions.BackgroundFunction;
 import com.google.cloud.functions.Context;
 import com.google.cloud.functions.RawBackgroundFunction;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
-import com.google.gson.JsonParseException;
 import com.google.gson.TypeAdapter;
+import io.cloudevents.CloudEvent;
+import io.cloudevents.format.builder.HeadersStep;
+import io.cloudevents.v1.AttributesImpl;
+import io.cloudevents.v1.http.Unmarshallers;
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.lang.reflect.Type;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
 import java.util.Optional;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -19,13 +29,13 @@
 import javax.servlet.http.HttpServletResponse;
 
 /** Executes the user's background function. */
-public class NewBackgroundFunctionExecutor extends HttpServlet {
+public final class NewBackgroundFunctionExecutor extends HttpServlet {
   private static final Logger logger = Logger.getLogger("com.google.cloud.functions.invoker");
 
-  private final RawBackgroundFunction function;
+  private final FunctionExecutor<?> functionExecutor;
 
-  private NewBackgroundFunctionExecutor(RawBackgroundFunction function) {
-    this.function = function;
+  private NewBackgroundFunctionExecutor(FunctionExecutor<?> functionExecutor) {
+    this.functionExecutor = functionExecutor;
   }
 
   /**
@@ -58,24 +68,24 @@ public static Optional<NewBackgroundFunctionExecutor> forTarget(
     } catch (ReflectiveOperationException e) {
       throw new RuntimeException("Could not construct an instance of " + target + ": " + e, e);
     }
-    RawBackgroundFunction function =
-        (instance instanceof RawBackgroundFunction)
-            ? (RawBackgroundFunction) instance
-            : asRaw((BackgroundFunction<?>) instance);
-    return Optional.of(new NewBackgroundFunctionExecutor(function));
-  }
-
-  private static <T> RawBackgroundFunction asRaw(BackgroundFunction<T> backgroundFunction) {
-    Optional<Type> maybeTargetType = backgroundFunctionTypeArgument(backgroundFunction.getClass());
-    if (!maybeTargetType.isPresent()) {
-      // This is probably because the user implemented just BackgroundFunction rather than
-      // BackgroundFunction<T>.
-      throw new RuntimeException(
-          "Could not determine the payload type for BackgroundFunction of type "
-              + backgroundFunction.getClass().getName()
-              + "; must implement BackgroundFunction<T> for some T");
+    FunctionExecutor<?> executor;
+    if (instance instanceof RawBackgroundFunction) {
+      executor = new RawFunctionExecutor((RawBackgroundFunction) instance);
+    } else {
+      BackgroundFunction<?> backgroundFunction = (BackgroundFunction<?>) instance;
+      Optional<Type> maybeTargetType =
+          backgroundFunctionTypeArgument(backgroundFunction.getClass());
+      if (!maybeTargetType.isPresent()) {
+        // This is probably because the user implemented just BackgroundFunction rather than
+        // BackgroundFunction<T>.
+        throw new RuntimeException(
+            "Could not determine the payload type for BackgroundFunction of type "
+                + instance.getClass().getName()
+                + "; must implement BackgroundFunction<T> for some T");
+      }
+      executor = new TypedFunctionExecutor<>(maybeTargetType.get(), backgroundFunction);
     }
-    return new AsRaw<T>(maybeTargetType.get(), backgroundFunction);
+    return Optional.of(new NewBackgroundFunctionExecutor(executor));
   }
 
   /**
@@ -97,55 +107,224 @@ static Optional<Type> backgroundFunctionTypeArgument(
         .findFirst();
   }
 
+  private static Event parseLegacyEvent(HttpServletRequest req) throws IOException {
+    try (BufferedReader bodyReader = req.getReader()) {
+      // A Type Adapter is required to set the type of the JsonObject because CloudFunctionsContext
+      // is abstract and Gson default behavior instantiates the type provided.
+      TypeAdapter<CloudFunctionsContext> typeAdapter =
+          CloudFunctionsContext.typeAdapter(new Gson());
+      Gson gson = new GsonBuilder()
+          .registerTypeAdapter(CloudFunctionsContext.class, typeAdapter)
+          .registerTypeAdapter(Event.class, new Event.EventDeserializer())
+          .create();
+      return gson.fromJson(bodyReader, Event.class);
+    }
+  }
+
+  private static Context contextFromCloudEvent(CloudEvent<AttributesImpl, ?> cloudEvent) {
+    AttributesImpl attributes = cloudEvent.getAttributes();
+    ZonedDateTime timestamp = attributes.getTime().orElse(ZonedDateTime.now());
+    String timestampString = DateTimeFormatter.ISO_INSTANT.format(timestamp);
+    // We don't have an obvious replacement for the Context.resource field, which with legacy events
+    // corresponded to a value present for some proprietary Google event types.
+    String resource = "{}";
+    return CloudFunctionsContext.builder()
+        .setEventId(attributes.getId())
+        .setEventType(attributes.getType())
+        .setResource(resource)
+        .setTimestamp(timestampString)
+        .build();
+  }
+
   /**
-   * Wraps a typed {@link BackgroundFunction} as a {@link RawBackgroundFunction} that takes its
-   * input JSON string and deserializes it into the payload type of the {@link BackgroundFunction}/
+   * Convert the HTTP headers from the given request into a Map. The headers of interest are
+   * the CE-* headers defined for CloudEvents in the binary encoding (where the metadata is in
+   * the HTTP headers and the payload is the HTTP body), plus Content-Type. In both cases we don't
+   * need to worry about repeated headers, so {@link HttpServletRequest#getHeader(String)} is fine.
    */
-  private static class AsRaw<T> implements RawBackgroundFunction {
-    private final Gson gson = new Gson();
-    private final Type targetType;
-    private final BackgroundFunction<T> backgroundFunction;
+  private static Map<String, Object> httpHeaderMap(HttpServletRequest req) {
+    return Collections.list(req.getHeaderNames())
+        .stream()
+        .collect(toMap(header -> header, req::getHeader));
+  }
 
-    private AsRaw(Type targetType, BackgroundFunction<T> backgroundFunction) {
-      this.targetType = targetType;
-      this.backgroundFunction = backgroundFunction;
+  /**
+   * A background function, either "raw" or "typed". A raw background function is one where the user
+   * code receives a String parameter that is the JSON payload of the triggering event. A typed
+   * background function is one where the payload is deserialized into a user-provided class whose
+   * field names correspond to the keys of the JSON object.
+   *
+   * <p>In addition to these two flavours, events can be either "legacy events" or "CloudEvents".
+   * Legacy events are the only kind that GCF originally supported, and use proprietary encodings
+   * for the various triggers. CloudEvents are ones that follow the standards defined by
+   * <a href="https://cloudevents.io">cloudevents.io</a>.
+   *
+   * @param <CloudEventDataT> the type to be used in the {@link Unmarshallers} call when
+   *     unmarshalling this event, if it is a CloudEvent.
+   */
+  private abstract static class FunctionExecutor<CloudEventDataT> {
+    private final String functionName;
+
+    FunctionExecutor(String functionName) {
+      this.functionName = functionName;
+    }
+
+    final String functionName() {
+      return functionName;
+    }
+
+    abstract void serviceLegacyEvent(HttpServletRequest req)
+        throws IOException;
+
+    abstract void serviceCloudEvent(
+        HttpServletRequest req,
+        HeadersStep<AttributesImpl, CloudEventDataT, String> unmarshaller)
+        throws IOException;
+
+    abstract Class<CloudEventDataT> cloudEventDataType();
+  }
+
+  private static class RawFunctionExecutor extends FunctionExecutor<Map<?, ?>> {
+    private final RawBackgroundFunction function;
+
+    RawFunctionExecutor(RawBackgroundFunction function) {
+      super(function.getClass().getCanonicalName());
+      this.function = function;
     }
 
     @Override
-    public void accept(String json, Context context) {
-      T payload;
-      try {
-        payload = gson.fromJson(json, targetType);
-      } catch (JsonParseException e) {
-        logger.log(Level.WARNING,
-            "Could not convert payload to target type " + targetType.getTypeName(), e);
-        return;
+    void serviceLegacyEvent(HttpServletRequest req) throws IOException {
+      Event event = parseLegacyEvent(req);
+      function.accept(new Gson().toJson(event.getData()), event.getContext());
+    }
+
+    @Override
+    void serviceCloudEvent(
+        HttpServletRequest req, HeadersStep<AttributesImpl, Map<?, ?>, String> unmarshaller)
+        throws IOException {
+      Map<String, Object> httpHeaders = httpHeaderMap(req);
+      String body = req.getReader().lines().collect(joining("\n"));
+      CloudEvent<AttributesImpl, Map<?, ?>> cloudEvent =
+          unmarshaller
+              .withHeaders(() -> httpHeaders)
+              .withPayload(() -> body)
+              .unmarshal();
+      Context context = contextFromCloudEvent(cloudEvent);
+      String jsonData = cloudEvent.getData().map(data -> new Gson().toJson(data)).orElse("{}");
+      function.accept(jsonData, context);
+    }
+
+    @Override
+    Class<Map<?, ?>> cloudEventDataType() {
+      // This messing about with casts and @SuppressWarnings allows us to limit the use of the raw
+      // Map type to just here.
+      @SuppressWarnings("unchecked")
+      Class<Map<?, ?>> c = (Class<Map<?, ?>>) (Class<?>) Map.class;
+      return c;
+    }
+  }
+
+  private static class TypedFunctionExecutor<T> extends FunctionExecutor<T> {
+    private final Type type; // T
+    private final BackgroundFunction<T> function;
+
+    private TypedFunctionExecutor(Type type, BackgroundFunction<T> function) {
+      super(function.getClass().getCanonicalName());
+      this.type = type;
+      this.function = function;
+    }
+
+    static <T> TypedFunctionExecutor<T> of(Type type, BackgroundFunction<?> instance) {
+      @SuppressWarnings("unchecked")
+      BackgroundFunction<T> function = (BackgroundFunction<T>) instance;
+      return new TypedFunctionExecutor<>(type, function);
+    }
+
+    @Override
+    void serviceLegacyEvent(HttpServletRequest req) throws IOException {
+      Event event = parseLegacyEvent(req);
+      T payload = new Gson().fromJson(event.getData(), type);
+      function.accept(payload, event.getContext());
+    }
+
+    @Override
+    void serviceCloudEvent(
+        HttpServletRequest req, HeadersStep<AttributesImpl, T, String> unmarshaller)
+        throws IOException {
+      Map<String, Object> httpHeaders = httpHeaderMap(req);
+      String body = req.getReader().lines().collect(joining("\n"));
+      CloudEvent<AttributesImpl, T> cloudEvent =
+          unmarshaller
+              .withHeaders(() -> httpHeaders)
+              .withPayload(() -> body).unmarshal();
+      if (cloudEvent.getData().isPresent()) {
+        Context context = contextFromCloudEvent(cloudEvent);
+        function.accept(cloudEvent.getData().get(), context);
+      } else {
+        throw new IllegalStateException("Event has no \"data\" component");
       }
-      backgroundFunction.accept(payload, context);
+    }
+
+    @Override
+    Class<T> cloudEventDataType() {
+      if (!(type instanceof Class<?>)) {
+        throw new IllegalStateException(
+            "CloudEvents SDK currently does not permit deserializing types other than classes:"
+            + " cannot deserialize " + type);
+      }
+      @SuppressWarnings("unchecked")
+      Class<T> c = (Class<T>) type;
+      return c;
     }
   }
 
-  /** Executes the user's background function, can handle all HTTP type methods. */
+  /** Executes the user's background function. This can handle all HTTP methods. */
   @Override
   public void service(HttpServletRequest req, HttpServletResponse res) throws IOException {
-    BufferedReader body = req.getReader();
-
-    // A Type Adapter is required to set the type of the JsonObject because CloudFunctionsContext
-    // is abstract and Gson default behavior instantiates the type provided.
-    TypeAdapter<CloudFunctionsContext> typeAdapter =
-        CloudFunctionsContext.typeAdapter(new Gson());
-    Gson gson = new GsonBuilder()
-        .registerTypeAdapter(CloudFunctionsContext.class, typeAdapter)
-        .registerTypeAdapter(Event.class, new Event.EventDeserializer())
-        .create();
-
-    Event event = gson.fromJson(body, Event.class);
+    String contentType = req.getContentType();
     try {
-      function.accept(gson.toJson(event.getData()), event.getContext());
+      if (contentType != null && contentType.startsWith("application/cloudevents+json")) {
+        serviceCloudEvent(req, CloudEventKind.STRUCTURED);
+      } else if (req.getHeader("ce-specversion") != null) {
+        serviceCloudEvent(req, CloudEventKind.BINARY);
+      } else {
+        serviceLegacyEvent(req);
+      }
       res.setStatus(HttpServletResponse.SC_OK);
     } catch (Throwable t) {
       res.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
-      logger.log(Level.WARNING, "Failed to execute " + function.getClass().getName(), t);
+      logger.log(Level.WARNING, "Failed to execute " + functionExecutor.functionName(), t);
+    }
+  }
+
+  private enum CloudEventKind {BINARY, STRUCTURED}
+
+  /**
+   * Service a CloudEvent.
+   *
+   * @param <CloudEventT> a fake type parameter, which corresponds to the type parameter of
+   *     {@link FunctionExecutor}.
+   */
+  private <CloudEventT> void serviceCloudEvent(
+      HttpServletRequest req, CloudEventKind kind) throws IOException {
+    @SuppressWarnings("unchecked")
+    FunctionExecutor<CloudEventT> executor = (FunctionExecutor<CloudEventT>) functionExecutor;
+    Class<CloudEventT> cloudEventDataType = executor.cloudEventDataType();
+    HeadersStep<AttributesImpl, CloudEventT, String> unmarshaller;
+    switch (kind) {
+      case BINARY:
+        unmarshaller = Unmarshallers.binary(cloudEventDataType);
+        break;
+      case STRUCTURED:
+        unmarshaller = Unmarshallers.structured(cloudEventDataType);
+        break;
+      default:
+        throw new AssertionError(kind);
     }
+    executor.serviceCloudEvent(req, unmarshaller);
+  }
+
+  private void serviceLegacyEvent(HttpServletRequest req) throws IOException {
+    functionExecutor.serviceLegacyEvent(req);
   }
 }