Tutorial: Dapr.Messaging by example

• 1: Part 1: Publish events with Dapr.Messaging
• 2: Part 2: Declarative streaming subscriptions
• 3: Part 3: Content-based routing and dead-letter topics
• 4: Part 4: Bulk subscriptions
• 5: Part 5: gRPC AppCallback push subscriptions
• 6: Part 6: HTTP push subscriptions
• 7: Part 7: Dynamic streaming subscriptions

1 - Part 1: Publish events with Dapr.Messaging

The publishing example introduces IDaprPublishSubscribeClient, the publishing API in Dapr.Messaging.

Register the client

AddDaprMessaging() registers the client with dependency injection:

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

Inject IDaprPublishSubscribeClient into an endpoint, service, or background worker. The client is intended to be reused rather than created for every message.

Publish typed events

app.MapPost("/orders", async (
    OrderPlaced order,
    IDaprPublishSubscribeClient client,
    CancellationToken cancellationToken) =>
{
    await client.PublishEventAsync(
        "pubsub",
        "orders",
        order,
        cancellationToken);

    return Results.Accepted($"/orders/{order.OrderId}");
});

Typed payloads are serialized as JSON and published as Dapr CloudEvents. Pass the request cancellation token through the SDK call.

Add metadata and publish in bulk

Use PublishOptions for CloudEvent attributes and component metadata such as a message TTL:

var options = new PublishOptions
{
    Metadata =
    {
        ["cloudevent.type"] = "priority.order",
        ["ttlInSeconds"] = "60"
    }
};

await client.PublishEventAsync("pubsub", "orders", priorityOrder, options, cancellationToken);

For batches, inspect the response because publishing can partially succeed:

var response = await client.BulkPublishEventAsync(
    "pubsub",
    "orders",
    orders,
    cancellationToken: cancellationToken);

foreach (var failed in response.FailedEntries)
{
    logger.LogError("Failed entry {EntryId}: {Error}",
        failed.Entry.EntryId, failed.ErrorMessage);
}

Publish raw bytes

Use PublishByteEventAsync when the payload is already serialized or uses a non-JSON format:

var payload = Encoding.UTF8.GetBytes("RAW_PAYLOAD_DATA_SAMPLE");

await client.PublishByteEventAsync(
    "pubsub",
    "raw-orders",
    payload,
    dataContentType: "text/plain",
    cancellationToken: cancellationToken);

Test the publisher

The unit tests use Moq to verify that the client receives the expected topic, payload, options, and content type. They do not start Dapr. The integration tests use DaprHarnessBuilder.BuildPubSub(), connect a real client to the harness’s dynamic gRPC and HTTP ports, and verify typed, bulk, and raw publishing against the sidecar and Redis.

Run the example from the SDK repository:

dotnet test examples\Messaging\01-Publishing\Publishing.Example01.Tests\Publishing.Example01.Tests.csproj

Next

• Part 2: Declarative streaming subscriptions
• Publish events how-to

2 - Part 2: Declarative streaming subscriptions

The streaming example shows the recommended declarative subscription model for fixed topics.

Define a handler

Implement ITopicHandler<TMessage> and decorate the handler with DaprTopic:

[DaprTopic("pubsub", "orders", Delivery = DeliveryMode.Streaming)]
public sealed class OrderProcessingHandler : ITopicHandler<OrderPlaced>
{
    public Task<TopicResponseAction> HandleAsync(
        OrderPlaced order,
        TopicContext context,
        CancellationToken cancellationToken)
    {
        if (order.Quantity <= 0 || string.IsNullOrWhiteSpace(order.ItemSku))
        {
            return Task.FromResult(TopicResponseAction.Drop);
        }

        if (order.ItemSku.StartsWith("RETRY-", StringComparison.OrdinalIgnoreCase))
        {
            return Task.FromResult(TopicResponseAction.Retry);
        }

        return Task.FromResult(TopicResponseAction.Success);
    }
}

Success acknowledges the message, Retry requests redelivery for a transient failure, and Drop rejects an invalid or poison message. The handler also receives TopicContext, which contains the pub/sub name, topic, message ID, headers, and metadata.

Configure the application

builder.Services.AddDaprMessaging();

The source generator discovers the attributed handlers, registers them in DI, and starts the streaming subscriber hosted service. Streaming is application-initiated, so the application does not need an inbound HTTP or gRPC endpoint for delivery. The SDK manages the stream and reconnects when the connection drops.

Test the handler and the stream

Unit tests construct handlers with mocked loggers and a small TopicContext, then assert Success, Retry, and Drop for representative inputs. This keeps validation and acknowledgement logic fast and independent of infrastructure.

The integration test uses DaprTestApplicationBuilder and configures AddDaprMessaging with the harness’s gRPC endpoint. It publishes through the sidecar and waits on a test log sink for successful processing, inventory handling, and dropping an invalid order. This verifies the hosted service and real streaming delivery without requiring an application ingress port.

dotnet test examples\Messaging\02-Streaming\Streaming.Example02.Tests\Streaming.Example02.Tests.csproj

Next

• Part 3: Content-based routing and dead-letter topics
• Subscribe to topics how-to

3 - Part 3: Content-based routing and dead-letter topics

The routing example uses multiple generated subscriptions to route shipments and handle rejected messages.

Add CEL matching rules

Match is evaluated against the CloudEvent. Priority controls rule ordering, with lower values evaluated first:

[DaprTopic(
    "pubsub",
    "express-shipments",
    Match = "event.data.priorityTier == 'express'",
    Priority = 1,
    DeadLetterTopic = "deadletter-shipments")]
[DaprTopicMetadata("routingType", "express-tier")]
public sealed class ExpressShippingHandler : ITopicHandler<ShipmentPackage>
{
    public Task<TopicResponseAction> HandleAsync(
        ShipmentPackage shipment,
        TopicContext context,
        CancellationToken cancellationToken)
    {
        return Task.FromResult(
            shipment.WeightKg <= 0
                ? TopicResponseAction.Drop
                : TopicResponseAction.Success);
    }
}

The example adds an international rule using event.data.destinationCountry != 'US' and a priority-10 catch-all handler for standard domestic shipments. DaprTopicMetadata adds custom key-value data to the generated subscription manifest.

Handle the dead-letter topic

Subscribe to the configured dead-letter topic like any other topic:

[DaprTopic("pubsub", "deadletter-shipments")]
public sealed class DeadLetterShipmentHandler : ITopicHandler<ShipmentPackage>
{
    public Task<TopicResponseAction> HandleAsync(
        ShipmentPackage shipment,
        TopicContext context,
        CancellationToken cancellationToken)
    {
        // Record or alert on the rejected message.
        return Task.FromResult(TopicResponseAction.Success);
    }
}

Messages can reach the dead-letter topic after a handler returns Drop or after retry delivery is exhausted according to the component configuration.

Test routing decisions

The unit tests invoke each handler directly and cover an express shipment, an invalid weight, an international destination, a standard fallback, and a dead-letter message. These tests prove business decisions without pretending to validate CEL evaluation.

The integration tests are the place to validate the generated subscription manifest, CEL evaluation, priority ordering, and dead-letter forwarding against a real Dapr sidecar and broker.

dotnet test examples\Messaging\03-RoutingAndDeadLetter\Routing.Example03.Tests\Routing.Example03.Tests.csproj

Next

• Part 4: Bulk subscriptions
• Configuration and usage guide

4 - Part 4: Bulk subscriptions

The bulk subscription example combines bulk publishing with a subscription configured for high-throughput ingestion.

Configure batching

Set BulkSubscribe and its limits on the topic attribute:

[DaprTopic(
    "pubsub",
    "telemetry",
    BulkSubscribe = true,
    MaxMessagesCount = 50,
    MaxAwaitDurationMs = 500)]
public sealed class TelemetryBulkHandler : ITopicHandler<DeviceTelemetry>
{
    public Task<TopicResponseAction> HandleAsync(
        DeviceTelemetry reading,
        TopicContext context,
        CancellationToken cancellationToken)
    {
        if (reading.TemperatureCelsius < -273.15 ||
            reading.HumidityPercent is < 0 or > 100)
        {
            return Task.FromResult(TopicResponseAction.Drop);
        }

        return Task.FromResult(TopicResponseAction.Success);
    }
}

MaxMessagesCount bounds the batch size and MaxAwaitDurationMs bounds how long Dapr waits before delivering a partial batch. Choose values based on throughput, latency, and broker behavior. The handler remains responsible for the individual message decision.

Publish single events or batches

await client.PublishEventAsync("pubsub", "telemetry", reading, cancellationToken);

var response = await client.BulkPublishEventAsync(
    "pubsub",
    "telemetry",
    readings,
    options: null,
    cancellationToken: cancellationToken);

Inspect response.FailedEntries when bulk publishing because a batch can contain partial failures.

Test batching-related behavior

The unit tests cover valid telemetry and invalid temperature and humidity values. The integration tests publish readings and batches through a real Dapr pub/sub harness to validate the generated bulk subscription configuration and delivery semantics.

dotnet test examples\Messaging\04-BulkSubscribe\BulkSubscribe.Example04.Tests\BulkSubscribe.Example04.Tests.csproj

Next

• Part 5: gRPC AppCallback push
• Dapr Messaging configuration and usage guide

5 - Part 5: gRPC AppCallback push subscriptions

The AppCallback example uses DeliveryMode.Programmatic for sidecar-to-application gRPC push.

Define a programmatic subscriber

[DaprTopic("pubsub", "payments", Delivery = DeliveryMode.Programmatic)]
public sealed class PaymentProcessingHandler : ITopicHandler<PaymentReceived>
{
    public Task<TopicResponseAction> HandleAsync(
        PaymentReceived payment,
        TopicContext context,
        CancellationToken cancellationToken)
    {
        return Task.FromResult(
            payment.Amount <= 0
                ? TopicResponseAction.Drop
                : TopicResponseAction.Success);
    }
}

In this mode Dapr initiates the gRPC call to the application’s AppCallback service. Unlike streaming subscriptions, the app must expose a reachable gRPC port.

Map the generated endpoint

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

var app = builder.Build();
app.MapDaprMessaging();
app.Run();

MapDaprMessaging() exposes the generated AppCallback gRPC service used by the sidecar to list subscriptions and deliver events. Run the application with an app port and gRPC protocol, for example:

dapr run --app-id appcallback-example --app-port 5000 --app-protocol grpc `
  --dapr-grpc-port 50001 -- dotnet run

Test both layers

The unit tests verify that valid payments succeed and zero or negative amounts are dropped. The integration tests configure the test application with a gRPC server, start the harness, publish real payments, and observe the handler result. This distinction is important: unit tests validate business rules; integration tests validate AppCallback registration, reachability, and delivery.

dotnet test examples\Messaging\05-AppCallbackPush\AppCallback.Example05.Tests\AppCallback.Example05.Tests.csproj

Next

• Part 6: HTTP push subscriptions
• Subscribe to topics how-to

6 - Part 6: HTTP push subscriptions

The HTTP subscription example shows DeliveryMode.Http for applications that use normal HTTP ingress or reverse proxies.

Define a routed handler

[DaprTopic(
    "pubsub",
    "invoices",
    Delivery = DeliveryMode.Http,
    Route = "api/events/invoices")]
public sealed class InvoiceProcessingHandler : ITopicHandler<InvoiceGenerated>
{
    public Task<TopicResponseAction> HandleAsync(
        InvoiceGenerated invoice,
        TopicContext context,
        CancellationToken cancellationToken)
    {
        return Task.FromResult(
            invoice.TotalAmount <= 0
                ? TopicResponseAction.Drop
                : TopicResponseAction.Success);
    }
}

Map the generated routes

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

var app = builder.Build();
app.MapDaprMessaging();
app.Run();

The generated mapping exposes GET /dapr/subscribe for Dapr to discover subscriptions and maps POST /api/events/invoices for the topic’s CloudEvents. The route is generated from the Route property; it does not need a separate controller or action.

Run with the application’s HTTP port:

dapr run --app-id http-subscriber-example --app-port 5000 `
  --dapr-grpc-port 50001 -- dotnet run

Inspect the generated manifest with curl http://localhost:5000/dapr/subscribe.

Test HTTP delivery

The unit tests invoke the invoice handler directly for valid, zero, and negative amounts. The integration tests use the test application and a real Dapr sidecar to verify subscription discovery and HTTP push delivery. This lets the test cover route mapping and CloudEvent dispatch without coupling business-rule tests to ASP.NET hosting.

dotnet test examples\Messaging\06-HttpSubscription\HttpSubscription.Example06.Tests\HttpSubscription.Example06.Tests.csproj

Next

• Part 7: Dynamic streaming subscriptions
• Dapr Messaging configuration and usage guide

7 - Part 7: Dynamic streaming subscriptions

The dynamic streaming example uses SubscribeAsync when topics or subscription lifetime cannot be fixed at compile time.

Register the client and worker

builder.Services.AddDaprPubSubClient();
builder.Services.AddHostedService<DynamicSubscriberWorker>();

The worker opens a subscription to a runtime-selected topic:

var options = new DaprSubscriptionOptions(
    new MessageHandlingPolicy(
        TimeoutDuration: TimeSpan.FromSeconds(10),
        DefaultResponseAction: TopicResponseAction.Retry))
{
    DeadLetterTopic = "tenant-events-dlq",
    ErrorHandler = exception =>
    {
        logger.LogWarning("Subscription error: {Message}", exception.Message);
        return Task.CompletedTask;
    }
};

await using var subscription = await client.SubscribeAsync(
    "pubsub",
    "tenant-events",
    options,
    HandleDynamicMessageAsync,
    stoppingToken);

await ((IDaprSubscription)subscription).Completion.WaitAsync(stoppingToken);

SubscribeAsync returns an async-disposable subscription and an IDaprSubscription.Completion task. A supervisor can await completion, log faults, delay, and open a new subscription. Always dispose the subscription and honor the host cancellation token.

Handle untyped messages safely

Dynamic subscriptions receive TopicMessage, so the handler is responsible for decoding and validating the payload:

try
{
    var tenantEvent = JsonSerializer.Deserialize<TenantEvent>(message.Data.Span);
    if (tenantEvent is null || string.IsNullOrWhiteSpace(tenantEvent.TenantId))
    {
        return Task.FromResult(TopicResponseAction.Drop);
    }

    return Task.FromResult(
        tenantEvent.EventType.StartsWith("RETRY-", StringComparison.OrdinalIgnoreCase)
            ? TopicResponseAction.Retry
            : TopicResponseAction.Success);
}
catch (JsonException)
{
    return Task.FromResult(TopicResponseAction.Drop);
}

Malformed JSON is a permanent input problem and is dropped; transient processing errors should request retry.

Test the supervisor and handler

The unit tests create TopicMessage instances with serialized valid, retry, invalid, and malformed payloads and assert the response action. The integration test registers the worker with DaprTestApplicationBuilder, publishes real tenant events, and waits on a concurrent log sink for processing and drop results. This pattern tests asynchronous background work without relying on arbitrary sleeps.

dotnet test examples\Messaging\07-DynamicStreaming\DynamicStreaming.Example07.Tests\DynamicStreaming.Example07.Tests.csproj

Next steps

• Dapr Messaging overview
• Subscribe to topics how-to