Declare CloudEvent generic to allow automatic payload deserialization using Jackson

daprdocs/content/en/java-sdk-docs/_index.md

@@ -150,7 +150,7 @@ public class SubscriberController {
 
   @Topic(name = "testingtopic", pubsubName = "${myAppProperty:messagebus}")
   @PostMapping(path = "/testingtopic")
-  public Mono<Void> handleMessage(@RequestBody(required = false) CloudEvent cloudEvent) {
+  public Mono<Void> handleMessage(@RequestBody(required = false) CloudEvent<?> cloudEvent) {
     return Mono.fromRunnable(() -> {
       try {
         System.out.println("Subscriber got: " + cloudEvent.getData());

examples/src/main/java/io/dapr/examples/pubsub/http/CloudEventPublisher.java

@@ -47,7 +47,7 @@ public static void main(String[] args) throws Exception {
     //Creating the DaprClient: Using the default builder client produces an HTTP Dapr Client
     try (DaprClient client = new DaprClientBuilder().build()) {
       for (int i = 0; i < NUM_MESSAGES; i++) {
-        CloudEvent cloudEvent = new CloudEvent();
+        CloudEvent<String> cloudEvent = new CloudEvent<>();
         cloudEvent.setId(UUID.randomUUID().toString());
         cloudEvent.setType("example");
         cloudEvent.setSpecversion("1");

examples/src/main/java/io/dapr/examples/pubsub/http/SubscriberController.java

@@ -28,7 +28,7 @@ public class SubscriberController {
    */
   @Topic(name = "testingtopic", pubsubName = "${myAppProperty:messagebus}")
   @PostMapping(path = "/testingtopic")
-  public Mono<Void> handleMessage(@RequestBody(required = false) CloudEvent cloudEvent) {
+  public Mono<Void> handleMessage(@RequestBody(required = false) CloudEvent<String> cloudEvent) {
     return Mono.fromRunnable(() -> {
       try {
         System.out.println("Subscriber got: " + cloudEvent.getData());

sdk-tests/pom.xml

@@ -31,6 +31,7 @@
     <grpc.version>1.39.0</grpc.version>
     <protobuf.version>3.13.0</protobuf.version>
     <opentelemetry.version>0.14.0</opentelemetry.version>
+    <spring-boot.version>2.3.5.RELEASE</spring-boot.version>
   </properties>
 
   <dependencies>
@@ -112,13 +113,13 @@
     <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-web</artifactId>
-      <version>2.3.5.RELEASE</version>
+      <version>${spring-boot.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-autoconfigure</artifactId>
-      <version>2.3.5.RELEASE</version>
+      <version>${spring-boot.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>

sdk-tests/src/test/java/io/dapr/it/pubsub/http/PubSubIT.java

@@ -43,13 +43,15 @@ public class PubSubIT extends BaseIT {
   private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
   private static final TypeRef<List<CloudEvent>> CLOUD_EVENT_LIST_TYPE_REF = new TypeRef<>() {};
+  private static final TypeRef<List<CloudEvent<MyObject>>> CLOUD_EVENT_MYOBJECT_LIST_TYPE_REF = new TypeRef<>() {};
 
   //Number of messages to be sent: 10
   private static final int NUM_MESSAGES = 10;
 
   private static final String PUBSUB_NAME = "messagebus";
   //The title of the topic to be used for publishing
   private static final String TOPIC_NAME = "testingtopic";
+  private static final String TYPED_TOPIC_NAME = "typedtestingtopic";
   private static final String ANOTHER_TOPIC_NAME = "anothertopic";
   // Topic used for TTL test
   private static final String TTL_TOPIC_NAME = "ttltopic";
@@ -166,6 +168,9 @@ public String getContentType() {
       client.publishEvent(PUBSUB_NAME, TOPIC_NAME, object).block();
       System.out.println("Published one object.");
 
+      client.publishEvent(PUBSUB_NAME, TYPED_TOPIC_NAME, object).block();
+      System.out.println("Published another object.");
+
       //Publishing a single byte: Example of non-string based content published
       client.publishEvent(
           PUBSUB_NAME,
@@ -235,6 +240,25 @@ public String getContentType() {
             .count() == 1);
       }, 2000);
 
+      callWithRetry(() -> {
+        System.out.println("Checking results for topic " + TYPED_TOPIC_NAME);
+        // Validate object payload.
+        final List<CloudEvent<MyObject>> messages = client.invokeMethod(
+                daprRun.getAppName(),
+                "messages/typedtestingtopic",
+                null,
+                HttpExtension.GET,
+                CLOUD_EVENT_MYOBJECT_LIST_TYPE_REF).block();
+
+        assertTrue(messages
+                .stream()
+                .filter(m -> m.getData() != null)
+                .filter(m -> m.getData() instanceof MyObject)
+                .map(m -> (MyObject)m.getData())
+                .filter(m -> "123".equals(m.getId()))
+                .count() == 1);
+      }, 2000);
+
       callWithRetry(() -> {
         System.out.println("Checking results for topic " + ANOTHER_TOPIC_NAME);
         final List<CloudEvent> messages = client.invokeMethod(

sdk-tests/src/test/java/io/dapr/it/pubsub/http/SubscriberController.java

@@ -8,43 +8,30 @@
 import io.dapr.Topic;
 import io.dapr.client.domain.CloudEvent;
 import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;
 import reactor.core.publisher.Mono;
 
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.function.BiFunction;
 
 /**
  * SpringBoot Controller to handle input binding.
  */
 @RestController
 public class SubscriberController {
 
-  private static final List<CloudEvent> messagesReceivedTestingTopic = new ArrayList();
-  private static final List<CloudEvent> messagesReceivedBinaryTopic = new ArrayList();
-  private static final List<CloudEvent> messagesReceivedAnotherTopic = new ArrayList();
-  private static final List<CloudEvent> messagesReceivedTTLTopic = new ArrayList();
+  private final Map<String, List<CloudEvent<?>>> messagesByTopic = new HashMap<>();
 
-  @GetMapping(path = "/messages/testingtopic")
-  public List<CloudEvent> getMessagesReceivedTestingTopic() {
-    return messagesReceivedTestingTopic;
-  }
-
-  @GetMapping(path = "/messages/binarytopic")
-  public List<CloudEvent> getMessagesReceivedBinaryTopic() {
-    return messagesReceivedBinaryTopic;
-  }
-
-  @GetMapping(path = "/messages/anothertopic")
-  public List<CloudEvent> getMessagesReceivedAnotherTopic() {
-    return messagesReceivedAnotherTopic;
-  }
-
-  @GetMapping(path = "/messages/ttltopic")
-  public List<CloudEvent> getMessagesReceivedTTLTopic() {
-    return messagesReceivedTTLTopic;
+  @GetMapping(path = "/messages/{topic}")
+  public List<CloudEvent<?>> getMessagesByTopic(@PathVariable("topic") String topic) {
+    return messagesByTopic.getOrDefault(topic, Collections.emptyList());
   }
 
   @Topic(name = "testingtopic", pubsubName = "messagebus")
@@ -55,7 +42,22 @@ public Mono<Void> handleMessage(@RequestBody(required = false) CloudEvent envelo
         String message = envelope.getData() == null ? "" : envelope.getData().toString();
         String contentType = envelope.getDatacontenttype() == null ? "" : envelope.getDatacontenttype();
         System.out.println("Testing topic Subscriber got message: " + message + "; Content-type: " + contentType);
-        messagesReceivedTestingTopic.add(envelope);
+        messagesByTopic.compute("testingtopic", merge(envelope));
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+    });
+  }
+
+  @Topic(name = "typedtestingtopic", pubsubName = "messagebus")
+  @PostMapping(path = "/route1b")
+  public Mono<Void> handleMessageTyped(@RequestBody(required = false) CloudEvent<PubSubIT.MyObject> envelope) {
+    return Mono.fromRunnable(() -> {
+      try {
+        String id = envelope.getData() == null ? "" : envelope.getData().getId();
+        String contentType = envelope.getDatacontenttype() == null ? "" : envelope.getDatacontenttype();
+        System.out.println("Testing typed topic Subscriber got message with ID: " + id + "; Content-type: " + contentType);
+        messagesByTopic.compute("typedtestingtopic", merge(envelope));
       } catch (Exception e) {
         throw new RuntimeException(e);
       }
@@ -70,7 +72,7 @@ public Mono<Void> handleBinaryMessage(@RequestBody(required = false) CloudEvent
         String message = envelope.getData() == null ? "" : envelope.getData().toString();
         String contentType = envelope.getDatacontenttype() == null ? "" : envelope.getDatacontenttype();
         System.out.println("Binary topic Subscriber got message: " + message + "; Content-type: " + contentType);
-        messagesReceivedBinaryTopic.add(envelope);
+        messagesByTopic.compute("binarytopic", merge(envelope));
       } catch (Exception e) {
         throw new RuntimeException(e);
       }
@@ -84,7 +86,7 @@ public Mono<Void> handleMessageAnotherTopic(@RequestBody(required = false) Cloud
       try {
         String message = envelope.getData() == null ? "" : envelope.getData().toString();
         System.out.println("Another topic Subscriber got message: " + message);
-        messagesReceivedAnotherTopic.add(envelope);
+        messagesByTopic.compute("anothertopic", merge(envelope));
       } catch (Exception e) {
         throw new RuntimeException(e);
       }
@@ -98,11 +100,18 @@ public Mono<Void> handleMessageTTLTopic(@RequestBody(required = false) CloudEven
       try {
         String message = envelope.getData() == null ? "" : envelope.getData().toString();
         System.out.println("TTL topic Subscriber got message: " + message);
-        messagesReceivedTTLTopic.add(envelope);
+        messagesByTopic.compute("ttltopic", merge(envelope));
       } catch (Exception e) {
         throw new RuntimeException(e);
       }
     });
   }
 
+  private BiFunction<String, List<CloudEvent<?>>, List<CloudEvent<?>>> merge(final CloudEvent<?> item) {
+    return (key, value) -> {
+      final List<CloudEvent<?>> list = value == null ? new ArrayList<>() : value;
+      list.add(item);
+      return list;
+    };
+  }
 }

sdk-tests/src/test/resources/logback.xml

@@ -1,7 +1,7 @@
 <configuration>
     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
         <encoder>
-            <pattern>%d {HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
         </encoder>
     </appender>
 

sdk/src/main/java/io/dapr/client/domain/CloudEvent.java

@@ -16,8 +16,9 @@
 
 /**
  * A cloud event in Dapr.
+ * @param <T> The type of the payload.
  */
-public final class CloudEvent {
+public final class CloudEvent<T> {
 
   /**
    * Mime type used for CloudEvent.
@@ -59,7 +60,7 @@ public final class CloudEvent {
   /**
    * Cloud event specs says data can be a JSON object or string.
    */
-  private Object data;
+  private T data;
 
   /**
    * Cloud event specs says binary data should be in data_base64.
@@ -88,7 +89,7 @@ public CloudEvent(
       String type,
       String specversion,
       String datacontenttype,
-      Object data) {
+      T data) {
     this.id = id;
     this.source = source;
     this.type = type;
@@ -126,7 +127,7 @@ public CloudEvent(
    * @return Message (can be null if input is null)
    * @throws IOException If cannot parse.
    */
-  public static CloudEvent deserialize(byte[] payload) throws IOException {
+  public static CloudEvent<?> deserialize(byte[] payload) throws IOException {
     if (payload == null) {
       return null;
     }
@@ -218,15 +219,15 @@ public void setDatacontenttype(String datacontenttype) {
    * Gets the cloud event data.
    * @return Cloud event's data. As per specs, data can be a JSON object or string.
    */
-  public Object getData() {
+  public T getData() {
     return data;
   }
 
   /**
    * Sets the cloud event data. As per specs, data can be a JSON object or string.
    * @param data Cloud event's data. As per specs, data can be a JSON object or string.
    */
-  public void setData(Object data) {
+  public void setData(T data) {
     this.data = data;
   }
 
@@ -257,7 +258,7 @@ public boolean equals(Object o) {
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-    CloudEvent that = (CloudEvent) o;
+    CloudEvent<?> that = (CloudEvent<?>) o;
     return Objects.equals(id, that.id)
         && Objects.equals(source, that.source)
         && Objects.equals(type, that.type)