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 index ccd272efc5..8b42b295f8 100644 --- a/examples/src/main/java/io/dapr/examples/pubsub/http/Publisher.java +++ b/examples/src/main/java/io/dapr/examples/pubsub/http/Publisher.java @@ -7,10 +7,13 @@ import io.dapr.client.DaprClient; import io.dapr.client.DaprClientBuilder; +import io.dapr.client.domain.Metadata; import java.io.IOException; import java.util.Collections; +import static java.util.Collections.singletonMap; + /** * Message publisher. * 1. Build and install jars: @@ -22,12 +25,16 @@ */ public class Publisher { - //Number of messages to be sent: 10 + //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 pubseb + //The name of the pubsub private static final String PUBSUB_NAME = "messagebus"; /** @@ -41,7 +48,11 @@ public static void main(String[] args) throws Exception { for (int i = 0; i < NUM_MESSAGES; i++) { String message = String.format("This is message #%d", i); //Publishing messages - client.publishEvent(PUBSUB_NAME, TOPIC_NAME, message).block(); + client.publishEvent( + PUBSUB_NAME, + TOPIC_NAME, + message, + singletonMap(Metadata.TTL_IN_SECONDS, MESSAGE_TTL_IN_SECONDS)).block(); System.out.println("Published message: " + message); try { @@ -53,14 +64,6 @@ public static void main(String[] args) throws Exception { } } - //Publishing a single bite: Example of non-string based content published - client.publishEvent( - PUBSUB_NAME, - TOPIC_NAME, - new byte[]{1}, - Collections.singletonMap("content-type", "application/octet-stream")).block(); - System.out.println("Published one byte."); - // 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/README.md b/examples/src/main/java/io/dapr/examples/pubsub/http/README.md index c76a67c5d2..8b7933631c 100644 --- a/examples/src/main/java/io/dapr/examples/pubsub/http/README.md +++ b/examples/src/main/java/io/dapr/examples/pubsub/http/README.md @@ -135,14 +135,112 @@ dapr run --components-path ./components/pubsub --app-id publisher -- java -jar t Once running, the Publisher should print the output as follows: -![publisheroutput](../../../../../../resources/img/publisher.png) +```txt +✅ You're up and running! Both Dapr and your app logs will appear here. + +== APP == Published message: This is message #0 + +== APP == Published message: This is message #1 + +== APP == Published message: This is message #2 + +== APP == Published message: This is message #3 + +== APP == Published message: This is message #4 + +== APP == Published message: This is message #5 + +== APP == Published message: This is message #6 + +== APP == Published message: This is message #7 + +== APP == Published message: This is message #8 + +== APP == Published message: This is message #9 + +== APP == Done. + +``` Messages have been published in the topic. Once running, the Subscriber should print the output as follows: -![publisheroutput](../../../../../../resources/img/subscriber.png) +```txt +== APP == Subscriber got: {"id":"1f646657-0032-4797-b59b-c57b4f40743b","source":"publisher","type":"com.dapr.event.sent","specversion":"1.0","datacontenttype":"application/json","data":"This is message #3","expiration":"2020-12-24T05:29:12Z"} + +== APP == Subscriber got: {"id":"a22b97ce-9008-4fba-8b57-c3c3e1f031b6","source":"publisher","type":"com.dapr.event.sent","specversion":"1.0","datacontenttype":"application/json","data":"This is message #8","expiration":"2020-12-24T05:29:15Z"} + +== APP == Subscriber got: {"id":"abb2f110-6862-49f7-8c8d-189f6dcd177d","source":"publisher","type":"com.dapr.event.sent","specversion":"1.0","datacontenttype":"application/json","data":"This is message #0","expiration":"2020-12-24T05:29:11Z"} + +== APP == Subscriber got: {"id":"043f31d3-c13a-4a02-ac89-64ecca946598","source":"publisher","type":"com.dapr.event.sent","specversion":"1.0","datacontenttype":"application/json","data":"This is message #7","expiration":"2020-12-24T05:29:14Z"} + +== APP == Subscriber got: {"id":"acc554f4-7109-4c31-9374-0e5936b90180","source":"publisher","type":"com.dapr.event.sent","specversion":"1.0","datacontenttype":"application/json","data":"This is message #2","expiration":"2020-12-24T05:29:12Z"} + +== APP == Subscriber got: {"id":"8b3ad160-368d-4b0f-9925-8fa2a2fbf5ca","source":"publisher","type":"com.dapr.event.sent","specversion":"1.0","datacontenttype":"application/json","data":"This is message #9","expiration":"2020-12-24T05:29:15Z"} + +== APP == Subscriber got: {"id":"e41d4512-511a-4a2b-80f3-a0a4d091c9a5","source":"publisher","type":"com.dapr.event.sent","specversion":"1.0","datacontenttype":"application/json","data":"This is message #1","expiration":"2020-12-24T05:29:11Z"} + +== APP == Subscriber got: {"id":"33e21664-128e-4fc4-b5c4-ed257f758336","source":"publisher","type":"com.dapr.event.sent","specversion":"1.0","datacontenttype":"application/json","data":"This is message #4","expiration":"2020-12-24T05:29:13Z"} + +== APP == Subscriber got: {"id":"bd14f1ee-ca6b-47f7-8130-dd1e6de5b03c","source":"publisher","type":"com.dapr.event.sent","specversion":"1.0","datacontenttype":"application/json","data":"This is message #6","expiration":"2020-12-24T05:29:14Z"} + +== APP == Subscriber got: {"id":"acc57cd6-71da-4ba3-9a12-9c921ca49af7","source":"publisher","type":"com.dapr.event.sent","specversion":"1.0","datacontenttype":"application/json","data":"This is message #5","expiration":"2020-12-24T05:29:13Z"} + +``` Messages have been retrieved from the topic. +### Message expiration (Optional) + +Optionally, you can see how Dapr can automatically drop expired messages on behalf of the subscriber. +First, make sure the publisher and the subscriber applications are stopped. +Then, change the TTL constant in the `Publisher.java` file from: +```java +private static final String MESSAGE_TTL_IN_SECONDS = "1000"; +``` +To: +```java +private static final String MESSAGE_TTL_IN_SECONDS = "1"; +``` + +Now rebuild the example: +```sh +mvn install +``` + +Run the publisher app: +```sh +dapr run --components-path ./components/pubsub --app-id publisher -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.http.Publisher +``` + +Wait until all 10 messages are published like before, then wait for a few more seconds and run the subscriber app: +```sh +dapr run --components-path ./components/pubsub --app-id subscriber --app-port 3000 -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.http.Subscriber -p 3000 +``` + +No message is consumed by the subscriber app and warnings messages are emitted from Dapr sidecar: +```txt +== DAPR == time="2020-12-23T21:21:59.085797-08:00" level=warning msg="dropping expired pub/sub event 461546c1-d2df-42bd-a6b8-3beeb952fe1e as of 2020-12-24T05:21:50Z" app_id=subscriber instance=myhost scope=dapr.runtime type=log ver=edge + +== DAPR == time="2020-12-23T21:21:59.085841-08:00" level=warning msg="dropping expired pub/sub event 2d8cf9a6-4019-4dda-95fd-59218a19381b as of 2020-12-24T05:21:48Z" app_id=subscriber instance=myhost scope=dapr.runtime type=log ver=edge + +== DAPR == time="2020-12-23T21:21:59.085871-08:00" level=warning msg="dropping expired pub/sub event d2a199e0-a4b8-4067-9618-6688391ad68f as of 2020-12-24T05:21:53Z" app_id=subscriber instance=myhost scope=dapr.runtime type=log ver=edge + +== DAPR == time="2020-12-23T21:21:59.085894-08:00" level=warning msg="dropping expired pub/sub event 30719f17-ad8f-4dea-91b5-b77958f360d4 as of 2020-12-24T05:21:49Z" app_id=subscriber instance=myhost scope=dapr.runtime type=log ver=edge + +== DAPR == time="2020-12-23T21:21:59.085797-08:00" level=warning msg="dropping expired pub/sub event d136d5ae-5561-418c-a850-9d1698bc8840 as of 2020-12-24T05:21:51Z" app_id=subscriber instance=myhost scope=dapr.runtime type=log ver=edge + +== DAPR == time="2020-12-23T21:21:59.085958-08:00" level=warning msg="dropping expired pub/sub event 82b334a2-e295-48ea-8c6c-c45b1c4fcd2d as of 2020-12-24T05:21:50Z" app_id=subscriber instance=myhost scope=dapr.runtime type=log ver=edge + +== DAPR == time="2020-12-23T21:21:59.085973-08:00" level=warning msg="dropping expired pub/sub event f6eb3f9f-185f-492f-9df9-45af8c91932b as of 2020-12-24T05:21:53Z" app_id=subscriber instance=myhost scope=dapr.runtime type=log ver=edge + +== DAPR == time="2020-12-23T21:21:59.086041-08:00" level=warning msg="dropping expired pub/sub event a536eb9f-34e0-49fc-ba29-a34854398d96 as of 2020-12-24T05:21:52Z" app_id=subscriber instance=myhost scope=dapr.runtime type=log ver=edge + +== DAPR == time="2020-12-23T21:21:59.085995-08:00" level=warning msg="dropping expired pub/sub event 52cc9528-f9d4-44f4-8f78-8f32341a743a as of 2020-12-24T05:21:49Z" app_id=subscriber instance=myhost scope=dapr.runtime type=log ver=edge + +== DAPR == time="2020-12-23T21:21:59.085797-08:00" level=warning msg="dropping expired pub/sub event 7cf927e8-e832-4f8a-911a-1cae5a1369d2 as of 2020-12-24T05:21:48Z" app_id=subscriber instance=myhost scope=dapr.runtime type=log ver=edge + +``` + For more details on Dapr Spring Boot integration, please refer to [Dapr Spring Boot](../../../springboot/DaprApplication.java) Application implementation. diff --git a/examples/src/main/java/io/dapr/examples/pubsub/http/SubscriberController.java b/examples/src/main/java/io/dapr/examples/pubsub/http/SubscriberController.java index 18f32dcd84..2db835f0ae 100644 --- a/examples/src/main/java/io/dapr/examples/pubsub/http/SubscriberController.java +++ b/examples/src/main/java/io/dapr/examples/pubsub/http/SubscriberController.java @@ -5,40 +5,33 @@ package io.dapr.examples.pubsub.http; +import com.fasterxml.jackson.databind.ObjectMapper; import io.dapr.Topic; import io.dapr.client.domain.CloudEvent; -import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; -import org.springframework.web.bind.annotation.RequestHeader; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Mono; -import java.util.Map; - /** * SpringBoot Controller to handle input binding. */ @RestController public class SubscriberController { + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + /** * Handles a registered publish endpoint on this app. - * @param body The body of the http message. - * @param headers The headers of the http message. + * @param cloudEvent The cloud event received. * @return A message containing the time. */ @Topic(name = "testingtopic", pubsubName = "messagebus") @PostMapping(path = "/testingtopic") - public Mono handleMessage(@RequestBody(required = false) byte[] body, - @RequestHeader Map headers) { + public Mono handleMessage(@RequestBody(required = false) CloudEvent cloudEvent) { return Mono.fromRunnable(() -> { try { - // Dapr's event is compliant to CloudEvent. - CloudEvent envelope = CloudEvent.deserialize(body); - - String message = envelope.getData() == null ? "" : envelope.getData(); - System.out.println("Subscriber got message: " + message); + System.out.println("Subscriber got: " + OBJECT_MAPPER.writeValueAsString(cloudEvent)); } catch (Exception e) { throw new RuntimeException(e); } diff --git a/examples/src/main/java/io/dapr/springboot/OpenTelemetryConfig.java b/examples/src/main/java/io/dapr/springboot/OpenTelemetryConfig.java index 0be1761921..a607442bf0 100644 --- a/examples/src/main/java/io/dapr/springboot/OpenTelemetryConfig.java +++ b/examples/src/main/java/io/dapr/springboot/OpenTelemetryConfig.java @@ -10,7 +10,6 @@ import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.api.trace.propagation.HttpTraceContext; import io.opentelemetry.context.propagation.DefaultContextPropagators; -import io.opentelemetry.exporter.logging.LoggingSpanExporter; import io.opentelemetry.exporter.zipkin.ZipkinSpanExporter; import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; @@ -64,10 +63,6 @@ public static Tracer createTracer(String instrumentationName) { System.out.println("WARNING: Zipkin is not available."); } - final LoggingSpanExporter loggingExporter = new LoggingSpanExporter(); - OpenTelemetrySdk.getGlobalTracerManagement() - .addSpanProcessor(SimpleSpanProcessor.builder(loggingExporter).build()); - return tracer; } diff --git a/examples/src/main/resources/img/publisher.png b/examples/src/main/resources/img/publisher.png deleted file mode 100644 index cea3d8e60f..0000000000 Binary files a/examples/src/main/resources/img/publisher.png and /dev/null differ diff --git a/examples/src/main/resources/img/subscriber.png b/examples/src/main/resources/img/subscriber.png deleted file mode 100644 index 506456d986..0000000000 Binary files a/examples/src/main/resources/img/subscriber.png and /dev/null differ diff --git a/pom.xml b/pom.xml index fabdb7b0f9..76e32ddbdf 100644 --- a/pom.xml +++ b/pom.xml @@ -16,7 +16,7 @@ UTF-8 1.33.1 3.13.0 - https://raw.githubusercontent.com/dapr/dapr/3cca3cc1567f1cb955ae69b8fd784f075f62ad42/dapr/proto + https://raw.githubusercontent.com/dapr/dapr/33dc3361e99f75e88628e0abfc8b04d3d7e8bc5f/dapr/proto 1.6.2 3.1.1 1.8 diff --git a/sdk-tests/src/test/java/io/dapr/it/pubsub/http/SubscriberController.java b/sdk-tests/src/test/java/io/dapr/it/pubsub/http/SubscriberController.java index ed2fe08c7d..2d62ee4974 100644 --- a/sdk-tests/src/test/java/io/dapr/it/pubsub/http/SubscriberController.java +++ b/sdk-tests/src/test/java/io/dapr/it/pubsub/http/SubscriberController.java @@ -21,29 +21,25 @@ @RestController public class SubscriberController { - private static final List messagesReceivedTestingTopic = new ArrayList(); - private static final List messagesReceivedAnotherTopic = new ArrayList(); + private static final List messagesReceivedTestingTopic = new ArrayList(); + private static final List messagesReceivedAnotherTopic = new ArrayList(); @GetMapping(path = "/messages/testingtopic") - public List getMessagesReceivedTestingTopic() { + public List getMessagesReceivedTestingTopic() { return messagesReceivedTestingTopic; } @GetMapping(path = "/messages/anothertopic") - public List getMessagesReceivedAnotherTopic() { + public List getMessagesReceivedAnotherTopic() { return messagesReceivedAnotherTopic; } @Topic(name = "testingtopic", pubsubName = "messagebus") @PostMapping(path = "/route1") - public Mono handleMessage(@RequestBody(required = false) byte[] body, - @RequestHeader Map headers) { + public Mono handleMessage(@RequestBody(required = false) CloudEvent envelope) { return Mono.fromRunnable(() -> { try { - // Dapr's event is compliant to CloudEvent. - CloudEvent envelope = CloudEvent.deserialize(body); - - String message = envelope.getData() == null ? "" : envelope.getData(); + String message = envelope.getData() == null ? "" : envelope.getData().toString(); System.out.println("Testing topic Subscriber got message: " + message); messagesReceivedTestingTopic.add(envelope.getData()); } catch (Exception e) { @@ -54,14 +50,10 @@ public Mono handleMessage(@RequestBody(required = false) byte[] body, @Topic(name = "anothertopic", pubsubName = "messagebus") @PostMapping(path = "/route2") - public Mono handleMessageAnotherTopic(@RequestBody(required = false) byte[] body, - @RequestHeader Map headers) { + public Mono handleMessageAnotherTopic(@RequestBody(required = false) CloudEvent envelope) { return Mono.fromRunnable(() -> { try { - // Dapr's event is compliant to CloudEvent. - CloudEvent envelope = CloudEvent.deserialize(body); - - String message = envelope.getData() == null ? "" : envelope.getData(); + String message = envelope.getData() == null ? "" : envelope.getData().toString(); System.out.println("Another topic Subscriber got message: " + message); messagesReceivedAnotherTopic.add(envelope.getData()); } catch (Exception e) { diff --git a/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java b/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java index 79d455d7bc..1d73b264db 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java @@ -140,16 +140,22 @@ public Mono> publishEvent(PublishEventRequest request) { String pubsubName = request.getPubsubName(); String topic = request.getTopic(); Object data = request.getData(); - // TODO(artursouza): handle metadata. - // Map metadata = request.getMetadata(); Context context = request.getContext(); - DaprProtos.PublishEventRequest envelope = DaprProtos.PublishEventRequest.newBuilder() + DaprProtos.PublishEventRequest.Builder envelopeBuilder = DaprProtos.PublishEventRequest.newBuilder() .setTopic(topic) .setPubsubName(pubsubName) - .setData(ByteString.copyFrom(objectSerializer.serialize(data))).build(); + .setData(ByteString.copyFrom(objectSerializer.serialize(data))); + Map metadata = request.getMetadata(); + if (metadata != null) { + envelopeBuilder.putAllMetadata(metadata); + String contentType = metadata.get(io.dapr.client.domain.Metadata.CONTENT_TYPE); + if (contentType != null) { + envelopeBuilder.setDataContentType(contentType); + } + } return Mono.fromCallable(wrap(context, () -> { - get(client.publishEvent(envelope)); + get(client.publishEvent(envelopeBuilder.build())); return null; })); } catch (Exception ex) { diff --git a/sdk/src/main/java/io/dapr/client/DaprHttp.java b/sdk/src/main/java/io/dapr/client/DaprHttp.java index 6b1b3d1c6d..5b91de2d0e 100644 --- a/sdk/src/main/java/io/dapr/client/DaprHttp.java +++ b/sdk/src/main/java/io/dapr/client/DaprHttp.java @@ -7,6 +7,7 @@ import com.fasterxml.jackson.core.JsonParseException; import com.fasterxml.jackson.databind.ObjectMapper; +import io.dapr.client.domain.Metadata; import io.dapr.config.Properties; import io.dapr.exceptions.DaprError; import io.dapr.exceptions.DaprException; @@ -242,7 +243,7 @@ private Response doInvokeApi(String method, final String requestId = UUID.randomUUID().toString(); RequestBody body; - String contentType = headers != null ? headers.get("content-type") : null; + String contentType = headers != null ? headers.get(Metadata.CONTENT_TYPE) : null; MediaType mediaType = contentType == null ? MEDIA_TYPE_APPLICATION_JSON : MediaType.get(contentType); if (content == null) { body = mediaType.equals(MEDIA_TYPE_APPLICATION_JSON) diff --git a/sdk/src/main/java/io/dapr/client/domain/CloudEvent.java b/sdk/src/main/java/io/dapr/client/domain/CloudEvent.java index 756404d42e..d57b325ff5 100644 --- a/sdk/src/main/java/io/dapr/client/domain/CloudEvent.java +++ b/sdk/src/main/java/io/dapr/client/domain/CloudEvent.java @@ -7,7 +7,6 @@ import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.databind.DeserializationFeature; -import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import java.io.IOException; @@ -28,36 +27,41 @@ public final class CloudEvent { /** * Identifier of the message being processed. */ - private final String id; + private String id; /** * Event's source. */ - private final String source; + private String source; /** * Envelope type. */ - private final String type; + private String type; /** * Version of the specification. */ - private final String specversion; + private String specversion; /** * Type of the data's content. */ - private final String datacontenttype; + private String datacontenttype; /** * Cloud event specs says data can be a JSON object or string. */ - private final String data; + private Object data; /** - * Instantiates a new input request. - * + * Instantiates a CloudEvent. + */ + public CloudEvent() { + } + + /** + * Instantiates a CloudEvent. * @param id Identifier of the message being processed. * @param source Source for this event. * @param type Type of event. @@ -71,7 +75,7 @@ public CloudEvent( String type, String specversion, String datacontenttype, - String data) { + Object data) { this.id = id; this.source = source; this.type = type; @@ -80,9 +84,24 @@ public CloudEvent( this.data = data; } + /** - * Gets the identifier of the message being processed. + * Deserialize a message topic from Dapr. * + * @param payload Payload sent from Dapr. + * @return Message (can be null if input is null) + * @throws IOException If cannot parse. + */ + public static CloudEvent deserialize(byte[] payload) throws IOException { + if (payload == null) { + return null; + } + + return OBJECT_MAPPER.readValue(payload, CloudEvent.class); + } + + /** + * Gets the identifier of the message being processed. * @return Identifier of the message being processed. */ public String getId() { @@ -90,60 +109,104 @@ public String getId() { } /** - * Gets the source for this event. - * - * @return Source for this event. + * Sets the identifier of the message being processed. + * @param id Identifier of the message being processed. + */ + public void setId(String id) { + this.id = id; + } + + /** + * Gets the event's source. + * @return Event's source. */ public String getSource() { return source; } /** - * Gets the type of event. - * - * @return Type of event. + * Sets the event's source. + * @param source Event's source. + */ + public void setSource(String source) { + this.source = source; + } + + /** + * Gets the envelope type. + * @return Envelope type. */ public String getType() { return type; } /** - * Gets the version of the event spec. - * - * @return Version of the event spec. + * Sets the envelope type. + * @param type Envelope type. + */ + public void setType(String type) { + this.type = type; + } + + /** + * Gets the version of the specification. + * @return Version of the specification. */ public String getSpecversion() { return specversion; } /** - * Gets the type of the payload. - * - * @return Type of the payload. + * Sets the version of the specification. + * @param specversion Version of the specification. + */ + public void setSpecversion(String specversion) { + this.specversion = specversion; + } + + /** + * Gets the type of the data's content. + * @return Type of the data's content. */ public String getDatacontenttype() { return datacontenttype; } /** - * Gets the payload. - * - * @return Payload + * Sets the type of the data's content. + * @param datacontenttype Type of the data's content. */ - public String getData() { + public void setDatacontenttype(String datacontenttype) { + this.datacontenttype = 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() { 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) { + this.data = data; + } + + /** + * {@inheritDoc} + */ @Override public boolean equals(Object o) { if (this == o) { return true; } - if (o == null || getClass() != o.getClass()) { return false; } - CloudEvent that = (CloudEvent) o; return Objects.equals(id, that.id) && Objects.equals(source, that.source) @@ -153,64 +216,11 @@ public boolean equals(Object o) { && Objects.equals(data, that.data); } + /** + * {@inheritDoc} + */ @Override public int hashCode() { return Objects.hash(id, source, type, specversion, datacontenttype, data); } - - /** - * Deserialized a message topic from Dapr. - * - * @param payload Payload sent from Dapr. - * @return Message (can be null if input is null) - * @throws IOException If cannot parse. - */ - public static CloudEvent deserialize(byte[] payload) throws IOException { - if (payload == null) { - return null; - } - - JsonNode node = OBJECT_MAPPER.readTree(payload); - - if (node == null) { - return null; - } - - String id = null; - if (node.has("id") && !node.get("id").isNull()) { - id = node.get("id").asText(); - } - - String source = null; - if (node.has("source") && !node.get("source").isNull()) { - source = node.get("source").asText(); - } - - String type = null; - if (node.has("type") && !node.get("type").isNull()) { - type = node.get("type").asText(); - } - - String specversion = null; - if (node.has("specversion") && !node.get("specversion").isNull()) { - specversion = node.get("specversion").asText(); - } - - String datacontenttype = null; - if (node.has("datacontenttype") && !node.get("datacontenttype").isNull()) { - datacontenttype = node.get("datacontenttype").asText(); - } - - String data = null; - if (node.has("data") && !node.get("data").isNull()) { - JsonNode dataNode = node.get("data"); - if (dataNode.isTextual()) { - data = dataNode.textValue(); - } else { - data = node.get("data").toString(); - } - } - - return new CloudEvent(id, source, type, specversion, datacontenttype, data); - } } diff --git a/sdk/src/main/java/io/dapr/client/domain/Metadata.java b/sdk/src/main/java/io/dapr/client/domain/Metadata.java new file mode 100644 index 0000000000..1d4a1bddc0 --- /dev/null +++ b/sdk/src/main/java/io/dapr/client/domain/Metadata.java @@ -0,0 +1,19 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.client.domain; + +/** + * Enumerates commonly used metadata attributes. + */ +public final class Metadata { + + public static final String CONTENT_TYPE = "content-type"; + + public static final String TTL_IN_SECONDS = "ttlInSeconds"; + + private Metadata() { + } +} diff --git a/sdk/src/test/java/io/dapr/serializer/DefaultObjectSerializerTest.java b/sdk/src/test/java/io/dapr/serializer/DefaultObjectSerializerTest.java index 0a380bdb83..9455a65739 100644 --- a/sdk/src/test/java/io/dapr/serializer/DefaultObjectSerializerTest.java +++ b/sdk/src/test/java/io/dapr/serializer/DefaultObjectSerializerTest.java @@ -14,7 +14,9 @@ import java.io.Serializable; import java.util.ArrayList; import java.util.Base64; +import java.util.HashMap; import java.util.List; +import java.util.TreeMap; import java.util.function.Function; import static org.junit.Assert.assertEquals; @@ -780,7 +782,7 @@ public void serializeDeserializeCloudEventEnvelope() throws Exception { public void deserializeCloudEventEnvelopeData() throws Exception { - Function deserializeData = (jsonData -> { + Function deserializeData = (jsonData -> { try { String payload = String.format("{\"data\": %s}", jsonData); return CloudEvent.deserialize(payload.getBytes()).getData(); @@ -789,26 +791,28 @@ public void deserializeCloudEventEnvelopeData() throws Exception { } }); - assertEquals("123", - deserializeData.apply("123")); - assertEquals("true", - deserializeData.apply("true")); - assertEquals("123.45", - deserializeData.apply("123.45")); + assertEquals(123, + deserializeData.apply("123")); + assertEquals(true, + deserializeData.apply("true")); + assertEquals(123.45, + deserializeData.apply("123.45")); assertEquals("AAEI", - deserializeData.apply(quote(Base64.getEncoder().encodeToString(new byte[] { 0, 1, 8})))); + deserializeData.apply(quote(Base64.getEncoder().encodeToString(new byte[]{0, 1, 8})))); assertEquals("hello world", - deserializeData.apply(quote("hello world"))); + deserializeData.apply(quote("hello world"))); assertEquals("\"hello world\"", - deserializeData.apply(quote("\\\"hello world\\\""))); + deserializeData.apply(quote("\\\"hello world\\\""))); assertEquals("\"hello world\"", - deserializeData.apply(new ObjectMapper().writeValueAsString("\"hello world\""))); + deserializeData.apply(new ObjectMapper().writeValueAsString("\"hello world\""))); assertEquals("hello world", - deserializeData.apply(new ObjectMapper().writeValueAsString("hello world"))); - assertEquals("{\"id\":\"123:\",\"name\":\"Jon Doe\"}", - deserializeData.apply("{\"id\": \"123:\", \"name\": \"Jon Doe\"}")); - assertEquals("{\"id\": \"123:\", \"name\": \"Jon Doe\"}", - deserializeData.apply(new ObjectMapper().writeValueAsString("{\"id\": \"123:\", \"name\": \"Jon Doe\"}"))); + deserializeData.apply(new ObjectMapper().writeValueAsString("hello world"))); + assertEquals(new TreeMap() {{ + put("id", "123"); + put("name", "Jon Doe"); + }}, deserializeData.apply("{\"id\": \"123\", \"name\": \"Jon Doe\"}")); + assertEquals("{\"id\": \"123\", \"name\": \"Jon Doe\"}", + deserializeData.apply(new ObjectMapper().writeValueAsString("{\"id\": \"123\", \"name\": \"Jon Doe\"}"))); } @Test