From 5947bbf8a37da0cd99917562064362122dc7d433 Mon Sep 17 00:00:00 2001 From: Maarten Mulders Date: Sat, 31 Jul 2021 21:49:16 +0200 Subject: [PATCH 1/6] Declare CloudEvent generic --- .../java/io/dapr/it/pubsub/http/PubSubIT.java | 24 +++++++++++++++++++ .../it/pubsub/http/SubscriberController.java | 22 +++++++++++++++++ .../io/dapr/client/domain/CloudEvent.java | 15 ++++++------ 3 files changed, 54 insertions(+), 7 deletions(-) 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 index 0cd387c1ae..9ea72dbf1b 100644 --- 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 @@ -43,6 +43,7 @@ public class PubSubIT extends BaseIT { private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); private static final TypeRef> CLOUD_EVENT_LIST_TYPE_REF = new TypeRef<>() {}; + private static final TypeRef>> CLOUD_EVENT_MYOBJECT_LIST_TYPE_REF = new TypeRef<>() {}; //Number of messages to be sent: 10 private static final int NUM_MESSAGES = 10; @@ -50,6 +51,7 @@ public class PubSubIT extends BaseIT { 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> 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 messages = client.invokeMethod( 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 eb387a8bf7..9cbc0d9be1 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 @@ -26,12 +26,19 @@ public class SubscriberController { private static final List messagesReceivedBinaryTopic = new ArrayList(); private static final List messagesReceivedAnotherTopic = new ArrayList(); private static final List messagesReceivedTTLTopic = new ArrayList(); + private static final List> typedMessagesReceivedTopic = new ArrayList<>(); @GetMapping(path = "/messages/testingtopic") public List getMessagesReceivedTestingTopic() { return messagesReceivedTestingTopic; } + @GetMapping(path = "/messages/typedtestingtopic") + public List> getMessagesReceivedTypedTestingTopic() { + System.out.println("Returning " + typedMessagesReceivedTopic.size() + " messages from typedtestingtopic: " + typedMessagesReceivedTopic); + return typedMessagesReceivedTopic; + } + @GetMapping(path = "/messages/binarytopic") public List getMessagesReceivedBinaryTopic() { return messagesReceivedBinaryTopic; @@ -62,6 +69,21 @@ public Mono handleMessage(@RequestBody(required = false) CloudEvent envelo }); } + @Topic(name = "typedtestingtopic", pubsubName = "messagebus") + @PostMapping(path = "/route1b") + public Mono handleMessageTyped(@RequestBody(required = false) CloudEvent 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); + typedMessagesReceivedTopic.add(envelope); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + @Topic(name = "binarytopic", pubsubName = "messagebus") @PostMapping(path = "/route2") public Mono handleBinaryMessage(@RequestBody(required = false) CloudEvent envelope) { 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 d303f96929..f3b9f39b68 100644 --- a/sdk/src/main/java/io/dapr/client/domain/CloudEvent.java +++ b/sdk/src/main/java/io/dapr/client/domain/CloudEvent.java @@ -16,8 +16,9 @@ /** * A cloud event in Dapr. + * @param The type of the payload. */ -public final class CloudEvent { +public final class CloudEvent { /** * 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,7 +219,7 @@ 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; } @@ -226,7 +227,7 @@ public Object getData() { * 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) From 2c9d3067cd729a1cdae6c24f235210163c521a7c Mon Sep 17 00:00:00 2001 From: Maarten Mulders Date: Sat, 31 Jul 2021 21:49:46 +0200 Subject: [PATCH 2/6] Fix timestamp in logging --- sdk-tests/src/test/resources/logback.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk-tests/src/test/resources/logback.xml b/sdk-tests/src/test/resources/logback.xml index 32b69a2602..6156c2188e 100644 --- a/sdk-tests/src/test/resources/logback.xml +++ b/sdk-tests/src/test/resources/logback.xml @@ -1,7 +1,7 @@ - %d {HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n From e65d095a052bd115a772ecec3af5ad4c2aba7e21 Mon Sep 17 00:00:00 2001 From: Maarten Mulders Date: Sat, 31 Jul 2021 21:50:44 +0200 Subject: [PATCH 3/6] Declare Spring Boot version once --- sdk-tests/pom.xml | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/sdk-tests/pom.xml b/sdk-tests/pom.xml index 2ecaedd199..48afb1fa44 100644 --- a/sdk-tests/pom.xml +++ b/sdk-tests/pom.xml @@ -31,6 +31,7 @@ 1.39.0 3.13.0 0.14.0 + 2.3.5.RELEASE @@ -112,13 +113,13 @@ org.springframework.boot spring-boot-starter-web - 2.3.5.RELEASE + ${spring-boot.version} test org.springframework.boot spring-boot-autoconfigure - 2.3.5.RELEASE + ${spring-boot.version} test From a3c830c09cc7497dcf6df79d990178a2a3a8a5df Mon Sep 17 00:00:00 2001 From: Maarten Mulders Date: Sun, 1 Aug 2021 12:20:31 +0200 Subject: [PATCH 4/6] Simplify administration of received events --- .../it/pubsub/http/SubscriberController.java | 54 +++++++------------ 1 file changed, 20 insertions(+), 34 deletions(-) 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 9cbc0d9be1..4cc9f41639 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 @@ -8,13 +8,17 @@ 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.HashMap; import java.util.List; +import java.util.Map; +import java.util.function.BiFunction; /** * SpringBoot Controller to handle input binding. @@ -22,36 +26,11 @@ @RestController public class SubscriberController { - private static final List messagesReceivedTestingTopic = new ArrayList(); - private static final List messagesReceivedBinaryTopic = new ArrayList(); - private static final List messagesReceivedAnotherTopic = new ArrayList(); - private static final List messagesReceivedTTLTopic = new ArrayList(); - private static final List> typedMessagesReceivedTopic = new ArrayList<>(); + private final Map>> messagesByTopic = new HashMap<>(); - @GetMapping(path = "/messages/testingtopic") - public List getMessagesReceivedTestingTopic() { - return messagesReceivedTestingTopic; - } - - @GetMapping(path = "/messages/typedtestingtopic") - public List> getMessagesReceivedTypedTestingTopic() { - System.out.println("Returning " + typedMessagesReceivedTopic.size() + " messages from typedtestingtopic: " + typedMessagesReceivedTopic); - return typedMessagesReceivedTopic; - } - - @GetMapping(path = "/messages/binarytopic") - public List getMessagesReceivedBinaryTopic() { - return messagesReceivedBinaryTopic; - } - - @GetMapping(path = "/messages/anothertopic") - public List getMessagesReceivedAnotherTopic() { - return messagesReceivedAnotherTopic; - } - - @GetMapping(path = "/messages/ttltopic") - public List getMessagesReceivedTTLTopic() { - return messagesReceivedTTLTopic; + @GetMapping(path = "/messages/{topic}") + public List> getMessagesByTopic(@PathVariable("topic") String topic) { + return messagesByTopic.get(topic); } @Topic(name = "testingtopic", pubsubName = "messagebus") @@ -62,7 +41,7 @@ public Mono 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); } @@ -77,7 +56,7 @@ public Mono handleMessageTyped(@RequestBody(required = false) CloudEvent

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); } @@ -106,7 +85,7 @@ public Mono 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); } @@ -120,11 +99,18 @@ public Mono 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>, List>> merge(final CloudEvent item) { + return (key, value) -> { + final List> list = value == null ? new ArrayList<>() : value; + list.add(item); + return list; + }; + } } From c7dd2702909ce117244d94a4b4ba12a9ec17acb9 Mon Sep 17 00:00:00 2001 From: Maarten Mulders Date: Sun, 1 Aug 2021 12:34:07 +0200 Subject: [PATCH 5/6] Update docs --- daprdocs/content/en/java-sdk-docs/_index.md | 2 +- .../java/io/dapr/examples/pubsub/http/CloudEventPublisher.java | 2 +- .../java/io/dapr/examples/pubsub/http/SubscriberController.java | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/daprdocs/content/en/java-sdk-docs/_index.md b/daprdocs/content/en/java-sdk-docs/_index.md index 6f49445af3..07a89e66b8 100644 --- a/daprdocs/content/en/java-sdk-docs/_index.md +++ b/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 handleMessage(@RequestBody(required = false) CloudEvent cloudEvent) { + public Mono handleMessage(@RequestBody(required = false) CloudEvent cloudEvent) { return Mono.fromRunnable(() -> { try { System.out.println("Subscriber got: " + cloudEvent.getData()); 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 index 53fb75f550..771452ed46 100644 --- a/examples/src/main/java/io/dapr/examples/pubsub/http/CloudEventPublisher.java +++ b/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 cloudEvent = new CloudEvent<>(); cloudEvent.setId(UUID.randomUUID().toString()); cloudEvent.setType("example"); cloudEvent.setSpecversion("1"); 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 83de3cba07..801068ab0d 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 @@ -28,7 +28,7 @@ public class SubscriberController { */ @Topic(name = "testingtopic", pubsubName = "${myAppProperty:messagebus}") @PostMapping(path = "/testingtopic") - public Mono handleMessage(@RequestBody(required = false) CloudEvent cloudEvent) { + public Mono handleMessage(@RequestBody(required = false) CloudEvent cloudEvent) { return Mono.fromRunnable(() -> { try { System.out.println("Subscriber got: " + cloudEvent.getData()); From 8ab9b189ce35c8cf3885d65cdbb3901ae3d8201a Mon Sep 17 00:00:00 2001 From: Maarten Mulders Date: Thu, 12 Aug 2021 10:10:34 +0200 Subject: [PATCH 6/6] Always return a list, even if it's empty Rather than returning `null` when no messages are present for a topic. --- .../test/java/io/dapr/it/pubsub/http/SubscriberController.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 4cc9f41639..619eb0f412 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 @@ -15,6 +15,7 @@ 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; @@ -30,7 +31,7 @@ public class SubscriberController { @GetMapping(path = "/messages/{topic}") public List> getMessagesByTopic(@PathVariable("topic") String topic) { - return messagesByTopic.get(topic); + return messagesByTopic.getOrDefault(topic, Collections.emptyList()); } @Topic(name = "testingtopic", pubsubName = "messagebus")