From 159c0aad3d9f8a994e43b9ffc55b1be15db778e0 Mon Sep 17 00:00:00 2001 From: Phil Kedy Date: Fri, 13 Aug 2021 13:55:40 -0400 Subject: [PATCH 1/3] Adding pubsub routing support Signed-off-by: Artur Souza --- daprdocs/content/en/java-sdk-docs/_index.md | 14 +++ .../pubsub/http/SubscriberController.java | 20 ++++ .../springboot/DaprBeanPostProcessor.java | 12 +- .../java/io/dapr/springboot/DaprRuntime.java | 42 ++++--- .../springboot/DaprSubscriptionBuilder.java | 108 ++++++++++++++++++ .../java/io/dapr/springboot/DaprTopicKey.java | 43 +++++++ .../io/dapr/springboot/DaprTopicRoutes.java | 30 +++++ .../io/dapr/springboot/DaprTopicRule.java | 43 +++++++ .../springboot/DaprTopicSubscription.java | 10 +- sdk-tests/components/kafka_bindings.yaml | 2 + sdk-tests/configuration.yaml | 8 ++ sdk-tests/deploy/start-dockers.sh | 3 + .../src/test/java/io/dapr/it/DaprRun.java | 2 +- .../java/io/dapr/it/pubsub/http/PubSubIT.java | 54 ++++++++- .../it/pubsub/http/SubscriberController.java | 54 ++++++++- sdk/src/main/java/io/dapr/Rule.java | 30 +++++ sdk/src/main/java/io/dapr/Topic.java | 6 + 17 files changed, 458 insertions(+), 23 deletions(-) create mode 100644 sdk-springboot/src/main/java/io/dapr/springboot/DaprSubscriptionBuilder.java create mode 100644 sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicKey.java create mode 100644 sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRoutes.java create mode 100644 sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRule.java create mode 100644 sdk-tests/configuration.yaml create mode 100755 sdk-tests/deploy/start-dockers.sh create mode 100644 sdk/src/main/java/io/dapr/Rule.java diff --git a/daprdocs/content/en/java-sdk-docs/_index.md b/daprdocs/content/en/java-sdk-docs/_index.md index 438e297389..d4fb7be7b9 100644 --- a/daprdocs/content/en/java-sdk-docs/_index.md +++ b/daprdocs/content/en/java-sdk-docs/_index.md @@ -172,6 +172,20 @@ public class SubscriberController { }); } + @Topic(name = "testingtopic", pubsubName = "${myAppProperty:messagebus}", + rule = @Rule(match = "event.type == 'myevent.v2'", priority = 1)) + @PostMapping(path = "/testingtopicV2") + public Mono handleMessageV2(@RequestBody(required = false) CloudEvent envelope) { + return Mono.fromRunnable(() -> { + try { + System.out.println("Subscriber got: " + cloudEvent.getData()); + 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/examples/pubsub/http/SubscriberController.java b/examples/src/main/java/io/dapr/examples/pubsub/http/SubscriberController.java index fc0f09527f..19cb26a918 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 @@ -14,6 +14,7 @@ package io.dapr.examples.pubsub.http; import com.fasterxml.jackson.databind.ObjectMapper; +import io.dapr.Rule; import io.dapr.Topic; import io.dapr.client.domain.CloudEvent; import org.springframework.web.bind.annotation.PostMapping; @@ -47,4 +48,23 @@ public Mono handleMessage(@RequestBody(required = false) CloudEvent handleMessageV2(@RequestBody(required = false) CloudEvent cloudEvent) { + return Mono.fromRunnable(() -> { + try { + System.out.println("Subscriber got: " + cloudEvent.getData()); + System.out.println("Subscriber got: " + OBJECT_MAPPER.writeValueAsString(cloudEvent)); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + } diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprBeanPostProcessor.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprBeanPostProcessor.java index cc591184a2..6afd1ba9aa 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprBeanPostProcessor.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprBeanPostProcessor.java @@ -16,6 +16,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; +import io.dapr.Rule; import io.dapr.Topic; import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanPostProcessor; @@ -26,7 +27,6 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; -import java.lang.annotation.Annotation; import java.lang.reflect.Method; import java.util.ArrayList; import java.util.HashMap; @@ -85,8 +85,11 @@ private static void subscribeToTopics(Class clazz, EmbeddedValueResolver embedde continue; } + + Rule rule = topic.rule(); String topicName = embeddedValueResolver.resolveStringValue(topic.name()); String pubSubName = embeddedValueResolver.resolveStringValue(topic.pubsubName()); + String match = embeddedValueResolver.resolveStringValue(rule.match()); if ((topicName != null) && (topicName.length() > 0) && pubSubName != null && pubSubName.length() > 0) { try { TypeReference> typeRef @@ -94,12 +97,11 @@ private static void subscribeToTopics(Class clazz, EmbeddedValueResolver embedde Map metadata = MAPPER.readValue(topic.metadata(), typeRef); List routes = getAllCompleteRoutesForPost(clazz, method, topicName); for (String route : routes) { - DaprRuntime.getInstance().addSubscribedTopic(pubSubName, topicName, route, - metadata); + DaprRuntime.getInstance().addSubscribedTopic( + pubSubName, topicName, match, rule.priority(), route, metadata); } - } catch (JsonProcessingException e) { - throw new IllegalArgumentException("Error while parsing metadata: " + e.toString()); + throw new IllegalArgumentException("Error while parsing metadata: " + e); } } } diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprRuntime.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprRuntime.java index abb5898d20..298b8bca34 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprRuntime.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprRuntime.java @@ -13,11 +13,12 @@ package io.dapr.springboot; -import java.util.ArrayList; -import java.util.HashSet; +import io.dapr.Rule; + +import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Set; +import java.util.stream.Collectors; /** * Internal Singleton to handle Dapr configuration. @@ -30,14 +31,9 @@ class DaprRuntime { private static volatile DaprRuntime instance; /** - * List of subscribed topics. + * Map of subscription builders. */ - private final Set subscribedTopics = new HashSet<>(); - - /** - * List of subscriptions. - */ - private final List subscriptions = new ArrayList<>(); + private final Map subscriptionBuilders = new HashMap<>(); /** * Private constructor to make this singleton. @@ -67,20 +63,38 @@ public static DaprRuntime getInstance() { * * @param pubsubName Pubsub name to subcribe to. * @param topicName Name of the topic being subscribed to. + * @param rule The optional rule for this route. * @param route Destination route for requests. * @param metadata Metadata for extended subscription functionality. */ public synchronized void addSubscribedTopic(String pubsubName, String topicName, + String match, + int priority, String route, Map metadata) { - if (!this.subscribedTopics.contains(topicName)) { - this.subscribedTopics.add(topicName); - this.subscriptions.add(new DaprTopicSubscription(pubsubName, topicName, route, metadata)); + DaprTopicKey topicKey = new DaprTopicKey(pubsubName, topicName); + + DaprSubscriptionBuilder builder = subscriptionBuilders.get(topicKey); + if (builder == null) { + builder = new DaprSubscriptionBuilder(pubsubName, topicName); + subscriptionBuilders.put(topicKey, builder); + } + + if (match.length() > 0) { + builder.addRule(route, match, priority); + } else { + builder.setDefaultPath(route); + } + + if (metadata != null && !metadata.isEmpty()) { + builder.setMetadata(metadata); } } public synchronized DaprTopicSubscription[] listSubscribedTopics() { - return this.subscriptions.toArray(new DaprTopicSubscription[0]); + List values = subscriptionBuilders.values().stream() + .map(b -> b.build()).collect(Collectors.toList()); + return values.toArray(new DaprTopicSubscription[0]); } } diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprSubscriptionBuilder.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprSubscriptionBuilder.java new file mode 100644 index 0000000000..1974e02656 --- /dev/null +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprSubscriptionBuilder.java @@ -0,0 +1,108 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr.springboot; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +public class DaprSubscriptionBuilder { + private final String pubsubName; + private final String topic; + private final List rules; + private String defaultPath; + private Map metadata; + + /** + * Create a subscription topic. + * @param pubsubName The pubsub name to subscribe to. + * @param topic The topic to subscribe to. + */ + public DaprSubscriptionBuilder(String pubsubName, String topic) { + this.pubsubName = pubsubName; + this.topic = topic; + this.rules = new ArrayList<>(); + this.defaultPath = null; + this.metadata = Collections.emptyMap(); + } + + /** + * Sets the default path for the subscription. + * @param path The default path. + */ + public void setDefaultPath(String path) { + if (defaultPath != null) { + throw new RuntimeException( + String.format( + "a default route is already set for topic %s on pubsub %s", + this.topic, this.pubsubName)); + } + defaultPath = path; + } + + /** + * Adds a rule to the subscription. + * @param path The path to route to. + * @param match The CEL expression the event must match. + * @param priority The priority of the rule. + */ + public void addRule(String path, String match, int priority) { + if (rules.stream().anyMatch(e -> e.getPriority() == priority)) { + throw new RuntimeException( + String.format( + "a rule priority of %d is already used for topic %s on pubsub %s", + priority, this.topic, this.pubsubName)); + } + rules.add(new TopicRule(path, match, priority)); + } + + /** + * Sets the metadata for the subscription. + * @param metadata The metadata. + */ + public void setMetadata(Map metadata) { + this.metadata = metadata; + } + + /** + * Builds the DaprTopicSubscription that is returned by the application to Dapr. + * @return The DaprTopicSubscription. + */ + public DaprTopicSubscription build() { + String route = null; + DaprTopicRoutes routes = null; + + if (!rules.isEmpty()) { + Collections.sort(rules, Comparator.comparingInt(TopicRule::getPriority)); + List topicRules = rules.stream() + .map(e -> new DaprTopicRule(e.match, e.path)).collect(Collectors.toList()); + routes = new DaprTopicRoutes(topicRules, defaultPath); + } else { + route = defaultPath; + } + + return new DaprTopicSubscription(this.pubsubName, this.topic, route, routes, metadata); + } + + static class TopicRule { + private final String path; + private final String match; + private final int priority; + + public TopicRule(String path, String match, int priority) { + this.path = path; + this.match = match; + this.priority = priority; + } + + public int getPriority() { + return priority; + } + } +} diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicKey.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicKey.java new file mode 100644 index 0000000000..b1e9c4ef3b --- /dev/null +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicKey.java @@ -0,0 +1,43 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr.springboot; + +import java.util.Objects; + +public class DaprTopicKey { + private final String pubsubName; + private final String topic; + + public DaprTopicKey(String pubsubName, String topic) { + this.pubsubName = pubsubName; + this.topic = topic; + } + + public String getPubsubName() { + return pubsubName; + } + + public String getTopic() { + return topic; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + DaprTopicKey that = (DaprTopicKey) o; + return pubsubName.equals(that.pubsubName) && topic.equals(that.topic); + } + + @Override + public int hashCode() { + return Objects.hash(pubsubName, topic); + } +} diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRoutes.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRoutes.java new file mode 100644 index 0000000000..e78d06dab3 --- /dev/null +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRoutes.java @@ -0,0 +1,30 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr.springboot; + +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.util.List; +import java.util.Optional; + +public class DaprTopicRoutes { + private final List rules; + @JsonProperty("default") + private final String defaultRoute; + + public DaprTopicRoutes(List rules, String defaultRoute) { + this.rules = rules; + this.defaultRoute = defaultRoute; + } + + public List getRules() { + return rules; + } + + public String getDefaultRoute() { + return defaultRoute; + } +} diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRule.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRule.java new file mode 100644 index 0000000000..5fc8b905e4 --- /dev/null +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRule.java @@ -0,0 +1,43 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr.springboot; + +import java.util.Objects; + +public class DaprTopicRule { + private final String match; + private final String path; + + public DaprTopicRule(String match, String path) { + this.match = match; + this.path = path; + } + + public String getMatch() { + return match; + } + + public String getPath() { + return path; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + DaprTopicRule that = (DaprTopicRule) o; + return match.equals(that.match) && path.equals(that.path); + } + + @Override + public int hashCode() { + return Objects.hash(match, path); + } +} diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicSubscription.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicSubscription.java index b55d586f3e..956c0bbcb8 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicSubscription.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicSubscription.java @@ -23,6 +23,7 @@ public class DaprTopicSubscription { private final String pubsubName; private final String topic; private final String route; + private final DaprTopicRoutes routes; private final Map metadata; /** @@ -30,12 +31,15 @@ public class DaprTopicSubscription { * @param pubsubName The pubsub name to subscribe to. * @param topic The topic to subscribe to. * @param route Destination route for messages. + * @param routes Destination routes with rules for messages. * @param metadata Metdata for extended subscription functionality. */ - public DaprTopicSubscription(String pubsubName, String topic, String route, Map metadata) { + public DaprTopicSubscription(String pubsubName, String topic, String route, + DaprTopicRoutes routes, Map metadata) { this.pubsubName = pubsubName; this.topic = topic; this.route = route; + this.routes = routes; this.metadata = Collections.unmodifiableMap(metadata); } @@ -51,6 +55,10 @@ public String getRoute() { return route; } + public DaprTopicRoutes getRoutes() { + return routes; + } + public Map getMetadata() { return metadata; } diff --git a/sdk-tests/components/kafka_bindings.yaml b/sdk-tests/components/kafka_bindings.yaml index 0ced720765..574ce4140d 100644 --- a/sdk-tests/components/kafka_bindings.yaml +++ b/sdk-tests/components/kafka_bindings.yaml @@ -19,3 +19,5 @@ spec: value: sample - name: authRequired value: "false" + - name: initialOffset + value: oldest diff --git a/sdk-tests/configuration.yaml b/sdk-tests/configuration.yaml new file mode 100644 index 0000000000..f2f21100e2 --- /dev/null +++ b/sdk-tests/configuration.yaml @@ -0,0 +1,8 @@ +apiVersion: dapr.io/v1alpha1 +kind: Configuration +metadata: + name: pubsubroutingconfig +spec: + features: + - name: PubSub.Routing + enabled: true \ No newline at end of file diff --git a/sdk-tests/deploy/start-dockers.sh b/sdk-tests/deploy/start-dockers.sh new file mode 100755 index 0000000000..7dc1f2d417 --- /dev/null +++ b/sdk-tests/deploy/start-dockers.sh @@ -0,0 +1,3 @@ +#!/bin/sh +docker-compose -f local-test-kafka.yml -p java-sdk-kafka up -d +docker-compose -f local-test-vault.yml -p java-sdk-vault up -d diff --git a/sdk-tests/src/test/java/io/dapr/it/DaprRun.java b/sdk-tests/src/test/java/io/dapr/it/DaprRun.java index cdf1d10d75..817e887bb6 100644 --- a/sdk-tests/src/test/java/io/dapr/it/DaprRun.java +++ b/sdk-tests/src/test/java/io/dapr/it/DaprRun.java @@ -28,7 +28,7 @@ public class DaprRun implements Stoppable { private static final String DAPR_SUCCESS_MESSAGE = "You're up and running!"; - private static final String DAPR_RUN = "dapr run --app-id %s --app-protocol %s --components-path ./components"; + private static final String DAPR_RUN = "dapr run --app-id %s --app-protocol %s --config configuration.yaml --components-path ./components"; // the arg in -Dexec.args is the app's port private static final String DAPR_COMMAND = 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 b7d5f65837..27b117c7d3 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 @@ -210,7 +210,35 @@ public String getContentType() { .setContentType("application/cloudevents+json")).block(); System.out.println("Published one cloud event."); - Thread.sleep(3000); + { + CloudEvent cloudEventV2 = new CloudEvent(); + cloudEventV2.setId("2222"); + cloudEventV2.setData("message from cloudevent v2"); + cloudEventV2.setSource("test"); + cloudEventV2.setSpecversion("1"); + cloudEventV2.setType("myevent.v2"); + cloudEventV2.setDatacontenttype("text/plain"); + client.publishEvent(new PublishEventRequestBuilder(PUBSUB_NAME, TOPIC_NAME, cloudEventV2) + .withContentType("application/cloudevents+json") + .build()).block(); + System.out.println("Published one cloud event for v2."); + } + + { + CloudEvent cloudEventV3 = new CloudEvent(); + cloudEventV3.setId("3333"); + cloudEventV3.setData("message from cloudevent v3"); + cloudEventV3.setSource("test"); + cloudEventV3.setSpecversion("1"); + cloudEventV3.setType("myevent.v3"); + cloudEventV3.setDatacontenttype("text/plain"); + client.publishEvent(new PublishEventRequestBuilder(PUBSUB_NAME, TOPIC_NAME, cloudEventV3) + .withContentType("application/cloudevents+json") + .build()).block(); + System.out.println("Published one cloud event for v3."); + } + + Thread.sleep(2000); callWithRetry(() -> { System.out.println("Checking results for topic " + TOPIC_NAME); @@ -258,6 +286,30 @@ public String getContentType() { .count() == 1); }, 2000); + callWithRetry(() -> { + System.out.println("Checking results for topic " + TOPIC_NAME + " V2"); + // Validate text payload. + final List messages = client.invokeMethod( + daprRun.getAppName(), + "messages/testingtopicV2", + null, + HttpExtension.GET, + CLOUD_EVENT_LIST_TYPE_REF).block(); + assertEquals(1, messages.size()); + }, 2000); + + callWithRetry(() -> { + System.out.println("Checking results for topic " + TOPIC_NAME + " V3"); + // Validate text payload. + final List messages = client.invokeMethod( + daprRun.getAppName(), + "messages/testingtopicV3", + null, + HttpExtension.GET, + CLOUD_EVENT_LIST_TYPE_REF).block(); + assertEquals(1, messages.size()); + }, 2000); + callWithRetry(() -> { System.out.println("Checking results for topic " + TYPED_TOPIC_NAME); // Validate object payload. 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 726e8990bc..dc6c7d2afe 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 @@ -13,6 +13,7 @@ package io.dapr.it.pubsub.http; +import io.dapr.Rule; import io.dapr.Topic; import io.dapr.client.domain.CloudEvent; import org.springframework.web.bind.annotation.GetMapping; @@ -42,6 +43,25 @@ public List> getMessagesByTopic(@PathVariable("topic") String topi return messagesByTopic.getOrDefault(topic, Collections.emptyList()); } + private static final List messagesReceivedTestingTopic = new ArrayList(); + private static final List messagesReceivedTestingTopicV2 = new ArrayList(); + private static final List messagesReceivedTestingTopicV3 = new ArrayList(); + + @GetMapping(path = "/messages/testingtopic") + public List getMessagesReceivedTestingTopic() { + return messagesReceivedTestingTopic; + } + + @GetMapping(path = "/messages/testingtopicV2") + public List getMessagesReceivedTestingTopicV2() { + return messagesReceivedTestingTopicV2; + } + + @GetMapping(path = "/messages/testingtopicV3") + public List getMessagesReceivedTestingTopicV3() { + return messagesReceivedTestingTopicV3; + } + @Topic(name = "testingtopic", pubsubName = "messagebus") @PostMapping("/route1") public Mono handleMessage(@RequestBody(required = false) CloudEvent envelope) { @@ -50,7 +70,39 @@ 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); - messagesByTopic.compute("testingtopic", merge(envelope)); + messagesReceivedTestingTopic.add(envelope); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + @Topic(name = "testingtopic", pubsubName = "messagebus", + rule = @Rule(match = "event.type == 'myevent.v2'", priority = 2)) + @PostMapping(path = "/route1_v2") + public Mono handleMessageV2(@RequestBody(required = false) CloudEvent envelope) { + return Mono.fromRunnable(() -> { + try { + 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); + messagesReceivedTestingTopicV2.add(envelope); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + @Topic(name = "testingtopic", pubsubName = "messagebus", + rule = @Rule(match = "event.type == 'myevent.v3'", priority = 1)) + @PostMapping(path = "/route1_v3") + public Mono handleMessageV3(@RequestBody(required = false) CloudEvent envelope) { + return Mono.fromRunnable(() -> { + try { + 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); + messagesReceivedTestingTopicV3.add(envelope); } catch (Exception e) { throw new RuntimeException(e); } diff --git a/sdk/src/main/java/io/dapr/Rule.java b/sdk/src/main/java/io/dapr/Rule.java new file mode 100644 index 0000000000..68de688467 --- /dev/null +++ b/sdk/src/main/java/io/dapr/Rule.java @@ -0,0 +1,30 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Documented +@Target(ElementType.ANNOTATION_TYPE) +@Retention(RetentionPolicy.RUNTIME) +public @interface Rule { + /** + * The Common Expression Language (CEL) expression to use + * to match the incoming cloud event. + * @return the CEL expression. + */ + String match(); + + /** + * Priority of the rule used for ordering. + * @return the rule priority. + */ + int priority(); +} diff --git a/sdk/src/main/java/io/dapr/Topic.java b/sdk/src/main/java/io/dapr/Topic.java index 665aab9490..2cd7624b02 100644 --- a/sdk/src/main/java/io/dapr/Topic.java +++ b/sdk/src/main/java/io/dapr/Topic.java @@ -36,6 +36,12 @@ */ String pubsubName(); + /** + * The rules used to match the incoming cloud event. + * @return the CEL expression. + */ + Rule rule() default @Rule(match = "", priority = 0); + /** * Metadata in the form of a json object. * { From 56304a45fb061477e4a50dce88907fe23360d7f1 Mon Sep 17 00:00:00 2001 From: Mukundan Sundararajan <65565396+mukundansundar@users.noreply.github.com> Date: Tue, 15 Mar 2022 23:29:12 +0530 Subject: [PATCH 2/3] remove deprecated classes (#704) * remove deprecated classes Signed-off-by: Mukundan Sundararajan * remove builder class ref in ITs Signed-off-by: Mukundan Sundararajan * fix setter in IT Signed-off-by: Mukundan Sundararajan Signed-off-by: Artur Souza --- .../client/domain/QueryStateResponse.java | 50 ------------------- 1 file changed, 50 deletions(-) delete mode 100644 sdk/src/main/java/io/dapr/client/domain/QueryStateResponse.java diff --git a/sdk/src/main/java/io/dapr/client/domain/QueryStateResponse.java b/sdk/src/main/java/io/dapr/client/domain/QueryStateResponse.java deleted file mode 100644 index e084a45661..0000000000 --- a/sdk/src/main/java/io/dapr/client/domain/QueryStateResponse.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * Copyright 2021 The Dapr Authors - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and -limitations under the License. -*/ - -package io.dapr.client.domain; - - -import java.util.Collections; -import java.util.List; -import java.util.Map; - -public class QueryStateResponse { - - private final List> results; - - private final String token; - - private Map metadata; - - public QueryStateResponse(List> results, String token) { - this.results = results == null ? null : Collections.unmodifiableList(results); - this.token = token; - } - - public List> getResults() { - return results; - } - - public String getToken() { - return token; - } - - public Map getMetadata() { - return metadata; - } - - public QueryStateResponse setMetadata(Map metadata) { - this.metadata = metadata == null ? null : Collections.unmodifiableMap(metadata); - return this; - } -} From 18f5b45c332194ded3308b7e2a891f0e76a91381 Mon Sep 17 00:00:00 2001 From: Artur Souza Date: Wed, 22 Jun 2022 12:47:36 -0700 Subject: [PATCH 3/3] Address PR comments. Signed-off-by: Artur Souza --- .../springboot/DaprSubscriptionBuilder.java | 32 ++++++++---- .../java/io/dapr/springboot/DaprTopicKey.java | 18 +++++-- .../io/dapr/springboot/DaprTopicRoutes.java | 18 +++++-- .../io/dapr/springboot/DaprTopicRule.java | 18 +++++-- .../springboot/DaprTopicSubscription.java | 15 +++++- sdk-tests/configuration.yaml | 8 --- sdk-tests/configurations/configuration.yaml | 12 +++++ sdk-tests/deploy/start-dockers.sh | 3 -- .../src/test/java/io/dapr/it/DaprRun.java | 4 +- .../java/io/dapr/it/pubsub/http/PubSubIT.java | 12 ++--- .../java/io/dapr/it/tracing/Validation.java | 5 +- sdk/src/main/java/io/dapr/Rule.java | 19 +++++-- .../client/domain/QueryStateResponse.java | 50 +++++++++++++++++++ 13 files changed, 163 insertions(+), 51 deletions(-) delete mode 100644 sdk-tests/configuration.yaml create mode 100644 sdk-tests/configurations/configuration.yaml delete mode 100755 sdk-tests/deploy/start-dockers.sh create mode 100644 sdk/src/main/java/io/dapr/client/domain/QueryStateResponse.java diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprSubscriptionBuilder.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprSubscriptionBuilder.java index 1974e02656..9ed65508e6 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprSubscriptionBuilder.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprSubscriptionBuilder.java @@ -1,7 +1,15 @@ /* - * Copyright (c) Microsoft Corporation and Dapr Contributors. - * Licensed under the MIT License. - */ + * Copyright 2022 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ package io.dapr.springboot; @@ -12,7 +20,7 @@ import java.util.Map; import java.util.stream.Collectors; -public class DaprSubscriptionBuilder { +class DaprSubscriptionBuilder { private final String pubsubName; private final String topic; private final List rules; @@ -24,7 +32,7 @@ public class DaprSubscriptionBuilder { * @param pubsubName The pubsub name to subscribe to. * @param topic The topic to subscribe to. */ - public DaprSubscriptionBuilder(String pubsubName, String topic) { + DaprSubscriptionBuilder(String pubsubName, String topic) { this.pubsubName = pubsubName; this.topic = topic; this.rules = new ArrayList<>(); @@ -35,8 +43,9 @@ public DaprSubscriptionBuilder(String pubsubName, String topic) { /** * Sets the default path for the subscription. * @param path The default path. + * @return this instance. */ - public void setDefaultPath(String path) { + DaprSubscriptionBuilder setDefaultPath(String path) { if (defaultPath != null) { throw new RuntimeException( String.format( @@ -44,6 +53,7 @@ public void setDefaultPath(String path) { this.topic, this.pubsubName)); } defaultPath = path; + return this; } /** @@ -51,8 +61,9 @@ public void setDefaultPath(String path) { * @param path The path to route to. * @param match The CEL expression the event must match. * @param priority The priority of the rule. + * @return this instance. */ - public void addRule(String path, String match, int priority) { + public DaprSubscriptionBuilder addRule(String path, String match, int priority) { if (rules.stream().anyMatch(e -> e.getPriority() == priority)) { throw new RuntimeException( String.format( @@ -60,14 +71,17 @@ public void addRule(String path, String match, int priority) { priority, this.topic, this.pubsubName)); } rules.add(new TopicRule(path, match, priority)); + return this; } /** * Sets the metadata for the subscription. * @param metadata The metadata. + * @return this instance. */ - public void setMetadata(Map metadata) { + public DaprSubscriptionBuilder setMetadata(Map metadata) { this.metadata = metadata; + return this; } /** @@ -90,7 +104,7 @@ public DaprTopicSubscription build() { return new DaprTopicSubscription(this.pubsubName, this.topic, route, routes, metadata); } - static class TopicRule { + private static class TopicRule { private final String path; private final String match; private final int priority; diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicKey.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicKey.java index b1e9c4ef3b..e2a3ee6b65 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicKey.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicKey.java @@ -1,17 +1,25 @@ /* - * Copyright (c) Microsoft Corporation and Dapr Contributors. - * Licensed under the MIT License. - */ + * Copyright 2022 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ package io.dapr.springboot; import java.util.Objects; -public class DaprTopicKey { +class DaprTopicKey { private final String pubsubName; private final String topic; - public DaprTopicKey(String pubsubName, String topic) { + DaprTopicKey(String pubsubName, String topic) { this.pubsubName = pubsubName; this.topic = topic; } diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRoutes.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRoutes.java index e78d06dab3..fcb1bbb0f7 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRoutes.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRoutes.java @@ -1,7 +1,15 @@ /* - * Copyright (c) Microsoft Corporation and Dapr Contributors. - * Licensed under the MIT License. - */ + * Copyright 2022 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ package io.dapr.springboot; @@ -10,12 +18,12 @@ import java.util.List; import java.util.Optional; -public class DaprTopicRoutes { +class DaprTopicRoutes { private final List rules; @JsonProperty("default") private final String defaultRoute; - public DaprTopicRoutes(List rules, String defaultRoute) { + DaprTopicRoutes(List rules, String defaultRoute) { this.rules = rules; this.defaultRoute = defaultRoute; } diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRule.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRule.java index 5fc8b905e4..4342e242a5 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRule.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicRule.java @@ -1,17 +1,25 @@ /* - * Copyright (c) Microsoft Corporation and Dapr Contributors. - * Licensed under the MIT License. - */ + * Copyright 2022 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ package io.dapr.springboot; import java.util.Objects; -public class DaprTopicRule { +class DaprTopicRule { private final String match; private final String path; - public DaprTopicRule(String match, String path) { + DaprTopicRule(String match, String path) { this.match = match; this.path = path; } diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicSubscription.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicSubscription.java index 956c0bbcb8..1612bc1414 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicSubscription.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprTopicSubscription.java @@ -19,13 +19,24 @@ /** * Class to represent a subscription topic along with its metadata. */ -public class DaprTopicSubscription { +class DaprTopicSubscription { private final String pubsubName; private final String topic; private final String route; private final DaprTopicRoutes routes; private final Map metadata; + /** + * Create a subscription topic. + * @param pubsubName The pubsub name to subscribe to. + * @param topic The topic to subscribe to. + * @param route Destination route for messages. + * @param metadata Metdata for extended subscription functionality. + */ + DaprTopicSubscription(String pubsubName, String topic, String route, Map metadata) { + this(pubsubName, topic, route, null, metadata); + } + /** * Create a subscription topic. * @param pubsubName The pubsub name to subscribe to. @@ -34,7 +45,7 @@ public class DaprTopicSubscription { * @param routes Destination routes with rules for messages. * @param metadata Metdata for extended subscription functionality. */ - public DaprTopicSubscription(String pubsubName, String topic, String route, + DaprTopicSubscription(String pubsubName, String topic, String route, DaprTopicRoutes routes, Map metadata) { this.pubsubName = pubsubName; this.topic = topic; diff --git a/sdk-tests/configuration.yaml b/sdk-tests/configuration.yaml deleted file mode 100644 index f2f21100e2..0000000000 --- a/sdk-tests/configuration.yaml +++ /dev/null @@ -1,8 +0,0 @@ -apiVersion: dapr.io/v1alpha1 -kind: Configuration -metadata: - name: pubsubroutingconfig -spec: - features: - - name: PubSub.Routing - enabled: true \ No newline at end of file diff --git a/sdk-tests/configurations/configuration.yaml b/sdk-tests/configurations/configuration.yaml new file mode 100644 index 0000000000..e2d3161546 --- /dev/null +++ b/sdk-tests/configurations/configuration.yaml @@ -0,0 +1,12 @@ +apiVersion: dapr.io/v1alpha1 +kind: Configuration +metadata: + name: testconfiguration +spec: + tracing: + samplingRate: "1" + zipkin: + endpointAddress: http://localhost:9411/api/v2/spans + features: + - name: PubSub.Routing + enabled: true \ No newline at end of file diff --git a/sdk-tests/deploy/start-dockers.sh b/sdk-tests/deploy/start-dockers.sh deleted file mode 100755 index 7dc1f2d417..0000000000 --- a/sdk-tests/deploy/start-dockers.sh +++ /dev/null @@ -1,3 +0,0 @@ -#!/bin/sh -docker-compose -f local-test-kafka.yml -p java-sdk-kafka up -d -docker-compose -f local-test-vault.yml -p java-sdk-vault up -d diff --git a/sdk-tests/src/test/java/io/dapr/it/DaprRun.java b/sdk-tests/src/test/java/io/dapr/it/DaprRun.java index 817e887bb6..13b0f901c2 100644 --- a/sdk-tests/src/test/java/io/dapr/it/DaprRun.java +++ b/sdk-tests/src/test/java/io/dapr/it/DaprRun.java @@ -28,7 +28,9 @@ public class DaprRun implements Stoppable { private static final String DAPR_SUCCESS_MESSAGE = "You're up and running!"; - private static final String DAPR_RUN = "dapr run --app-id %s --app-protocol %s --config configuration.yaml --components-path ./components"; + private static final String DAPR_RUN = "dapr run --app-id %s --app-protocol %s " + + "--config ./configurations/configuration.yaml " + + "--components-path ./components"; // the arg in -Dexec.args is the app's port private static final String DAPR_COMMAND = 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 27b117c7d3..84f81239e3 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 @@ -218,9 +218,9 @@ public String getContentType() { cloudEventV2.setSpecversion("1"); cloudEventV2.setType("myevent.v2"); cloudEventV2.setDatacontenttype("text/plain"); - client.publishEvent(new PublishEventRequestBuilder(PUBSUB_NAME, TOPIC_NAME, cloudEventV2) - .withContentType("application/cloudevents+json") - .build()).block(); + client.publishEvent( + new PublishEventRequest(PUBSUB_NAME, TOPIC_NAME, cloudEventV2) + .setContentType("application/cloudevents+json")).block(); System.out.println("Published one cloud event for v2."); } @@ -232,9 +232,9 @@ public String getContentType() { cloudEventV3.setSpecversion("1"); cloudEventV3.setType("myevent.v3"); cloudEventV3.setDatacontenttype("text/plain"); - client.publishEvent(new PublishEventRequestBuilder(PUBSUB_NAME, TOPIC_NAME, cloudEventV3) - .withContentType("application/cloudevents+json") - .build()).block(); + client.publishEvent( + new PublishEventRequest(PUBSUB_NAME, TOPIC_NAME, cloudEventV3) + .setContentType("application/cloudevents+json")).block(); System.out.println("Published one cloud event for v3."); } diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/Validation.java b/sdk-tests/src/test/java/io/dapr/it/tracing/Validation.java index e7a175583f..0f6a7867e8 100644 --- a/sdk-tests/src/test/java/io/dapr/it/tracing/Validation.java +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/Validation.java @@ -46,12 +46,13 @@ public final class Validation { public static void validate(String spanName, String sleepSpanName) throws Exception { // Must wait for some time to make sure Zipkin receives all spans. - Thread.sleep(5000); + Thread.sleep(10000); HttpUrl.Builder urlBuilder = new HttpUrl.Builder(); urlBuilder.scheme("http") .host("localhost") .port(9411) - .addPathSegments("api/v2/traces"); + .addPathSegments("api/v2/traces") + .addQueryParameter("limit", "100"); Request.Builder requestBuilder = new Request.Builder() .url(urlBuilder.build()); requestBuilder.method("GET", null); diff --git a/sdk/src/main/java/io/dapr/Rule.java b/sdk/src/main/java/io/dapr/Rule.java index 68de688467..894e5d4b0b 100644 --- a/sdk/src/main/java/io/dapr/Rule.java +++ b/sdk/src/main/java/io/dapr/Rule.java @@ -1,7 +1,15 @@ /* - * Copyright (c) Microsoft Corporation and Dapr Contributors. - * Licensed under the MIT License. - */ + * Copyright 2022 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ package io.dapr; @@ -18,13 +26,14 @@ /** * The Common Expression Language (CEL) expression to use * to match the incoming cloud event. + * Examples. * @return the CEL expression. */ String match(); /** - * Priority of the rule used for ordering. - * @return the rule priority. + * Priority of the rule used for ordering. Lowest numbers have higher priority. + * @return the rule's priority. */ int priority(); } diff --git a/sdk/src/main/java/io/dapr/client/domain/QueryStateResponse.java b/sdk/src/main/java/io/dapr/client/domain/QueryStateResponse.java new file mode 100644 index 0000000000..e084a45661 --- /dev/null +++ b/sdk/src/main/java/io/dapr/client/domain/QueryStateResponse.java @@ -0,0 +1,50 @@ +/* + * Copyright 2021 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.client.domain; + + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +public class QueryStateResponse { + + private final List> results; + + private final String token; + + private Map metadata; + + public QueryStateResponse(List> results, String token) { + this.results = results == null ? null : Collections.unmodifiableList(results); + this.token = token; + } + + public List> getResults() { + return results; + } + + public String getToken() { + return token; + } + + public Map getMetadata() { + return metadata; + } + + public QueryStateResponse setMetadata(Map metadata) { + this.metadata = metadata == null ? null : Collections.unmodifiableMap(metadata); + return this; + } +}