Allows override of content-type in pubsub publish. (dapr#492)

artursouza · web-flow · commit ce7ecb996573 · 2021-02-16T00:03:08.000-08:00
diff --git a/examples/src/main/java/io/dapr/examples/pubsub/http/CloudEventPublisher.java b/examples/src/main/java/io/dapr/examples/pubsub/http/CloudEventPublisher.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) Microsoft Corporation and Dapr Contributors.
+ * Licensed under the MIT License.
+ */
+
+package io.dapr.examples.pubsub.http;
+
+import io.dapr.client.DaprClient;
+import io.dapr.client.DaprClientBuilder;
+import io.dapr.client.domain.CloudEvent;
+import io.dapr.client.domain.Metadata;
+import io.dapr.client.domain.PublishEventRequestBuilder;
+
+import java.util.UUID;
+
+import static java.util.Collections.singletonMap;
+
+/**
+ * Message publisher.
+ * 1. Build and install jars:
+ * mvn clean install
+ * 2. cd [repo root]/examples
+ * 3. Run the program:
+ * dapr run --components-path ./components/pubsub --app-id publisher -- \
+ *   java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.http.CloudEventPublisher
+ */
+public class CloudEventPublisher {
+
+  //Number of messages to be sent.
+  private static final int NUM_MESSAGES = 10;
+
+  //Time-to-live for messages published.
+  private static final String MESSAGE_TTL_IN_SECONDS = "1000";
+
+  //The title of the topic to be used for publishing
+  private static final String TOPIC_NAME = "testingtopic";
+
+  //The name of the pubsub
+  private static final String PUBSUB_NAME = "messagebus";
+
+  /**
+   * This is the entry point of the publisher app example.
+   * @param args Args, unused.
+   * @throws Exception A startup Exception.
+   */
+  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.setId(UUID.randomUUID().toString());
+        cloudEvent.setType("example");
+        cloudEvent.setSpecversion("1");
+        cloudEvent.setDatacontenttype("text/plain");
+        cloudEvent.setData(String.format("This is message #%d", i));
+
+        //Publishing messages
+        client.publishEvent(
+            new PublishEventRequestBuilder(PUBSUB_NAME, TOPIC_NAME, cloudEvent)
+                .withContentType(CloudEvent.CONTENT_TYPE)
+                .withMetadata(singletonMap(Metadata.TTL_IN_SECONDS, MESSAGE_TTL_IN_SECONDS))
+                .build()).block();
+        System.out.println("Published cloud event with message: " + cloudEvent.getData());
+
+        try {
+          Thread.sleep((long) (1000 * Math.random()));
+        } catch (InterruptedException e) {
+          e.printStackTrace();
+          Thread.currentThread().interrupt();
+          return;
+        }
+      }
+
+      // This is an example, so for simplicity we are just exiting here.
+      // Normally a dapr app would be a web service and not exit main.
+      System.out.println("Done.");
+    }
+  }
+}
diff --git a/examples/src/main/java/io/dapr/examples/pubsub/http/Publisher.java b/examples/src/main/java/io/dapr/examples/pubsub/http/Publisher.java
@@ -40,7 +40,6 @@ public class Publisher {
    * @throws Exception A startup Exception.
    */
   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++) {
         String message = String.format("This is message #%d", i);
diff --git a/examples/src/main/java/io/dapr/examples/pubsub/http/README.md b/examples/src/main/java/io/dapr/examples/pubsub/http/README.md
@@ -96,14 +96,15 @@ dapr run --components-path ./components/pubsub --app-id subscriber --app-port 30
 
 The other component is the publisher. It is a simple java application with a main method that uses the Dapr HTTP Client to publish 10 messages to an specific topic.
 
-In the `Publisher.java` file, you will find the `Publisher` class, containing the main method. The main method declares a Dapr Client using the `DaprClientBuilder` class. Notice that this builder gets two serializer implementations in the constructor: One is for Dapr's sent and recieved objects, and second is for objects to be persisted. The client publishes messages using `publishEvent` method. The Dapr client is also within a try-with-resource block to properly close the client at the end. See the code snippet below:  
+In the `Publisher.java` file, you will find the `Publisher` class, containing the main method. The main method declares a Dapr Client using the `DaprClientBuilder` class. Notice that this builder gets two serializer implementations in the constructor: One is for Dapr's sent and recieved objects, and second is for objects to be persisted. The client publishes messages using `publishEvent` method. The Dapr client is also within a try-with-resource block to properly close the client at the end. See the code snippet below:
+Dapr sidecar will automatically wrap the payload received into a CloudEvent object, which will later on parsed by the subscriber.
 ```java
 public class Publisher {
-    private static final int NUM_MESSAGES = 10;
-    private static final String TOPIC_NAME = "testingtopic";
-    private static final String PUBSUB_NAME = "messagebus";
+  private static final int NUM_MESSAGES = 10;
+  private static final String TOPIC_NAME = "testingtopic";
+  private static final String PUBSUB_NAME = "messagebus";
 
-///...
+  ///...
   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()) {
@@ -114,28 +115,41 @@ public class Publisher {
           System.out.println("Published message: " + message);
           //...
         }
+        ///...
       }
   }
-///...
 }
 ```
 
-This example also pushes a non-string content event, the follow code in same `Publisher` main method publishes a bite:
+The `CloudEventPublisher.java` file shows how the same can be accomplished if the application must send a CloudEvent object instead of relying on Dapr's automatic CloudEvent "wrapping".
+In this case, the app MUST override the content-type parameter via `withContentType()`, so Dapr sidecar knows that the payload is already a CloudEvent object.
 
 ```java
 public class Publisher {
-///...
-    public static void main(String[] args) throws Exception {
-///...
-    //Publishing a single bite: Example of non-string based content published
-    client.publishEvent(
-        TOPIC_NAME,
-        new byte[] { 1 },
-        Collections.singletonMap("content-type", "application/octet-stream")).block();
-    System.out.println("Published one byte.");
-    System.out.println("Done.");
+  ///...
+  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.setId(UUID.randomUUID().toString());
+        cloudEvent.setType("example");
+        cloudEvent.setSpecversion("1");
+        cloudEvent.setDatacontenttype("text/plain");
+        cloudEvent.setData(String.format("This is message #%d", i));
+
+        //Publishing messages
+        client.publishEvent(
+            new PublishEventRequestBuilder(PUBSUB_NAME, TOPIC_NAME, cloudEvent)
+                .withContentType(CloudEvent.CONTENT_TYPE)
+                .withMetadata(singletonMap(Metadata.TTL_IN_SECONDS, MESSAGE_TTL_IN_SECONDS))
+                .build()).block();
+        System.out.println("Published cloud event with message: " + cloudEvent.getData());
+        //...
+      }
+      //...
+    }
   }
-///...
 }
 ```
 
diff --git a/sdk-tests/src/test/java/io/dapr/it/pubsub/http/PubSubIT.java b/sdk-tests/src/test/java/io/dapr/it/pubsub/http/PubSubIT.java
@@ -12,6 +12,7 @@
 import io.dapr.client.domain.CloudEvent;
 import io.dapr.client.domain.HttpExtension;
 import io.dapr.client.domain.Metadata;
+import io.dapr.client.domain.PublishEventRequestBuilder;
 import io.dapr.it.BaseIT;
 import io.dapr.it.DaprRun;
 import io.dapr.serializer.DaprObjectSerializer;
@@ -26,6 +27,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
 
 import static io.dapr.it.Retry.callWithRetry;
@@ -158,24 +160,45 @@ public String getContentType() {
         System.out.println(String.format("Published message: '%s' to topic '%s' pubsub_name '%s'", message, ANOTHER_TOPIC_NAME, PUBSUB_NAME));
       }
 
+      //Publishing an object.
+      MyObject object = new MyObject();
+      object.setId("123");
+      client.publishEvent(PUBSUB_NAME, TOPIC_NAME, object).block();
+      System.out.println("Published one object.");
+
       //Publishing a single byte: Example of non-string based content published
       client.publishEvent(
           PUBSUB_NAME,
           TOPIC_NAME,
           new byte[]{1}).block();
       System.out.println("Published one byte.");
 
+      CloudEvent cloudEvent = new CloudEvent();
+      cloudEvent.setId("1234");
+      cloudEvent.setData("message from cloudevent");
+      cloudEvent.setSource("test");
+      cloudEvent.setSpecversion("1");
+      cloudEvent.setType("myevent");
+      cloudEvent.setDatacontenttype("text/plain");
+
+      //Publishing a cloud event.
+      client.publishEvent(new PublishEventRequestBuilder(PUBSUB_NAME, TOPIC_NAME, cloudEvent)
+          .withContentType("application/cloudevents+json")
+          .build()).block();
+      System.out.println("Published one cloud event.");
+
       Thread.sleep(3000);
 
       callWithRetry(() -> {
         System.out.println("Checking results for topic " + TOPIC_NAME);
+        // Validate text payload.
         final List<CloudEvent> messages = client.invokeMethod(
             daprRun.getAppName(),
             "messages/testingtopic",
             null,
             HttpExtension.GET,
             CLOUD_EVENT_LIST_TYPE_REF).block();
-        assertEquals(11, messages.size());
+        assertEquals(13, messages.size());
         for (int i = 0; i < NUM_MESSAGES; i++) {
           final int messageId = i;
           assertTrue(messages
@@ -186,13 +209,30 @@ public String getContentType() {
               .count() == 1);
         }
 
+        // Validate object payload.
+        assertTrue(messages
+            .stream()
+            .filter(m -> m.getData() != null)
+            .filter(m -> m.getData() instanceof LinkedHashMap)
+            .map(m -> (LinkedHashMap)m.getData())
+            .filter(m -> "123".equals(m.get("id")))
+            .count() == 1);
+
+        // Validate byte payload.
         assertTrue(messages
             .stream()
             .filter(m -> m.getData() != null)
             .map(m -> m.getData())
             .filter(m -> "AQ==".equals(m))
             .count() == 1);
 
+        // Validate cloudevent payload.
+        assertTrue(messages
+            .stream()
+            .filter(m -> m.getData() != null)
+            .map(m -> m.getData())
+            .filter(m -> "message from cloudevent".equals(m))
+            .count() == 1);
       }, 2000);
 
       callWithRetry(() -> {
@@ -330,4 +370,16 @@ public void testPubSubTTLMetadata() throws Exception {
 
     daprRun.stop();
   }
+
+  public static class MyObject {
+    private String id;
+
+    public String getId() {
+      return this.id;
+    }
+
+    public void setId(String id) {
+      this.id = id;
+    }
+  }
 }
diff --git a/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java b/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java
@@ -40,18 +40,15 @@
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.stub.StreamObserver;
-import okhttp3.HttpUrl;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
 import reactor.util.context.Context;
 
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
@@ -143,7 +140,15 @@ public Mono<Void> publishEvent(PublishEventRequest request) {
           .setTopic(topic)
           .setPubsubName(pubsubName)
           .setData(ByteString.copyFrom(objectSerializer.serialize(data)));
-      envelopeBuilder.setDataContentType(objectSerializer.getContentType());
+
+      // Content-type can be overwritten on a per-request basis.
+      // It allows CloudEvents to be handled differently, for example.
+      String contentType = request.getContentType();
+      if (contentType == null || contentType.isEmpty()) {
+        contentType = objectSerializer.getContentType();
+      }
+      envelopeBuilder.setDataContentType(contentType);
+
       Map<String, String> metadata = request.getMetadata();
       if (metadata != null) {
         envelopeBuilder.putAllMetadata(metadata);
diff --git a/sdk/src/main/java/io/dapr/client/DaprClientHttp.java b/sdk/src/main/java/io/dapr/client/DaprClientHttp.java
@@ -137,7 +137,13 @@ public Mono<Void> publishEvent(PublishEventRequest request) {
       }
 
       byte[] serializedEvent = objectSerializer.serialize(data);
-      Map<String, String> headers = Collections.singletonMap("content-type", objectSerializer.getContentType());
+      // Content-type can be overwritten on a per-request basis.
+      // It allows CloudEvents to be handled differently, for example.
+      String contentType = request.getContentType();
+      if (contentType == null || contentType.isEmpty()) {
+        contentType = objectSerializer.getContentType();
+      }
+      Map<String, String> headers = Collections.singletonMap("content-type", contentType);
 
       String[] pathSegments = new String[]{ DaprHttp.API_VERSION, "publish", pubsubName, topic };
 
diff --git a/sdk/src/main/java/io/dapr/client/domain/CloudEvent.java b/sdk/src/main/java/io/dapr/client/domain/CloudEvent.java
@@ -10,7 +10,6 @@
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
-import javax.annotation.PropertyKey;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Objects;
@@ -20,6 +19,11 @@
  */
 public final class CloudEvent {
 
+  /**
+   * Mime type used for CloudEvent.
+   */
+  public static final String CONTENT_TYPE = "application/cloudevents+json";
+
   /**
    * Shared Json serializer/deserializer as per Jackson's documentation.
    */
diff --git a/sdk/src/main/java/io/dapr/client/domain/PublishEventRequest.java b/sdk/src/main/java/io/dapr/client/domain/PublishEventRequest.java
@@ -18,6 +18,8 @@ public class PublishEventRequest {
 
   private Object data;
 
+  private String contentType;
+
   private Map<String, String> metadata;
 
   public String getPubsubName() {
@@ -44,6 +46,14 @@ void setData(Object data) {
     this.data = data;
   }
 
+  public String getContentType() {
+    return this.contentType;
+  }
+
+  void setContentType(String contentType) {
+    this.contentType = contentType;
+  }
+
   public Map<String, String> getMetadata() {
     return metadata;
   }
diff --git a/sdk/src/main/java/io/dapr/client/domain/PublishEventRequestBuilder.java b/sdk/src/main/java/io/dapr/client/domain/PublishEventRequestBuilder.java
@@ -20,6 +20,8 @@ public class PublishEventRequestBuilder {
 
   private final Object data;
 
+  private String contentType;
+
   private Map<String, String> metadata = new HashMap<>();
 
   /**
@@ -34,6 +36,11 @@ public PublishEventRequestBuilder(String pubsubName, String topic, Object data)
     this.data = data;
   }
 
+  public PublishEventRequestBuilder withContentType(String contentType) {
+    this.contentType = contentType;
+    return this;
+  }
+
   public PublishEventRequestBuilder withMetadata(Map<String, String> metadata) {
     this.metadata = metadata == null ? null : Collections.unmodifiableMap(metadata);
     return this;
@@ -48,6 +55,7 @@ public PublishEventRequest build() {
     request.setPubsubName(this.pubsubName);
     request.setTopic(this.topic);
     request.setData(this.data);
+    request.setContentType(this.contentType);
     request.setMetadata(this.metadata);
     return request;
   }
diff --git a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java
diff --git a/sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java b/sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java
