Declarative and programmatic subscription methods

Learn more about the methods by which Dapr allows you to subscribe to topics.

Pub/sub API subscription methods

Dapr applications can subscribe to published topics via two methods that support the same features: declarative and programmatic.

Subscription methodDescription
DeclarativeSubscription is defined in an external file. The declarative approach removes the Dapr dependency from your code and allows for existing applications to subscribe to topics, without having to change code.
ProgrammaticSubscription is defined in the application code. The programmatic approach implements the subscription in your code.

The examples below demonstrate pub/sub messaging between a checkout app and an orderprocessing app via the orders topic. The examples demonstrate the same Dapr pub/sub component used first declaratively, then programmatically.

Declarative subscriptions

You can subscribe declaratively to a topic using an external component file. This example uses a YAML component file named subscription.yaml:

  1. apiVersion: dapr.io/v1alpha1
  2. kind: Subscription
  3. metadata:
  4. name: order
  5. spec:
  6. topic: orders
  7. route: /checkout
  8. pubsubname: pubsub
  9. scopes:
  10. - orderprocessing
  11. - checkout

Here the subscription called order:

  • Uses the pub/sub component called pubsub to subscribes to the topic called orders.
  • Sets the route field to send all topic messages to the /checkout endpoint in the app.
  • Sets scopes field to scope this subscription for access only by apps with IDs orderprocessing and checkout.

When running Dapr, set the YAML component file path to point Dapr to the component.

  1. dapr run --app-id myapp --components-path ./myComponents -- dotnet run
  1. dapr run --app-id myapp --components-path ./myComponents -- mvn spring-boot:run
  1. dapr run --app-id myapp --components-path ./myComponents -- python3 app.py
  1. dapr run --app-id myapp --components-path ./myComponents -- npm start
  1. dapr run --app-id myapp --components-path ./myComponents -- go run app.go

In Kubernetes, apply the component to the cluster:

  1. kubectl apply -f subscription.yaml

In your application code, subscribe to the topic specified in the Dapr pub/sub component.

  1. //Subscribe to a topic
  2. [Topic("pubsub", "orders")]
  3. [HttpPost("checkout")]
  4. public void getCheckout([FromBody] int orderId)
  5. {
  6. Console.WriteLine("Subscriber received : " + orderId);
  7. }
  1. //Subscribe to a topic
  2. @Topic(name = "orders", pubsubName = "pubsub")
  3. @PostMapping(path = "/checkout")
  4. public Mono<Void> getCheckout(@RequestBody(required = false) CloudEvent<String> cloudEvent) {
  5. return Mono.fromRunnable(() -> {
  6. try {
  7. log.info("Subscriber received: " + cloudEvent.getData());
  8. } catch (Exception e) {
  9. throw new RuntimeException(e);
  10. }
  11. });
  12. }
  1. #Subscribe to a topic
  2. @app.subscribe(pubsub_name='pubsub', topic='orders')
  3. def mytopic(event: v1.Event) -> None:
  4. data = json.loads(event.Data())
  5. logging.info('Subscriber received: ' + str(data))
  6. app.run(6002)
  1. //Subscribe to a topic
  2. await server.pubsub.subscribe("pubsub", "orders", async (orderId) => {
  3. console.log(`Subscriber received: ${JSON.stringify(orderId)}`)
  4. });
  5. await server.startServer();
  1. //Subscribe to a topic
  2. if err := s.AddTopicEventHandler(sub, eventHandler); err != nil {
  3. log.Fatalf("error adding topic subscription: %v", err)
  4. }
  5. if err := s.Start(); err != nil && err != http.ErrServerClosed {
  6. log.Fatalf("error listenning: %v", err)
  7. }
  8. func eventHandler(ctx context.Context, e *common.TopicEvent) (retry bool, err error) {
  9. log.Printf("Subscriber received: %s", e.Data)
  10. return false, nil
  11. }

The /checkout endpoint matches the route defined in the subscriptions and this is where Dapr sends all topic messages to.

Programmatic subscriptions

The programmatic approach returns the routes JSON structure within the code, unlike the declarative approach’s route YAML structure. In the example below, you define the values found in the declarative YAML subscription above within the application code.

  1. [Topic("pubsub", "checkout", event.type ==\"order\"")]
  2. [HttpPost("orders")]
  3. public async Task<ActionResult<Stock>> HandleCheckout(Checkout checkout, [FromServices] DaprClient daprClient)
  4. {
  5. // Logic
  6. return stock;
  7. }
  1. private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
  2. @Topic(name = "checkout", pubsubName = "pubsub")
  3. @PostMapping(path = "/orders")
  4. public Mono<Void> handleMessage(@RequestBody(required = false) CloudEvent<String> cloudEvent) {
  5. return Mono.fromRunnable(() -> {
  6. try {
  7. System.out.println("Subscriber received: " + cloudEvent.getData());
  8. System.out.println("Subscriber received: " + OBJECT_MAPPER.writeValueAsString(cloudEvent));
  9. } catch (Exception e) {
  10. throw new RuntimeException(e);
  11. }
  12. });
  1. @app.route('/dapr/subscribe', methods=['GET'])
  2. def subscribe():
  3. subscriptions = [
  4. {
  5. 'pubsubname': 'pubsub',
  6. 'topic': 'checkout',
  7. 'routes': {
  8. 'rules': [
  9. {
  10. 'match': 'event.type == "order"',
  11. 'path': '/orders'
  12. },
  13. ],
  14. 'default': '/orders'
  15. }
  16. }]
  17. return jsonify(subscriptions)
  18. @app.route('/orders', methods=['POST'])
  19. def ds_subscriber():
  20. print(request.json, flush=True)
  21. return json.dumps({'success':True}), 200, {'ContentType':'application/json'}
  22. app.run()
  1. const express = require('express')
  2. const bodyParser = require('body-parser')
  3. const app = express()
  4. app.use(bodyParser.json({ type: 'application/*+json' }));
  5. const port = 3000
  6. app.get('/dapr/subscribe', (req, res) => {
  7. res.json([
  8. {
  9. pubsubname: "pubsub",
  10. topic: "checkout",
  11. routes: {
  12. rules: [
  13. {
  14. match: 'event.type == "order"',
  15. path: '/orders'
  16. },
  17. ],
  18. default: '/products'
  19. }
  20. }
  21. ]);
  22. })
  23. app.post('/orders', (req, res) => {
  24. console.log(req.body);
  25. res.sendStatus(200);
  26. });
  27. app.listen(port, () => console.log(`consumer app listening on port ${port}!`))
  1. package main
  2. "encoding/json"
  3. "fmt"
  4. "log"
  5. "net/http"
  6. "github.com/gorilla/mux"
  7. )
  8. const appPort = 3000
  9. type subscription struct {
  10. PubsubName string `json:"pubsubname"`
  11. Topic string `json:"topic"`
  12. Metadata map[string]string `json:"metadata,omitempty"`
  13. Routes routes `json:"routes"`
  14. }
  15. type routes struct {
  16. Rules []rule `json:"rules,omitempty"`
  17. Default string `json:"default,omitempty"`
  18. }
  19. type rule struct {
  20. Match string `json:"match"`
  21. Path string `json:"path"`
  22. }
  23. // This handles /dapr/subscribe
  24. func configureSubscribeHandler(w http.ResponseWriter, _ *http.Request) {
  25. t := []subscription{
  26. {
  27. PubsubName: "pubsub",
  28. Topic: "checkout",
  29. Routes: routes{
  30. Rules: []rule{
  31. {
  32. Match: `event.type == "order"`,
  33. Path: "/orders",
  34. },
  35. },
  36. Default: "/orders",
  37. },
  38. },
  39. }
  40. w.WriteHeader(http.StatusOK)
  41. json.NewEncoder(w).Encode(t)
  42. }
  43. func main() {
  44. router := mux.NewRouter().StrictSlash(true)
  45. router.HandleFunc("/dapr/subscribe", configureSubscribeHandler).Methods("GET")
  46. log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", appPort), router))
  47. }

Next Steps

Last modified June 23, 2022: Merge pull request #2550 from ItalyPaleAle/cosmosdb-harcoded-dapr-version (cf03237)