### Run Dapr with Declarative Subscriptions (Go)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Use the `dapr run` command to start your Go application with Dapr, specifying the resource path for components.

```bash
dapr run --app-id myapp --resources-path ./myComponents -- go run app.go

```

--------------------------------

### Run Dapr with Declarative Subscriptions (.NET)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Use the `dapr run` command to start your .NET application with Dapr, specifying the resource path for components.

```bash
dapr run --app-id myapp --resources-path ./myComponents -- dotnet run

```

--------------------------------

### Run Dapr with Declarative Subscriptions (Python)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Use the `dapr run` command to start your Python application with Dapr, specifying the resource path for components.

```bash
dapr run --app-id myapp --resources-path ./myComponents -- python3 app.py

```

--------------------------------

### Run Dapr with Declarative Subscriptions (JavaScript)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Use the `dapr run` command to start your Node.js application with Dapr, specifying the resource path for components.

```bash
dapr run --app-id myapp --resources-path ./myComponents -- npm start

```

--------------------------------

### Run Dapr with Declarative Subscriptions (Java)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Use the `dapr run` command to start your Java application with Dapr, specifying the resource path for components.

```bash
dapr run --app-id myapp --resources-path ./myComponents -- mvn spring-boot:run

```

--------------------------------

### Go Programmatic Subscription

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-route-messages

This Go example demonstrates programmatic subscription configuration using HTTP handlers. It defines routes with specific match conditions and a default path for the 'inventory' topic.

```go
package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"

	"github.com/gorilla/mux"
)

const appPort = 3000

type subscription struct {
	PubsubName string            `json:"pubsubname"`
	Topic      string            `json:"topic"`
	Metadata   map[string]string `json:"metadata,omitempty"`
	Routes     routes            `json:"routes"`
}

type routes struct {
	Rules   []rule `json:"rules,omitempty"`
	Default string `json:"default,omitempty"`
}

type rule struct {
	Match string `json:"match"`
	Path  string `json:"path"`
}

// This handles /dapr/subscribe
func configureSubscribeHandler(w http.ResponseWriter, _ *http.Request) {
	t := []subscription{
		{
			PubsubName: "pubsub",
			Topic:      "inventory",
			Routes: routes{
				Rules: []rule{
					{
						Match: `event.type == "widget"`,
						Path:  "/widgets",
					},
					{
						Match: `event.type == "gadget"`,
						Path:  "/gadgets",
					},
				},
				Default: "/products",
			},
		},
	}

	w.WriteHeader(http.StatusOK)
	json.NewEncoder(w).Encode(t)
}

func main() {
	router := mux.NewRouter().StrictSlash(true)
	router.HandleFunc("/dapr/subscribe", configureSubscribeHandler).Methods("GET")
	log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", appPort), router))
}
```

--------------------------------

### Start Dapr Instance

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Starts a Dapr sidecar instance with a specified app-id. This is often a prerequisite for other Dapr operations.

```bash
dapr run --app-id orderprocessing --dapr-http-port 3601

```

--------------------------------

### Run .NET Publisher Application

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Launches a Dapr sidecar and a .NET publisher application. This command is used to run the .NET code example.

```bash
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 --app-protocol https dotnet run

```

--------------------------------

### .NET Programmatic Subscription with Attributes

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-route-messages

This .NET example uses attributes to define Dapr Pub/Sub subscriptions. It configures specific routes for 'widget' and 'gadget' events, and a default route for other products.

```csharp
[Topic("pubsub", "inventory", "event.type ==\"widget\"", 1)]
        [HttpPost("widgets")]
        public async Task<ActionResult<Stock>> HandleWidget(Widget widget, [FromServices] DaprClient daprClient)
        {
            // Logic
            return stock;
        }

        [Topic("pubsub", "inventory", "event.type ==\"gadget\"", 2)]
        [HttpPost("gadgets")]
        public async Task<ActionResult<Stock>> HandleGadget(Gadget gadget, [FromServices] DaprClient daprClient)
        {
            // Logic
            return stock;
        }

        [Topic("pubsub", "inventory")]
        [HttpPost("products")]
        public async Task<ActionResult<Stock>> HandleProduct(Product product, [FromServices] DaprClient daprClient)
        {
            // Logic
            return stock;
        }
```

--------------------------------

### Python Flask Programmatic Subscription

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-route-messages

Use this Python Flask example to define subscriptions programmatically. It configures routes for specific event types and a default route for the 'inventory' topic.

```python
import flask
from flask import request, jsonify
from flask_cors import CORS
import json
import sys

app = flask.Flask(__name__)
CORS(app)

@app.route('/dapr/subscribe', methods=['GET'])
def subscribe():
    subscriptions = [
      {
        'pubsubname': 'pubsub',
        'topic': 'inventory',
        'routes': {
          'rules': [
            {
              'match': 'event.type == "widget"',
              'path': '/widgets'
            },
            {
              'match': 'event.type == "gadget"',
              'path': '/gadgets'
            },
          ],
          'default': '/products'
        }
      }]
    return jsonify(subscriptions)

@app.route('/products', methods=['POST'])
def ds_subscriber():
    print(request.json, flush=True)
    return json.dumps({'success':True}), 200, {'ContentType':'application/json'}
app.run()
```

--------------------------------

### Node.js: Bulk Subscribe with Default and Custom Configurations

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-bulk

Demonstrates how to subscribe to a topic using Dapr's bulk subscription feature in Node.js. Includes examples for both default configuration and custom configurations specifying max messages and await duration.

```typescript
import { DaprServer } from "@dapr/dapr";

const pubSubName = "orderPubSub";
const topic = "topicbulk";

const daprHost = process.env.DAPR_HOST || "127.0.0.1";
const daprPort = process.env.DAPR_HTTP_PORT || "3502";
const serverHost = process.env.SERVER_HOST || "127.0.0.1";
const serverPort = process.env.APP_PORT || 5001;

async function start() {
    const server = new DaprServer({
        serverHost,
        serverPort,
        clientOptions: {
            daprHost,
            daprPort,
        },
    });

    // Publish multiple messages to a topic with default config.
    await client.pubsub.bulkSubscribeWithDefaultConfig(pubSubName, topic, (data) => console.log("Subscriber received: " + JSON.stringify(data)));

    // Publish multiple messages to a topic with specific maxMessagesCount and maxAwaitDurationMs.
    await client.pubsub.bulkSubscribeWithConfig(pubSubName, topic, (data) => console.log("Subscriber received: " + JSON.stringify(data)), 100, 40);
}

```

--------------------------------

### Run Java Publisher Application

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Launches a Dapr sidecar and a Java publisher application using Maven. This command is used to run the Java code example.

```bash
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 mvn spring-boot:run

```

--------------------------------

### Limit Allowed Topics for Pub/Sub with Redis

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-scopes

This example shows how to configure `allowedTopics` for a Dapr Redis Pub/Sub component. Use this to govern topic creation and prevent an unlimited growth of topics by restricting the set of topics that can be used.

```yaml
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: pubsub
spec:
  type: pubsub.redis
  version: v1
  metadata:
  - name: redisHost
    value: "localhost:6379"
  - name: redisPassword
    value: ""
  - name: allowedTopics
    value: "topic1,topic2,topic3"
```

--------------------------------

### C#: Bulk Subscribe Endpoint Implementation

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-bulk

Shows how to implement a bulk subscription endpoint in a C# ASP.NET Core application using Dapr. This example defines a controller that handles bulk messages and returns appropriate responses for each entry.

```csharp
using Microsoft.AspNetCore.Mvc;
using Dapr.AspNetCore;
using Dapr;

namespace DemoApp.Controllers;

[ApiController]
[Route("[controller]")]
public class BulkMessageController : ControllerBase
{
    private readonly ILogger<BulkMessageController> logger;

    public BulkMessageController(ILogger<BulkMessageController> logger)
    {
        this.logger = logger;
    }

    [BulkSubscribe("messages", 10, 10)]
    [Topic("pubsub", "messages")]
    public ActionResult<BulkSubscribeAppResponse> HandleBulkMessages([FromBody] BulkSubscribeMessage<BulkMessageModel<BulkMessageModel>> bulkMessages)
    {
        List<BulkSubscribeAppResponseEntry> responseEntries = new List<BulkSubscribeAppResponseEntry>();
        logger.LogInformation($"Received {bulkMessages.Entries.Count()} messages");
        foreach (var message in bulkMessages.Entries)
        {
            try
            {
                logger.LogInformation($"Received a message with data '{message.Event.Data.MessageData}'");
                responseEntries.Add(new BulkSubscribeAppResponseEntry(message.EntryId, BulkSubscribeAppResponseStatus.SUCCESS));
            }
            catch (Exception e)
            {
                logger.LogError(e.Message);
                responseEntries.Add(new BulkSubscribeAppResponseEntry(message.EntryId, BulkSubscribeAppResponseStatus.RETRY));
            }
        }
        return new BulkSubscribeAppResponse(responseEntries);
    }
    public class BulkMessageModel
    {
        public string MessageData { get; set; }
    }
}

```

--------------------------------

### Publish Event with Dapr SDK (Python)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Publishes an event using the Dapr SDK for Python. This example demonstrates continuous publishing with basic logging.

```python
#dependencies  
import random
from time import sleep    
import requests
import logging
import json
from dapr.clients import DaprClient

#code
logging.basicConfig(level = logging.INFO)
while True:
    sleep(random.randrange(50, 5000) / 1000)
    orderId = random.randint(1, 1000)
    PUBSUB_NAME = 'order-pub-sub'
    TOPIC_NAME = 'orders'
    with DaprClient() as client:
        #Using Dapr SDK to publish a topic
        result = client.publish_event(
            pubsub_name=PUBSUB_NAME,
            topic_name=TOPIC_NAME,
            data=json.dumps(orderId),
            data_content_type='application/json',
        )
    logging.info('Published data: ' + str(orderId))


```

--------------------------------

### Configure Pub/Sub Scopes with Redis

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-scopes

This example demonstrates how to configure `publishingScopes` and `subscriptionScopes` for a Dapr Redis Pub/Sub component. Use this to restrict which applications can publish to or subscribe from specific topics.

```yaml
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: pubsub
spec:
  type: pubsub.redis
  version: v1
  metadata:
  - name: redisHost
    value: "localhost:6379"
  - name: redisPassword
    value: ""
  - name: publishingScopes
    value: "app1=topic1;app2=topic2,topic3;app3="
  - name: subscriptionScopes
    value: "app2=;app3=topic1"
```

--------------------------------

### PHP Programmatic Subscription with Dapr SDK

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-route-messages

This PHP example utilizes the Dapr SDK to programmatically define Pub/Sub subscriptions. It configures routing rules for specific event types and a default path.

```php
<?php

require_once __DIR__.'/vendor/autoload.php';

$app = \Dapr\App::create(configure: fn(\DI\ContainerBuilder $builder) => $builder->addDefinitions(['dapr.subscriptions' => [
    new \Dapr\PubSub\Subscription(pubsubname: 'pubsub', topic: 'inventory', routes: (
      rules: => [
        ('match': 'event.type == "widget"', path: '/widgets'),
        ('match': 'event.type == "gadget"', path: '/gadgets'),
      ]
      default: '/products')),]]));
$app->post('/products', function(
    #[
Dapr\Attributes\FromBody]
    \Dapr\PubSub\CloudEvent $cloudEvent,
    \Psr\Log\LoggerInterface $logger
    ) {
        $logger->alert('Received event: {event}', ['event' => $cloudEvent]);
        return ['status' => 'SUCCESS'];
    }
);
$app->start();

```

--------------------------------

### Publish Event with Dapr SDK (Java)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Publishes an event using the Dapr SDK for Java. This example includes dependencies and a loop for continuous publishing.

```java
//dependencies
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.domain.Metadata;
import static java.util.Collections.singletonMap;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Random;
import java.util.concurrent.TimeUnit;

//code
@SpringBootApplication
public class OrderProcessingServiceApplication {

	private static final Logger log = LoggerFactory.getLogger(OrderProcessingServiceApplication.class);

	public static void main(String[] args) throws InterruptedException{
		String MESSAGE_TTL_IN_SECONDS = "1000";
		String TOPIC_NAME = "orders";
		String PUBSUB_NAME = "order-pub-sub";

		while(true) {
			TimeUnit.MILLISECONDS.sleep(5000);
			Random random = new Random();
			int orderId = random.nextInt(1000-1) + 1;
			DaprClient client = new DaprClientBuilder().build();
      //Using Dapr SDK to publish a topic
			client.publishEvent(
					PUBSUB_NAME,
					TOPIC_NAME,
					orderId,
					singletonMap(Metadata.TTL_IN_SECONDS, MESSAGE_TTL_IN_SECONDS)).block();
			log.info("Published data:" + orderId);
		}
	}
}

```

--------------------------------

### Example CloudEvent Payload with Custom Values

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-cloudevents

Demonstrates the resulting JSON payload of a CloudEvent after custom ID and source values have been applied via metadata.

```json
{
  "topic": "orders",
  "pubsubname": "order_pub_sub",
  "traceid": "00-113ad9c4e42b27583ae98ba698d54255-e3743e35ff56f219-01",
  "tracestate": "",
  "data": {
    "orderId": 1
  },
  "id": "d99b228f-6c73-4e78-8c4d-3f80a043d317",
  "specversion": "1.0",
  "datacontenttype": "application/json; charset=utf-8",
  "source": "payment",
  "type": "com.dapr.event.sent",
  "time": "2020-09-23T06:23:21Z",
  "traceparent": "00-113ad9c4e42b27583ae98ba698d54255-e3743e35ff56f219-01"
}

```

--------------------------------

### Declarative Subscription Component

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Define a declarative subscription using a YAML component file. This example subscribes to the 'orders' topic using the 'pubsub' component and scopes it to 'orderprocessing'.

```yaml
apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: order
spec:
  topic: orders
  routes:
    default: /orders
  pubsubname: pubsub
scopes:
- orderprocessing

```

--------------------------------

### Publish Event with Dapr SDK (.NET)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Publishes an event using the Dapr SDK for .NET. This example demonstrates continuous publishing within a .NET application.

```csharp
using System;
using System.Collections.Generic;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Threading.Tasks;
using Dapr.Client;
using System.Threading;

const string PUBSUB_NAME = "order-pub-sub";
const string TOPIC_NAME = "orders";

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddDaprClient();

var app = builder.Build();
var random = new Random();

var client = app.Services.GetRequiredService<DaprClient>();

while(true) {
    await Task.Delay(TimeSpan.FromSeconds(5));
    var orderId = random.Next(1,1000);
    var source = new CancellationTokenSource();
    var cancellationToken = source.Token;
    
    //Using Dapr SDK to publish a topic
    await client.PublishEventAsync(PUBSUB_NAME, TOPIC_NAME, orderId, cancellationToken);
    Console.WriteLine("Published data: " + orderId);
}

```

--------------------------------

### Declarative Subscription with Routing Rules

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-route-messages

Define routing rules for a Dapr pub/sub subscription using `dapr.io/v2alpha1`. This example shows how to route messages based on `event.type` to different paths (`/widgets`, `/gadgets`) and includes a default route (`/products`).

```yaml
apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: myevent-subscription
spec:
  pubsubname: pubsub
  topic: inventory
  routes:
    rules:
      - match: event.type == "widget"
        path: /widgets
      - match: event.type == "gadget"
        path: /gadgets
    default: /products
scopes:
  - app1
  - app2

```

--------------------------------

### JavaScript Express Programmatic Subscription

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-route-messages

This JavaScript Express example shows how to programmatically define Dapr Pub/Sub subscriptions. It sets up rules for different event types and a default route.

```javascript
const express = require('express')
const bodyParser = require('body-parser')
const app = express()
app.use(bodyParser.json({ type: 'application/*+json' }));

const port = 3000

app.get('/dapr/subscribe', (req, res) => {
  res.json([
    {
      pubsubname: "pubsub",
      topic: "inventory",
      routes: {
        rules: [
          {
            match: 'event.type == "widget"',
            path: '/widgets'
          },
          {
            match: 'event.type == "gadget"',
            path: '/gadgets'
          },
        ],
        default: '/products'
      }
    }
  ]);
})

app.post('/products', (req, res) => {
  console.log(req.body);
  res.sendStatus(200);
});

app.listen(port, () => console.log(`consumer app listening on port ${port}!`))
```

--------------------------------

### Publish Message with TTL using Dapr PHP SDK

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-message-ttl

Publish a message with a time-to-live using the Dapr PHP SDK. This example shows how to configure the TTL metadata for a published message.

```php
<?php

require_once __DIR__.'/vendor/autoload.php';

$app = \Dapr\App::create();
$app->run(function(\DI\FactoryInterface $factory) {
    $publisher = $factory->make(\Dapr\PubSub\Publish::class, ['pubsub' => 'pubsub']);
    $publisher->topic('TOPIC_A')->publish('data', ['ttlInSeconds' => '120']);
});

```

--------------------------------

### Handle Orders Topic in Go

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Define a subscription configuration and an event handler function in Go to process messages from the 'orders' topic.

```go
//Subscribe to a topic
var sub = &common.Subscription{
	PubsubName: "pubsub",
	Topic:      "orders",
	Route:      "/orders",
}

func eventHandler(ctx context.Context, e *common.TopicEvent) (retry bool, err error) {
	log.Printf("Subscriber received: %s", e.Data)
	return false, nil
}

```

--------------------------------

### Run Go publisher application

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Launches a Dapr sidecar and the Go publisher application. Navigate to the directory containing the code before running.

```bash
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 go run OrderProcessingService.go
```

--------------------------------

### Create Dynamic Streaming Subscription (.NET)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Dynamically creates a streaming subscription with specified message handling policies. Use this when you need to programmatically manage subscription lifecycles and message processing.

```.cs
using System.Text;
using Dapr.Messaging.PublishSubscribe;
using Dapr.Messaging.PublishSubscribe.Extensions;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddDaprPubSubClient();
var app = builder.Build();

var messagingClient = app.Services.GetRequiredService<DaprPublishSubscribeClient>();

//Create a dynamic streaming subscription and subscribe with a timeout of 30 seconds and 10 seconds for message handling
var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
var subscription = await messagingClient.SubscribeAsync("pubsub", "myTopic",
    new DaprSubscriptionOptions(new MessageHandlingPolicy(TimeSpan.FromSeconds(10), TopicResponseAction.Retry)),
    HandleMessageAsync, cancellationTokenSource.Token);

await Task.Delay(TimeSpan.FromMinutes(1));

//When you're done with the subscription, simply dispose of it
await subscription.DisposeAsync();
return;

//Process each message returned from the subscription
Task<TopicResponseAction> HandleMessageAsync(TopicMessage message, CancellationToken cancellationToken = default)
{
    try
    {
        //Do something with the message
        Console.WriteLine(Encoding.UTF8.GetString(message.Data.Span));
        return Task.FromResult(TopicResponseAction.Success);
    }
    catch
    {
        return Task.FromResult(TopicResponseAction.Retry);
    }
}
```

--------------------------------

### Run Node.js publisher application

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Launches a Dapr sidecar and the Node.js publisher application. Navigate to the directory containing the code before running.

```bash
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 npm start
```

--------------------------------

### Dapr-generated CloudEvent with JSON data

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-cloudevents

This example demonstrates a CloudEvent generated by Dapr for a publish operation. It includes a W3C traceid unique to the message, and the data is serialized as JSON.

```json
{
  "topic": "orders",
  "pubsubname": "order_pub_sub",
  "traceid": "00-113ad9c4e42b27583ae98ba698d54255-e3743e35ff56f219-01",
  "tracestate": "",
  "data": {
    "orderId": 1
  },
  "id": "5929aaac-a5e2-4ca1-859c-edfe73f11565",
  "specversion": "1.0",
  "datacontenttype": "application/json; charset=utf-8",
  "source": "checkout",
  "type": "com.dapr.event.sent",
  "time": "2020-09-23T06:23:21Z",
  "traceparent": "00-113ad9c4e42b27583ae98ba698d54255-e3743e35ff56f219-01"
}
```

--------------------------------

### Handle Orders Topic in .NET

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Implement an HTTP POST endpoint in your .NET application to receive messages from the 'orders' topic, matching the route defined in the subscription.

```csharp
//Subscribe to a topic 
[HttpPost("orders")]
public void getCheckout([FromBody] int orderId)
{
    Console.WriteLine("Subscriber received : " + orderId);
}

```

--------------------------------

### Run Python Publisher Application

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Launches a Dapr sidecar and a Python publisher application. This command specifies the app protocol as gRPC.

```bash
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --app-protocol grpc python3 OrderProcessingService.py

```

--------------------------------

### Apply Subscription Component in Kubernetes

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Apply the declarative subscription component to your Kubernetes cluster using `kubectl apply`.

```bash
kubectl apply -f subscription.yaml

```

--------------------------------

### Handle Orders Topic in Java

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Implement a Spring Boot controller endpoint in your Java application to receive CloudEvents from the 'orders' topic.

```java
import io.dapr.client.domain.CloudEvent;

 //Subscribe to a topic
@PostMapping(path = "/orders")
public Mono<Void> getCheckout(@RequestBody(required = false) CloudEvent<String> cloudEvent) {
    return Mono.fromRunnable(() -> {
        try {
            log.info("Subscriber received: " + cloudEvent.getData());
        }
    });
}

```

--------------------------------

### Dapr-generated CloudEvent with XML data

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-cloudevents

This example shows data as XML content within a CloudEvent message, serialized as JSON. It includes fields like topic, pubsubname, traceid, and other standard CloudEvent attributes.

```json
{
  "topic": "orders",
  "pubsubname": "order_pub_sub",
  "traceid": "00-113ad9c4e42b27583ae98ba698d54255-e3743e35ff56f219-01",
  "tracestate": "",
  "data" : "<note><to></to><from>user2</from><message>Order</message></note>",
  "id" : "id-1234-5678-9101",
  "specversion" : "1.0",
  "datacontenttype" : "text/xml",
  "subject" : "Test XML Message",
  "source" : "https://example.com/message",
  "type" : "xml.message",
   "time" : "2020-09-23T06:23:21Z"
}
```

--------------------------------

### StatefulSet with Dapr enabled

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-subscribe-statefulset

Deploy a StatefulSet with Dapr annotations to enable pub/sub functionality. Ensure the app ID and port are correctly configured.

```yaml
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: python-subscriber
spec:
  selector:
    matchLabels:
      app: python-subscriber  # has to match .spec.template.metadata.labels
  serviceName: "python-subscriber"
  replicas: 3
  template:
    metadata:
      labels:
        app: python-subscriber # has to match .spec.selector.matchLabels
      annotations:
        dapr.io/enabled: "true"
        dapr.io/app-id: "python-subscriber"
        dapr.io/app-port: "5001"
    spec:
      containers:
      - name: python-subscriber
        image: ghcr.io/dapr/samples/pubsub-python-subscriber:latest
        ports:
        - containerPort: 5001
        imagePullPolicy: Always

```

--------------------------------

### Publish Raw Message using .NET SDK

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-raw

Publish a raw message using the Dapr .NET SDK. Set `rawPayload` to `true` in the metadata to disable CloudEvent wrapping. This example uses a custom message class.

```.NET
using Dapr.Client;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddControllers().AddDapr();

var app = builder.Build();

app.MapPost("/publish", async (DaprClient daprClient) =>
{
    var message = new Message(
        Guid.NewGuid().ToString(),
        $"Hello at {DateTime.UtcNow}",
        DateTime.UtcNow
    );

    await daprClient.PublishEventAsync(
        "pubsub",           // pubsub name
        "messages",         // topic name
        message,           // message data
        new Dictionary<string, string> 
        { 
            { "rawPayload", "true" },
            { "content-type", "application/json" }
        }
    );
    
    return Results.Ok(message);
});

app.Run();

```

--------------------------------

### Subscribe to Orders Topic in Go

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Define a Subscription object with pubsub name, topic, and route. Use AddTopicEventHandler to register the handler function for incoming messages.

```go
//dependencies
import (
	"log"
	"net/http"
	"context"

	"github.com/dapr/go-sdk/service/common"
	daprd "github.com/dapr/go-sdk/service/http"
)

//code
var sub = &common.Subscription{
	PubsubName: "order-pub-sub",
	Topic:      "orders",
	Route:      "/checkout",
}

func main() {
	s := daprd.NewService(":6002")
   //Subscribe to a topic
	if err := s.AddTopicEventHandler(sub, eventHandler); err != nil {
		log.Fatalf("error adding topic subscription: %v", err)
	}
	if err := s.Start(); err != nil && err != http.ErrServerClosed {
		log.Fatalf("error listenning: %v", err)
	}
}

func eventHandler(ctx context.Context, e *common.TopicEvent) (retry bool, err error) {
	log.Printf("Subscriber received: %s", e.Data)
	return false, nil
}
```

```bash
dapr run --app-id checkout --app-port 6002 --dapr-http-port 3602 --dapr-grpc-port 60002 go run CheckoutService.go
```

--------------------------------

### Subscribe to a Topic and Process Messages

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Use this method to subscribe to a topic and manually process incoming messages. Ensure to call `sub.Close()` and signal message processing with `msg.Success()`, `msg.Retry()`, or `msg.Drop()`.

```go
package main

import (
	"context"
	"log"

	"github.com/dapr/go-sdk/client"
)

func main() {
	cl, err := client.NewClient()
	if err != nil {
		log.Fatal(err)
	}

	sub, err := cl.Subscribe(context.Background(), client.SubscriptionOptions{
		PubsubName: "pubsub",
		Topic:      "orders",
	})
	if err != nil {
		panic(err)
	}
	// Close must always be called.
	defer sub.Close()

	for {
		msg, err := sub.Receive()
		if err != nil {
			panic(err)
		}

		// Process the event

		// We _MUST_ always signal the result of processing the message, else the
		// message will not be considered as processed and will be redelivered or
		// dead lettered.
		// msg.Retry()
		// msg.Drop()
		if err := msg.Success(); err != nil {
			panic(err)
		}
	}
}
```

--------------------------------

### Programmatic Subscription to Raw Events in Java

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-raw

Use the `@Topic` annotation in a Java Spring Boot application to subscribe to raw messages, specifying `rawPayload: 'true'` within the metadata. This example shows two different topics being subscribed to.

```java
@RequestMapping("/consumer")
@RestController
public class MessageConsumerController {

    @PostMapping
    @ResponseStatus(HttpStatus.OK)
    @Topic(pubsubName = "pubsub", name = "messages", metadata = "{\"rawPayload\":\"true\", \"content-type\": \"application/json\"}")
    public void consume(@RequestBody Message message) {
        System.out.println("Message received: " + message);
    }

    @PostMapping
    @ResponseStatus(HttpStatus.OK)
    @Topic(pubsubName = "pubsub", name = "another-topic", metadata = """
            {\"rawPayload\": \"true\", \"content-type\": \"application/json\"}
            """) // Using Java 15 text block
    public void consumeAnother(@RequestBody Message message) {
        System.out.println("Message received: " + message);
    }
}

```

--------------------------------

### Publish event using Dapr Go SDK

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Publishes events to a specified topic using the Dapr Go SDK. Ensure the Dapr client is initialized and the necessary dependencies are imported.

```go
//dependencies
import (
	"context"
	"log"
	"math/rand"
	"time"
	"strconv"
	dapr "github.com/dapr/go-sdk/client"
)

//code
var (
	PUBSUB_NAME = "order-pub-sub"
	TOPIC_NAME  = "orders"
)

func main() {
	for i := 0; i < 10; i++ {
		time.Sleep(5000)
		orderId := rand.Intn(1000-1) + 1
		client, err := dapr.NewClient()
		if err != nil {
			panic(err)
		}
		defer client.Close()
		ctx := context.Background()
    //Using Dapr SDK to publish a topic
		if err := client.PublishEvent(ctx, PUBSUB_NAME, TOPIC_NAME, []byte(strconv.Itoa(orderId))); 
		err != nil {
			panic(err)
		}

		log.Println("Published data: " + strconv.Itoa(orderId))
	}
}
```

--------------------------------

### Publish Raw Message using Java SDK

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-raw

Publish a raw message using the Dapr Java SDK. The `rawPayload` metadata must be set to `true` to send the data without CloudEvent encapsulation. This example uses a custom `Message` object.

```Java
@RestController
@PathMapping("/publish")
public class PublishController {

    @Inject
    DaprClient client;

    @PostMapping
    public void sendRawMessage() {

        Map<String, String> metadata = new HashMap<>();
        metatada.put("content-type", "application/json");
        metadata.put("rawPayload", "true");

        Message message = new Message(UUID.random().toString(), "Hello from Dapr");

        client.publishEvent(
            "pubsub", // pubsub name
            "messages", // topic name
            message, // message data
            metadata) // metadata
                .block(); // wait for completion
    }
}


```

--------------------------------

### Programmatic Subscription Handler (Minimal API - .NET)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Defines a Minimal API endpoint in .NET to subscribe to the 'orders' topic and process incoming messages.

```.cs
// Dapr subscription in [Topic] routes orders topic to this route
app.MapPost("/orders", [Topic("pubsub", "orders")] (Order order) => {
    Console.WriteLine("Subscriber received : " + order);
    return Results.Ok(order);
});

```

--------------------------------

### Subscribe to a Topic with a Handler Function

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

This method subscribes to a topic and uses a provided handler function to process events. Remember to call the `stop` function returned by `SubscribeWithHandler` and ensure your handler returns an appropriate `SubscriptionResponseStatus`.

```go
package main

import (
	"context"
	"log"

	"github.com/dapr/go-sdk/client"
	"github.com/dapr/go-sdk/service/common"
)

func main() {
	cl, err := client.NewClient()
	if err != nil {
		log.Fatal(err)
	}

	stop, err := cl.SubscribeWithHandler(context.Background(),
		client.SubscriptionOptions{
			PubsubName: "pubsub",
			Topic:      "orders",
		},
		eventHandler,
	)
	if err != nil {
		panic(err)
	}

	// Stop must always be called.
	defer stop()

	<-make(chan struct{})
}

func eventHandler(e *common.TopicEvent) common.SubscriptionResponseStatus {
	// Process message here
    // common.SubscriptionResponseStatusRetry
    // common.SubscriptionResponseStatusDrop
			common.SubscriptionResponseStatusDrop, status)
	}

	return common.SubscriptionResponseStatusSuccess
}
```

--------------------------------

### Programmatic Subscription to Raw Events in Python (Flask)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-raw

Set up a Python Flask application to subscribe to raw messages by including `rawPayload: 'true'` in the subscription metadata. The subscriber route will receive the raw message.

```python
import flask
from flask import request, jsonify
from flask_cors import CORS
import json
import sys

app = flask.Flask(__name__)
CORS(app)

@app.route('/dapr/subscribe', methods=['GET'])
def subscribe():
    subscriptions = [{'pubsubname': 'pubsub',
                      'topic': 'deathStarStatus',
                      'route': 'dsstatus',
                      'metadata': {
                          'rawPayload': 'true',
                      } }]
    return jsonify(subscriptions)

@app.route('/dsstatus', methods=['POST'])
def ds_subscriber():
    print(request.json, flush=True)
    return json.dumps({'success':True}), 200, {'ContentType':'application/json'}

app.run()

```

--------------------------------

### Programmatic Subscription Handler (.NET)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Defines a .NET controller action to handle incoming 'orders' topic messages from the 'pubsub' component.

```.cs
using Dapr.Client;
using Microsoft.AspNetCore.Mvc;

[Topic("pubsub", "orders")]
[HttpPost("/orders")]
public async Task<ActionResult<Order>>Checkout(Order order, [FromServices] DaprClient daprClient)
{
    // Logic
    return order;
}

```

--------------------------------

### Handle Orders Topic in Python

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Implement a Flask route in your Python application to receive messages from the 'orders' topic.

```python
from cloudevents.sdk.event import v1

#Subscribe to a topic 
@app.route('/orders', methods=['POST'])
def checkout(event: v1.Event) -> None:
    data = json.loads(event.Data())
    logging.info('Subscriber received: ' + str(data))

```

--------------------------------

### Subscribe to Orders Topic in C#

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Use the [Topic] attribute to subscribe to a topic from a specific pub/sub component. Ensure the application is configured to listen on the specified port.

```csharp
using System.Collections.Generic;
using System.Threading.Tasks;
using System;
using Microsoft.AspNetCore.Mvc;
using Dapr;
using Dapr.Client;

namespace CheckoutService.Controllers;

[ApiController]
public sealed class CheckoutServiceController : ControllerBase
{
    //Subscribe to a topic called "orders" from the "order-pub-sub" compoennt 
    [Topic("order-pub-sub", "orders")]
    [HttpPost("checkout")]
    public void GetCheckout([FromBody] int orderId)
    {
        Console.WriteLine("Subscriber received : " + orderId);
    }
}
```

```bash
dapr run --app-id checkout --app-port 6002 --dapr-http-port 3602 --dapr-grpc-port 60002 --app-protocol https dotnet run
```

--------------------------------

### Programmatic Subscription Handler (Go)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Implements a Go HTTP handler to define Dapr pub/sub subscriptions and respond to incoming messages.

```go
package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"

	"github.com/gorilla/mux"
)

const appPort = 3000

type subscription struct {
	PubsubName string            `json:"pubsubname"`
	Topic      string            `json:"topic"`
	Metadata   map[string]string `json:"metadata,omitempty"`
	Routes     routes            `json:"routes"`
}

type routes struct {
	Rules   []rule `json:"rules,omitempty"`
	Default string `json:"default,omitempty"`
}

type rule struct {
	Match string `json:"match"`
	Path  string `json:"path"`
}

// This handles /dapr/subscribe
func configureSubscribeHandler(w http.ResponseWriter, _ *http.Request) {
	t := []subscription{
		{
			PubsubName: "pubsub",
			Topic:      "orders",
			Routes: routes{
				Rules: []rule{
					{
						Match: `event.type == "order"`,
						Path:  "/orders",
					},
				},
				Default: "/orders",
			},
		},
	}

	w.WriteHeader(http.StatusOK)
	json.NewEncoder(w).Encode(t)
}

func main() {
	router := mux.NewRouter().StrictSlash(true)
	router.HandleFunc("/dapr/subscribe", configureSubscribeHandler).Methods("GET")
	log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", appPort), router))
}

```

--------------------------------

### Publish a Binary CloudEvent using HTTP API (Bash)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-cloudevents

Publish a binary CloudEvent by setting the `Content-Type` to `application/octet-stream` and providing CloudEvent attributes as transport metadata prefixed with `ce_`. The message body contains the raw binary payload.

```bash
curl -X POST http://localhost:3500/v1.0/publish/order-pub-sub/orders \
  -H "Content-Type: application/octet-stream" \
  -H "ce_specversion: 1.0" \
  -H "ce_type: com.example.order.created" \
  -H "ce_source: urn:example:/checkout" \
  -H "ce_id: 2a8bbf52-1222-4c2c-85f0-8a8875c7bc10" \
  -H "ce_subject: orders/100" \
  --data-binary $'
'
```

--------------------------------

### Publish messages in bulk using .NET

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-bulk

Publishes a list of objects as bulk events asynchronously. Checks the response for any failed entries and logs them.

```.NET
using System;
using System.Collections.Generic;
using Dapr.Client;

const string PubsubName = "my-pubsub-name";
const string TopicName = "topic-a";
IReadOnlyList<object> BulkPublishData = new List<object>() {
    new { Id = "17", Amount = 10m },
    new { Id = "18", Amount = 20m },
    new { Id = "19", Amount = 30m }
};

using var client = new DaprClientBuilder().Build();

var res = await client.BulkPublishEventAsync(PubsubName, TopicName, BulkPublishData);
if (res == null) {
    throw new Exception("null response from dapr");
}
if (res.FailedEntries.Count > 0)
{
    Console.WriteLine("Some events failed to be published!");
    foreach (var failedEntry in res.FailedEntries)
    {
        Console.WriteLine("EntryId: " + failedEntry.Entry.EntryId + " Error message: " +
                          failedEntry.ErrorMessage);
    }
}
else
{
    Console.WriteLine("Published all events!");
}


```

--------------------------------

### Python: Flask App for Bulk Subscription

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-bulk

A Python Flask application demonstrating how to configure and handle bulk subscriptions for Dapr Pub/Sub. It defines subscription routes and processes incoming bulk messages.

```python
import json
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/dapr/subscribe', methods=['GET'])
def subscribe():
    # Define the bulk subscribe configuration
    subscriptions = [{
        "pubsubname": "pubsub",
        "topic": "TOPIC_A",
        "route": "/checkout",
        "bulkSubscribe": {
            "enabled": True,
            "maxMessagesCount": 3,
            "maxAwaitDurationMs": 40
        }
    }]
    print('Dapr pub/sub is subscribed to: ' + json.dumps(subscriptions))
    return jsonify(subscriptions)


# Define the endpoint to handle incoming messages
@app.route('/checkout', methods=['POST'])
def checkout():
    messages = request.json
    print(messages)
    for message in messages:
        print(f"Received message: {message}")
    return json.dumps({'success': True}), 200, {'ContentType': 'application/json'}

if __name__ == '__main__':
    app.run(port=5000)

```

--------------------------------

### Kubernetes Pub/Sub Component Configuration (RabbitMQ)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Configure a RabbitMQ pub/sub component for Kubernetes deployments. This YAML includes connection details like connection string, protocol, and credentials.

```yaml
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: order-pub-sub
spec:
  type: pubsub.rabbitmq
  version: v1
  metadata:
  - name: connectionString
    value: "amqp://localhost:5672"
  - name: protocol
    value: amqp  
  - name: hostname
    value: localhost 
  - name: username
    value: username
  - name: password
    value: password 
  - name: durable
    value: "false"
  - name: deletedWhenUnused
    value: "false"
  - name: autoAck
    value: "false"
  - name: reconnectWait
    value: "0"
  - name: concurrency
    value: parallel
scopes:
  - orderprocessing
  - checkout

```

--------------------------------

### Enabling Dapr Subscribe Handler (.NET Minimal API)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Configures the Dapr subscribe handler endpoint in a .NET Minimal API application startup.

```.cs
app.UseEndpoints(endpoints =>
{
    endpoints.MapSubscribeHandler();
});

```

--------------------------------

### Programmatic Subscription Definition and Handler (Node.js Express)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/subscription-methods

Sets up Dapr pub/sub subscriptions and an Express route for handling 'orders' topic messages.

```javascript
const express = require('express')
const bodyParser = require('body-parser')
const app = express()
app.use(bodyParser.json({ type: 'application/*+json' }));

const port = 3000

app.get('/dapr/subscribe', (req, res) => {
  res.json([
    {
      pubsubname: "pubsub",
      topic: "orders",
      routes: {
        rules: [
          {
            match: 'event.type == "order"',
            path: '/orders'
          },
        ],
        default: '/products'
      }
    }
  ]);
})

app.post('/orders', (req, res) => {
  console.log(req.body);
  res.sendStatus(200);
});

app.listen(port, () => console.log(`consumer app listening on port ${port}!`))

```

--------------------------------

### Configure Redis Pub/Sub Component with Namespace Consumer ID

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-namespace

This configuration sets up a Redis pub/sub component, using `{namespace}` in the `consumerID` metadata. This allows Dapr to namespace consumer groups based on the Kubernetes namespace, enabling multi-tenancy.

```yaml
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: pubsub
spec:
  type: pubsub.redis
  version: v1
  metadata:
  - name: redisHost
    value: localhost:6379
  - name: redisPassword
    value: ""
  - name: consumerID
    value: "{namespace}"

```

--------------------------------

### Self-Hosted Pub/Sub Component Configuration (RabbitMQ)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-publish-subscribe

Configure a RabbitMQ pub/sub component for self-hosted Dapr environments. This YAML defines the connection details and settings for the RabbitMQ broker.

```yaml
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: order-pub-sub
spec:
  type: pubsub.rabbitmq
  version: v1
  metadata:
  - name: host
    value: "amqp://localhost:5672"
  - name: durable
    value: "false"
  - name: deletedWhenUnused
    value: "false"
  - name: autoAck
    value: "false"
  - name: reconnectWait
    value: "0"
  - name: concurrency
    value: parallel
scopes:
  - orderprocessing
  - checkout

```

--------------------------------

### Publish a CloudEvent using HTTP API (Bash)

Source: https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-cloudevents

Publish a CloudEvent to a topic using the Dapr HTTP API. The `Content-Type` header must be `application/cloudevent+json`. Dapr automatically adds missing required fields if not provided.

```bash
curl -X POST http://localhost:3601/v1.0/publish/order-pub-sub/orders -H "Content-Type: application/cloudevents+json" -d '{"specversion" : "1.0", "type" : "com.dapr.cloudevent.sent", "source" : "testcloudeventspubsub", "subject" : "Cloud Events Test", "id" : "someCloudEventId", "time" : "2021-08-02T09:00:00Z", "datacontenttype" : "application/cloudevents+json", "data" : {"orderId": "100"}}'
```