Switch to V2 of the CloudEvents SDK. (#50)

* Switch to V2 of the CloudEvents SDK. Also explicitly require Java version 11, since we no longer suport Java 8.

This version of the SDK makes a cleaner separation between API and implementation, and our use of it is accordingly simplified.

* Remove things that don't belong in this commit.

* Fix Java version, 11 not 1.11.

* Fix one source of javadoc problems.

* Remove some stray code from a later change.

* Add missing class.

Author: eamonnmcmanus

functions-framework-api/pom.xml

@@ -54,8 +54,8 @@
         <artifactId>maven-compiler-plugin</artifactId>
         <version>${maven-compiler-plugin.version}</version>
         <configuration>
-          <source>1.8</source>
-          <target>1.8</target>
+          <source>11</source>
+          <target>11</target>
         </configuration>
       </plugin>
       <plugin>
@@ -108,6 +108,7 @@
             </additionalOptions>
             <linksource>true</linksource>
             <source>8</source>
+            <detectJavaApiLink>false</detectJavaApiLink>
           </configuration>
           <executions>
             <execution>

invoker/core/pom.xml

@@ -20,6 +20,8 @@
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <junit.jupiter.version>5.3.2</junit.jupiter.version>
+    <maven.compiler.source>11</maven.compiler.source>
+    <maven.compiler.target>11</maven.compiler.target>
   </properties>
 
   <licenses>
@@ -50,7 +52,17 @@
     <dependency>
       <groupId>io.cloudevents</groupId>
       <artifactId>cloudevents-api</artifactId>
-      <version>1.2.0</version>
+      <version>2.0.0-milestone1</version>
+    </dependency>
+    <dependency>
+      <groupId>io.cloudevents</groupId>
+      <artifactId>cloudevents-core</artifactId>
+      <version>2.0.0-milestone1</version>
+    </dependency>
+    <dependency>
+      <groupId>io.cloudevents</groupId>
+      <artifactId>cloudevents-json-jackson</artifactId>
+      <version>2.0.0-milestone1</version>
     </dependency>
     <dependency>
       <groupId>com.google.code.gson</groupId>

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

@@ -0,0 +1,70 @@
+package com.google.cloud.functions.invoker;
+
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.message.impl.BaseGenericBinaryMessageReaderImpl;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.function.BiConsumer;
+import javax.servlet.http.HttpServletRequest;
+
+class CloudEventsServletBinaryMessageReader extends BaseGenericBinaryMessageReaderImpl<String, String> {
+  private final Map<String, List<String>> headers;
+
+  private CloudEventsServletBinaryMessageReader(Map<String, List<String>> headers, byte[] body) {
+    super(SpecVersion.V1, body);
+    this.headers = headers;
+  }
+
+  static CloudEventsServletBinaryMessageReader from(HttpServletRequest request, byte[] body) {
+    Map<String, List<String>> headerMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+    for (String header : Collections.list(request.getHeaderNames())) {
+      for (String value : Collections.list(request.getHeaders(header))) {
+        headerMap.computeIfAbsent(header, unused -> new ArrayList<>()).add(value);
+      }
+    }
+    return new CloudEventsServletBinaryMessageReader(headerMap, body);
+  }
+
+  @Override
+  protected boolean isContentTypeHeader(String header) {
+    return header.equalsIgnoreCase("content-type");
+  }
+
+  @Override
+  protected boolean isCloudEventsHeader(String header) {
+    return header.toLowerCase(Locale.ENGLISH).startsWith("ce-");
+  }
+
+  @Override
+  protected String toCloudEventsKey(String header) {
+    if (!isCloudEventsHeader(header)) {
+      throw new IllegalArgumentException("Not a CloudEvents header: " + header);
+    }
+    return header.substring(3).toLowerCase(Locale.ENGLISH);
+  }
+
+  @Override
+  protected void forEachHeader(BiConsumer<String, String> consumer) {
+    headers.forEach((header, values) -> values.forEach(value -> consumer.accept(header, value)));
+  }
+
+  @Override
+  protected String toCloudEventsValue(String value) {
+    return value;
+  }
+
+  static Map<String, List<String>> headerMap(HttpServletRequest request) {
+    Map<String, List<String>> headerMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+    for (String header : Collections.list(request.getHeaderNames())) {
+      for (String value : Collections.list(request.getHeaders(header))) {
+        headerMap.computeIfAbsent(header, unused -> new ArrayList<>()).add(value);
+      }
+    }
+    return headerMap;
+  }
+}

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

@@ -14,6 +14,7 @@
 
 package com.google.cloud.functions.invoker;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.stream.Collectors.joining;
 import static java.util.stream.Collectors.toMap;
 
@@ -24,16 +25,18 @@
 import com.google.gson.GsonBuilder;
 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 io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.message.impl.GenericStructuredMessageReader;
+import io.cloudevents.core.message.impl.MessageUtils;
+import io.cloudevents.core.message.impl.UnknownEncodingMessageReader;
 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.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.logging.Level;
@@ -78,8 +81,10 @@ public static NewBackgroundFunctionExecutor forClass(Class<?> functionClass) {
       executor = new RawFunctionExecutor((RawBackgroundFunction) instance);
     } else {
       BackgroundFunction<?> backgroundFunction = (BackgroundFunction<?>) instance;
-      Optional<Type> maybeTargetType =
-          backgroundFunctionTypeArgument(backgroundFunction.getClass());
+      @SuppressWarnings("unchecked")
+      Class<? extends BackgroundFunction<?>> c =
+          (Class<? extends BackgroundFunction<?>>) backgroundFunction.getClass();
+      Optional<Type> maybeTargetType = backgroundFunctionTypeArgument(c);
       if (!maybeTargetType.isPresent()) {
         // This is probably because the user implemented just BackgroundFunction rather than
         // BackgroundFunction<T>.
@@ -99,7 +104,7 @@ public static NewBackgroundFunctionExecutor forClass(Class<?> functionClass) {
    * {@code T} can't be determined.
    */
   static Optional<Type> backgroundFunctionTypeArgument(
-      Class<? extends BackgroundFunction> functionClass) {
+      Class<? extends BackgroundFunction<?>> functionClass) {
     // If this is BackgroundFunction<Foo> then the user must have implemented a method
     // accept(Foo, Context), so we look for that method and return the type of its first argument.
     // We must be careful because the compiler will also have added a synthetic method
@@ -126,35 +131,24 @@ private static Event parseLegacyEvent(HttpServletRequest req) throws IOException
     }
   }
 
-  private static Context contextFromCloudEvent(CloudEvent<AttributesImpl, ?> cloudEvent) {
-    AttributesImpl attributes = cloudEvent.getAttributes();
-    ZonedDateTime timestamp = attributes.getTime().orElse(ZonedDateTime.now());
+  private static Context contextFromCloudEvent(CloudEvent cloudEvent) {
+    ZonedDateTime timestamp = Optional.ofNullable(cloudEvent.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 = "{}";
-    Map<String, String> attributesMap = AttributesImpl.marshal(attributes);
+    Map<String, String> attributesMap =
+        cloudEvent.getAttributeNames().stream()
+            .collect(toMap(a -> a, a -> String.valueOf(cloudEvent.getAttribute(a))));
     return CloudFunctionsContext.builder()
-        .setEventId(attributes.getId())
-        .setEventType(attributes.getType())
+        .setEventId(cloudEvent.getId())
+        .setEventType(cloudEvent.getType())
         .setResource(resource)
         .setTimestamp(timestampString)
         .setAttributes(attributesMap)
         .build();
   }
 
-  /**
-   * 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 Map<String, Object> httpHeaderMap(HttpServletRequest req) {
-    return Collections.list(req.getHeaderNames())
-        .stream()
-        .collect(toMap(header -> header, req::getHeader));
-  }
-
   /**
    * 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
@@ -187,10 +181,7 @@ final ClassLoader functionClassLoader() {
     abstract void serviceLegacyEvent(HttpServletRequest req)
         throws Exception;
 
-    abstract void serviceCloudEvent(
-        HttpServletRequest req,
-        HeadersStep<AttributesImpl, CloudEventDataT, String> unmarshaller)
-        throws Exception;
+    abstract void serviceCloudEvent(CloudEvent cloudEvent) throws Exception;
 
     abstract Class<CloudEventDataT> cloudEventDataType();
   }
@@ -210,18 +201,9 @@ void serviceLegacyEvent(HttpServletRequest req) throws Exception {
     }
 
     @Override
-    void serviceCloudEvent(
-        HttpServletRequest req, HeadersStep<AttributesImpl, Map<?, ?>, String> unmarshaller)
-        throws Exception {
-      Map<String, Object> httpHeaders = httpHeaderMap(req);
-      String body = req.getReader().lines().collect(joining("\n"));
-      CloudEvent<AttributesImpl, Map<?, ?>> cloudEvent =
-          unmarshaller
-              .withHeaders(() -> httpHeaders)
-              .withPayload(() -> body)
-              .unmarshal();
+    void serviceCloudEvent(CloudEvent cloudEvent) throws Exception {
       Context context = contextFromCloudEvent(cloudEvent);
-      String jsonData = cloudEvent.getData().map(data -> new Gson().toJson(data)).orElse("{}");
+      String jsonData = cloudEvent.getData() == null ? "{}" : new String(cloudEvent.getData(), UTF_8);
       function.accept(jsonData, context);
     }
 
@@ -259,18 +241,12 @@ void serviceLegacyEvent(HttpServletRequest req) throws Exception {
     }
 
     @Override
-    void serviceCloudEvent(
-        HttpServletRequest req, HeadersStep<AttributesImpl, T, String> unmarshaller)
-        throws Exception {
-      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()) {
+    void serviceCloudEvent(CloudEvent cloudEvent) throws Exception {
+      if (cloudEvent.getData() != null) {
+        String data = new String(cloudEvent.getData(), UTF_8);
+        T payload = new Gson().fromJson(data, type);
         Context context = contextFromCloudEvent(cloudEvent);
-        function.accept(cloudEvent.getData().get(), context);
+        function.accept(payload, context);
       } else {
         throw new IllegalStateException("Event has no \"data\" component");
       }
@@ -296,10 +272,9 @@ public void service(HttpServletRequest req, HttpServletResponse res) throws IOEx
     ClassLoader oldContextLoader = Thread.currentThread().getContextClassLoader();
     try {
       Thread.currentThread().setContextClassLoader(functionExecutor.functionClassLoader());
-      if (contentType != null && contentType.startsWith("application/cloudevents+json")) {
-        serviceCloudEvent(req, CloudEventKind.STRUCTURED);
-      } else if (req.getHeader("ce-specversion") != null) {
-        serviceCloudEvent(req, CloudEventKind.BINARY);
+      if ((contentType != null && contentType.startsWith("application/cloudevents+json"))
+          || req.getHeader("ce-specversion") != null) {
+        serviceCloudEvent(req);
       } else {
         serviceLegacyEvent(req);
       }
@@ -320,23 +295,19 @@ private enum CloudEventKind {BINARY, STRUCTURED}
    * @param <CloudEventT> a fake type parameter, which corresponds to the type parameter of
    *     {@link FunctionExecutor}.
    */
-  private <CloudEventT> void serviceCloudEvent(
-      HttpServletRequest req, CloudEventKind kind) throws Exception {
+  private <CloudEventT> void serviceCloudEvent(HttpServletRequest req) throws Exception {
     @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);
+    Map<String, List<String>> headers = CloudEventsServletBinaryMessageReader.headerMap(req);
+    byte[] body = req.getInputStream().readAllBytes();
+    List<String> listOfNull = Collections.singletonList(null);
+    MessageReader reader = MessageUtils.parseStructuredOrBinaryMessage(
+        () -> headers.getOrDefault("content-type", listOfNull).get(0),
+        format -> new GenericStructuredMessageReader(format, body),
+        () -> headers.getOrDefault("ce-specversion", listOfNull).get(0),
+        unusedSpecVersion -> CloudEventsServletBinaryMessageReader.from(req, body),
+        UnknownEncodingMessageReader::new);
+    executor.serviceCloudEvent(reader.toEvent());
   }
 
   private void serviceLegacyEvent(HttpServletRequest req) throws Exception {