Dapr Messaging in the .NET SDK

• 1: How to: Publish events with the Dapr Messaging .NET SDK
• 2: How to: Author subscriptions and handle topic messages with the Dapr Messaging .NET SDK
• 3: Dapr Messaging configuration and usage
• 4: Tutorial: Dapr.Messaging by example
• 4.1: Part 1: Publish events with Dapr.Messaging
• 4.2: Part 2: Declarative streaming subscriptions
• 4.3: Part 3: Content-based routing and dead-letter topics
• 4.4: Part 4: Bulk subscriptions
• 4.5: Part 5: gRPC AppCallback push subscriptions
• 4.6: Part 6: HTTP push subscriptions
• 4.7: Part 7: Dynamic streaming subscriptions

<ItemGroup>
  <PackageReference Include="Dapr.Messaging" Version="..." />
</ItemGroup>

1 - How to: Publish events with the Dapr Messaging .NET SDK

This guide demonstrates how to publish events to Dapr pub/sub topics using the modern Dapr.Messaging SDK for .NET.

Prerequisites

• Dapr CLI installed
• Initialized Dapr environment
• .NET 8, .NET 9, or .NET 10 installed
• Dapr.Messaging package installed:

<ItemGroup>
  <PackageReference Include="Dapr.Messaging" Version="..." />
</ItemGroup>

Register the publishing client in Program.cs

Register Dapr messaging services and the pub/sub client using dependency injection:

var builder = WebApplication.CreateBuilder(args);

// Register Dapr Messaging services and the publish/subscribe client
builder.Services.AddDaprMessaging(options =>
{
    // Optional: configure sidecar connection endpoints or serializer
    options.DaprGrpcEndpoint = "http://localhost:50001";
});

var app = builder.Build();

Publish events

Inject IDaprPublishSubscribeClient (or DaprPublishSubscribeClient) into your controllers, minimal API route handlers, or background services.

1. Publish standard JSON events

To publish a serialized object to a topic, call PublishEventAsync<TData>:

public record Order(string Id, decimal Total, string CustomerEmail);

public sealed class OrderService(IDaprPublishSubscribeClient pubsub, ILogger<OrderService> logger)
{
    public async Task PlaceOrderAsync(Order order, CancellationToken cancellationToken = default)
    {
        logger.LogInformation("Publishing order {OrderId} to 'orders' topic", order.Id);

        // Publishes the order object serialized as JSON with Content-Type: application/json
        await pubsub.PublishEventAsync(
            pubsubName: "pubsub",
            topicName: "orders",
            data: order,
            cancellationToken: cancellationToken);
    }
}

2. Publish with custom CloudEvent metadata and PublishOptions

You can attach custom CloudEvent properties (such as event type, subject, traceparent, or component TTL metadata) using PublishOptions:

var options = new PublishOptions
{
    Id = Guid.NewGuid().ToString(),
    Type = "com.myapp.order.created",
    Subject = $"orders/{order.Id}",
    Metadata =
    {
        ["ttlInSeconds"] = "120", // Dapr message TTL
        ["cloudevent.source"] = "urn:service:checkout"
    }
};

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

3. Publish strongly-typed CloudEvents

When you want complete control over the CloudEvent envelope, wrap your payload in CloudEvent<TData>:

var cloudEvent = new CloudEvent<Order>(order)
{
    Id = Guid.NewGuid().ToString(),
    Source = new Uri("urn:service:checkout"),
    Type = "com.myapp.order.created",
    Subject = $"orders/{order.Id}",
    Time = DateTimeOffset.UtcNow,
    TraceId = Activity.Current?.TraceId.ToString()
};

// Publishes with Content-Type: application/cloudevents+json
await pubsub.PublishEventAsync("pubsub", "orders", cloudEvent, cancellationToken);

4. Publish raw byte payloads

If you are transmitting binary data, pre-serialized payloads, or proprietary encodings, use PublishByteEventAsync:

byte[] binaryPayload = GetCompressedPayload();

await pubsub.PublishByteEventAsync(
    pubsubName: "pubsub",
    topicName: "raw-telemetry",
    data: binaryPayload,
    dataContentType: "application/octet-stream",
    cancellationToken: cancellationToken);

5. Publish empty notification events

To publish a notification event with no payload body:

await pubsub.PublishEventAsync("pubsub", "cache-invalidated", cancellationToken);

Bulk publish multiple events

Bulk publishing lets you publish a batch of events to a topic in a single request, significantly reducing round-trip overhead:

var orders = new List<Order>
{
    new("101", 49.99m, "user1@example.com"),
    new("102", 99.50m, "user2@example.com"),
    new("103", 14.25m, "user3@example.com")
};

var response = await pubsub.BulkPublishEventAsync(
    pubsubName: "pubsub",
    topicName: "orders",
    events: orders,
    cancellationToken: cancellationToken);

if (response.FailedEntries.Count > 0)
{
    foreach (var failed in response.FailedEntries)
    {
        logger.LogError(
            "Failed to publish entry {EntryId} (Order {OrderId}): {Error}",
            failed.Entry.EntryId,
            failed.Entry.EventData.Id,
            failed.ErrorMessage);
    }
}
else
{
    logger.LogInformation("All {Count} orders published successfully", orders.Count);
}

Using the publisher without dependency injection

While dependency injection is recommended, you can construct a DaprPublishSubscribeClient instance directly using DaprPublishSubscribeClientBuilder:

var client = new DaprPublishSubscribeClientBuilder()
    .UseGrpcEndpoint("http://localhost:50001")
    .UseDaprApiToken("my-token")
    .Build();

await client.PublishEventAsync("pubsub", "orders", new Order("101", 19.99m, "a@b.com"));

Best practices

• Reuse the client: Register IDaprPublishSubscribeClient once as a singleton; do not instantiate and dispose clients per message.
• Pass cancellation tokens: Always pass a CancellationToken to handle application cancellation and HTTP request aborts gracefully.
• Inspect bulk publish responses: Bulk publishing returns partial success when only some entries fail. Always inspect response.FailedEntries.
• Propagate distributed tracing: The SDK automatically propagates active OpenTelemetry/W3C trace contexts (Activity.Current) in CloudEvent headers.

Next steps

• Tutorial: Dapr.Messaging by example
• How-To: Author subscriptions and handle topic messages
• Dapr Messaging configuration and usage reference
• Dapr Pub/Sub building block overview

2 - How to: Author subscriptions and handle topic messages with the Dapr Messaging .NET SDK

This guide demonstrates how to author and manage Dapr topic subscriptions using the modern Dapr.Messaging SDK for .NET. You will learn how to author strongly-typed topic handlers with ITopicHandler<TMessage>, configure delivery modes and routing rules with [DaprTopic], leverage compile-time source generation, and manage dynamic streaming subscriptions.

Prerequisites

• Dapr CLI installed
• Initialized Dapr environment
• .NET 8, .NET 9, or .NET 10 installed
• Dapr.Messaging package installed:

<ItemGroup>
  <PackageReference Include="Dapr.Messaging" Version="..." />
</ItemGroup>

Define a topic handler

Topic subscribers implement the strongly-typed ITopicHandler<TMessage> interface (or its generic variant ITopicHandler<TMessage, TResult>) and are decorated with the [DaprTopic] attribute.

Why use ITopicHandler?

• Strongly-typed payloads: Automatically deserializes incoming JSON or CloudEvent payloads into strongly-typed .NET models (TMessage) without manual boilerplate.
• Explicit message disposition: The HandleAsync method returns a TopicResponseAction (Success, Retry, or Drop), giving you precise control over acknowledgment, retries, and dead-letter queue routing.
• Scoped dependency injection: Handlers are resolved from an isolated IServiceScope for each message delivery, enabling clean constructor injection for scoped dependencies like Entity Framework DbContext, repositories, or loggers.
• Compile-time source generation: Annotated handlers are discovered by the Roslyn source generator at build time to emit reflection-free, Native AOT–compatible dispatchers.

Example: Topic handler with ITopicHandler<TMessage>

using Dapr.Messaging;
using Dapr.Messaging.PublishSubscribe;

public record Order(string Id, decimal Total, string CustomerEmail);

[DaprTopic("pubsub", "orders")]
public sealed class OrderHandler(ILogger<OrderHandler> logger, IInventoryService inventory) 
    : ITopicHandler<Order>
{
    public async Task<TopicResponseAction> HandleAsync(
        Order message, 
        TopicContext context, 
        CancellationToken cancellationToken)
    {
        logger.LogInformation("Processing order {OrderId} from pubsub {PubSub}", message.Id, context.PubsubName);

        try
        {
            await inventory.ReserveStockAsync(message.Id, cancellationToken);
            return TopicResponseAction.Success;
        }
        catch (InventoryUnavailableException ex)
        {
            logger.LogWarning(ex, "Insufficient inventory for order {OrderId}; dropping message", message.Id);
            return TopicResponseAction.Drop;
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "Transient error processing order {OrderId}; retrying", message.Id);
            return TopicResponseAction.Retry;
        }
    }
}

Response actions

The HandleAsync method returns a TopicResponseAction that tells the Dapr runtime how to acknowledge the message:

Accessing delivery context with TopicContext

The TopicContext passed to HandleAsync provides rich metadata about the delivery:

public async Task<TopicResponseAction> HandleAsync(Order message, TopicContext context, CancellationToken cancellationToken)
{
    var pubsubName = context.PubsubName;       // Name of the Dapr pub/sub component
    var topicName = context.TopicName;         // Name of the topic
    var messageId = context.MessageId;         // Unique message identifier
    var metadata = context.Metadata;           // Additional component-specific metadata
    var headers = context.Headers;             // CloudEvent headers forwarded with the delivery
    var rawBytes = context.RawPayload;         // Raw payload bytes (when EnableRawPayload is true)
    var cloudEvent = context.CloudEvent;       // Strongly-typed CloudEvent envelope
    
    return TopicResponseAction.Success;
}

Configure subscriptions with [DaprTopic]

The [DaprTopic] attribute specifies which topic to subscribe to and how messages are delivered.

Positional constructor arguments

The [DaprTopic] constructor requires two positional string arguments:

1. pubsubName (1st argument): The name of the Dapr pub/sub component (as defined in your component YAML’s metadata.name, e.g., "pubsub" or "messagebus").
2. topicName (2nd argument): The name of the topic or queue to subscribe to (e.g., "orders" or "orders.us").

// Subscribes to the "orders" topic on the "pubsub" component
[DaprTopic("pubsub", "orders")]
public class OrderHandler : ITopicHandler<Order> { /* ... */ }

Attribute definition and properties

[AttributeUsage(AttributeTargets.Class, AllowMultiple = true, Inherited = false)]
public sealed class DaprTopicAttribute(string pubsubName, string topicName) : Attribute
{
    public string PubsubName { get; }
    public string TopicName { get; }
    public DeliveryMode Delivery { get; set; } = DeliveryMode.Streaming;
    public string? Route { get; set; }
    public string? Match { get; set; }
    public int Priority { get; set; }
    public string? DeadLetterTopic { get; set; }
    public bool EnableRawPayload { get; set; }
    public bool BulkSubscribe { get; set; }
    public int MaxMessagesCount { get; set; } = 100;
    public int MaxAwaitDurationMs { get; set; } = 1000;
    public string[]? MetadataKeys { get; set; }
}

Property applicability by delivery mode

Not all properties on [DaprTopic] are relevant to every delivery mode. The table below outlines which properties apply to each mode:

The three delivery modes

DeliveryMode controls the communication protocol between the Dapr sidecar and your application:

1. DeliveryMode.Streaming (Default):

   • The application initiates a persistent bidirectional gRPC stream (SubscribeTopicEventsAlpha1) to the Dapr sidecar.
   • Messages are pulled by the application with client-side backpressure buffering.
   • No inbound HTTP or gRPC endpoints need to be mapped for streaming subscriptions.
   • Ignores HTTP routing properties like Route.

   [DaprTopic("pubsub", "orders", Delivery = DeliveryMode.Streaming)]
   public class StreamingOrderHandler : ITopicHandler<Order> { /* ... */ }
   

2. DeliveryMode.Programmatic:

   • The Dapr sidecar pushes messages to your application’s AppCallback gRPC service.
   • Subscriptions are discovered automatically at startup via ListTopicSubscriptions and invoked via OnTopicEvent.
   • Requires calling app.MapDaprMessaging() on the endpoint pipeline.
   • Supports CEL routing rules (Match, Priority) and BulkSubscribe. Ignores Route.

   [DaprTopic("pubsub", "orders", Delivery = DeliveryMode.Programmatic)]
   public class ProgrammaticOrderHandler : ITopicHandler<Order> { /* ... */ }
   

3. DeliveryMode.Http:

   • The Dapr sidecar discovers subscriptions by calling GET /dapr/subscribe and pushes events via POST <route>.
   • Requires calling app.MapDaprMessaging() on the endpoint pipeline.
   • Route defines the HTTP path where events are posted (defaults to the topic name if omitted).

   [DaprTopic("pubsub", "orders", Delivery = DeliveryMode.Http, Route = "/api/orders")]
   public class HttpOrderHandler : ITopicHandler<Order> { /* ... */ }
   

Content-based routing with CEL

You can filter and route messages based on Common Expression Language (CEL) expressions using the Match and Priority properties:

[DaprTopic("pubsub", "orders", Match = "event.type == \"priority\"", Priority = 1)]
public sealed class PriorityOrderHandler : ITopicHandler<Order> { /* ... */ }

[DaprTopic("pubsub", "orders", Priority = 2)]
public sealed class StandardOrderHandler : ITopicHandler<Order> { /* ... */ }

Dead-letter topics and bulk subscription

• Dead-letter topic: Set DeadLetterTopic = "orders-dlq" to route dropped messages to a dead-letter queue.
• Bulk subscription: Set BulkSubscribe = true, MaxMessagesCount, and MaxAwaitDurationMs to receive batches of messages.
• Raw payloads: Set EnableRawPayload = true to receive the unparsed event payload without CloudEvent envelope parsing.

Topic metadata with [DaprTopicMetadata]

You can attach key-value metadata to your topic subscription using the [DaprTopicMetadata] attribute:

[DaprTopic("pubsub", "orders", MetadataKeys = ["rawPayload"])]
[DaprTopicMetadata("rawPayload", "true")]
public sealed class RawOrderHandler : ITopicHandler<byte[]> { /* ... */ }

Handling multiple topics on a single handler

[DaprTopic] allows multiple attributes on a single handler class. The handler processes messages from each configured topic:

[DaprTopic("pubsub", "orders.us")]
[DaprTopic("pubsub", "orders.eu")]
[DaprTopic("pubsub", "orders.asia")]
public sealed class MultiRegionOrderHandler : ITopicHandler<Order>
{
    public Task<TopicResponseAction> HandleAsync(Order message, TopicContext context, CancellationToken cancellationToken)
    {
        Console.WriteLine($"Received order from topic: {context.TopicName}");
        return Task.FromResult(TopicResponseAction.Success);
    }
}

What the source generator produces

When you build your project, the Dapr.Messaging.Generators Roslyn source generator discovers all [DaprTopic] classes implementing ITopicHandler<TMessage> and emits:

Handler discovery and dispatch registration are generated at compile time. Message deserialization currently uses runtime System.Text.Json metadata, so Native AOT and trimming scenarios require additional validation and an application-owned JsonSerializerContext.

Register at startup

Register Dapr messaging services in Program.cs using builder.Services.AddDaprMessaging():

var builder = WebApplication.CreateBuilder(args);

// Register the complete Dapr messaging stack (options, publisher client, and source-generated subscribers)
builder.Services.AddDaprMessaging(options =>
{
    // Optional: configure Dapr sidecar connection options
    options.DaprGrpcEndpoint = "http://localhost:50001";
});

var app = builder.Build();

// Automatically map the endpoints required by the discovered delivery modes
app.MapDaprMessaging();

app.Run();

Dynamic streaming subscriptions

In addition to declarative [DaprTopic] handlers, DaprPublishSubscribeClient allows creating imperative, dynamic streaming subscriptions at runtime:

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

// Configure subscription options
var options = new DaprSubscriptionOptions(
    new MessageHandlingPolicy(
        TimeoutDuration: TimeSpan.FromSeconds(10), 
        DefaultResponseAction: TopicResponseAction.Retry))
{
    DeadLetterTopic = "orders-dlq",
    MaximumQueuedMessages = 500,
    MaximumCleanupTimeout = TimeSpan.FromSeconds(15),
    ErrorHandler = async (DaprException ex) =>
    {
        Console.WriteLine($"Subscription error: {ex.Message}");
    }
};

// Start the dynamic subscription
await using var subscription = await messagingClient.SubscribeAsync(
    "pubsub", 
    "dynamic-orders", 
    options, 
    async (TopicMessage message, CancellationToken ct) =>
    {
        var payload = Encoding.UTF8.GetString(message.Data.Span);
        Console.WriteLine($"Received dynamic message: {payload}");
        return TopicResponseAction.Success;
    });

// SubscribeAsync is typed as IAsyncDisposable; the runtime handle also exposes completion.
await ((IDaprSubscription)subscription).Completion;

Analyzers and diagnostics

Dapr.Messaging ships with Roslyn analyzers that enforce correct messaging practices at build time:

For a complete reference of all Roslyn analyzers, diagnostic severities, and available automated code fixes across the Dapr .NET SDK, see Dapr source code analyzers and generators.

Best practices

• Prefer DeliveryMode.Streaming for background workers, console apps, and services where you want bidirectional streaming and client-side backpressure without hosting an inbound HTTP or gRPC server.
• Prefer DeliveryMode.Programmatic for high-throughput gRPC services where the Dapr sidecar pushes events directly into ASP.NET Core gRPC endpoints.
• Always handle transient vs permanent errors: Return TopicResponseAction.Retry for transient errors (network timeouts, database locks) and TopicResponseAction.Drop for unrecoverable errors (poison messages, schema violations) to route them to a dead-letter topic.
• Inject scoped services: ITopicHandler<TMessage> instances are resolved per message within an isolated IServiceScope, ensuring safe resolution of scoped dependencies like Entity Framework DbContext.
• Validate Native AOT explicitly: Generated dispatchers currently use runtime System.Text.Json metadata; test Native AOT and trimming with your target SDK version and message types.

Next steps

• How-To: Publish events with IDaprPublishSubscribeClient
• Tutorial: Dapr.Messaging by example
• Dapr Messaging configuration and usage guide
• Dapr Pub/Sub component specification

3 - Dapr Messaging configuration and usage

This guide covers advanced configuration, service lifetimes, delivery mode trade-offs, serialization, Native AOT support, and resilience patterns when working with the Dapr.Messaging SDK.

Lifetime management and thread safety

Understanding the lifetime of messaging services ensures optimal performance and prevents socket exhaustion or memory leaks.

Client lifetime (IDaprPublishSubscribeClient)

IDaprPublishSubscribeClient (and its underlying implementation DaprPublishSubscribeClient) manages persistent TCP and gRPC channel resources to communicate with the Dapr sidecar.

• Singleton registration: The client is thread-safe and designed to be registered as a Singleton in your dependency injection container.
• Connection pooling: Reuses underlying HttpClient and GrpcChannel instances across requests.
• Do not create per operation: Avoid instantiating or disposing DaprPublishSubscribeClient per operation. A single shared client should serve your entire application.

// Registers the Dapr Messaging stack, including IDaprPublishSubscribeClient and DaprPublishSubscribeClient as a Singleton
builder.Services.AddDaprMessaging();

Handler lifetime (ITopicHandler<TMessage>)

When messages are delivered to an ITopicHandler<TMessage>:

• Scoped execution: The source-generated dispatcher resolves the topic handler from an isolated IServiceScope on every message delivery.
• Constructor injection: Handlers can safely inject transient or scoped dependencies (such as Entity Framework Core DbContext instances, repositories, or tenant context providers).
• Automatic disposal: Any disposable dependencies resolved by the handler are automatically disposed when the message handling turn completes.

public sealed class OrderHandler(ApplicationDbContext dbContext, ILogger<OrderHandler> logger) 
    : ITopicHandler<Order>
{
    public async Task<TopicResponseAction> HandleAsync(Order message, TopicContext context, CancellationToken ct)
    {
        dbContext.Orders.Add(message);
        await dbContext.SaveChangesAsync(ct);
        return TopicResponseAction.Success;
    }
}

Configuring options via DaprMessagingOptions

The SDK uses the standard Microsoft.Extensions.Options pattern. You can configure DaprMessagingOptions in code or bind it directly from IConfiguration.

builder.Services.AddDaprMessaging(options =>
{
    options.DaprGrpcEndpoint = "http://localhost:50001";
    options.DaprApiToken = "my-secret-token";
    options.StreamingReconnectDelay = TimeSpan.FromSeconds(5);
    options.JsonSerializerOptions = new JsonSerializerOptions(JsonSerializerDefaults.Web)
    {
        PropertyNamingPolicy = JsonNamingPolicy.CamelCase
    };
});

Supported configuration options

Environment variable fallbacks

If not explicitly configured in DaprMessagingOptions, the SDK automatically reads the standard Dapr environment variables:

• DAPR_GRPC_ENDPOINT: Used as the gRPC endpoint.
• DAPR_GRPC_PORT: Used to construct http://127.0.0.1:<port> when DAPR_GRPC_ENDPOINT is not set.
• DAPR_HTTP_ENDPOINT: Used for HTTP endpoints.
• DAPR_HTTP_PORT: Used to construct http://127.0.0.1:<port> for HTTP when DAPR_HTTP_ENDPOINT is not set.
• DAPR_API_TOKEN: Used as the Dapr API authentication token.

Choosing the right delivery mode

Dapr.Messaging unifies three delivery modes under a single [DaprTopic] attribute. Choose the mode that fits your architectural requirements:

Advanced messaging patterns

Bulk publish and subscribe

Bulk operations allow publishing and consuming multiple events in batches, reducing network round-trips to the sidecar:

Bulk publishing

var orders = new List<Order>
{
    new("101", 49.99m, "user1@example.com"),
    new("102", 99.50m, "user2@example.com")
};

var response = await pubsubClient.BulkPublishEventAsync("pubsub", "orders", orders);

foreach (var failed in response.FailedEntries)
{
    logger.LogError("Failed to publish order {OrderId}: {Error}", failed.Entry.EventData.Id, failed.ErrorMessage);
}

Bulk subscribing

Configure bulk consumption by enabling BulkSubscribe = true on the [DaprTopic] attribute:

[DaprTopic("pubsub", "orders", BulkSubscribe = true, MaxMessagesCount = 50, MaxAwaitDurationMs = 500)]
public sealed class BulkOrderHandler : ITopicHandler<Order>
{
    public async Task<TopicResponseAction> HandleAsync(
        Order message,
        TopicContext context, 
        CancellationToken cancellationToken)
    {
        logger.LogInformation("Processing order {OrderId}", message.Id);
        // Process one message from the runtime-delivered batch.
        return TopicResponseAction.Success;
    }
}

Bulk delivery batches messages at the Dapr protocol level, but the current SDK dispatches each entry to the handler separately. The handler therefore implements ITopicHandler<Order>, not ITopicHandler<IReadOnlyList<Order>>.

Dead-letter topics

Configure a dead-letter topic to capture unprocessable or poison messages:

[DaprTopic("pubsub", "orders", DeadLetterTopic = "orders-poison")]
public sealed class ResilientOrderHandler : ITopicHandler<Order>
{
    public Task<TopicResponseAction> HandleAsync(Order message, TopicContext context, CancellationToken cancellationToken)
    {
        if (string.IsNullOrEmpty(message.CustomerEmail))
        {
            // Dropping routes the message to 'orders-poison' because DeadLetterTopic is configured
            return Task.FromResult(TopicResponseAction.Drop);
        }

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

Working with CloudEvents

When messages adhere to the CloudEvents specification, you can publish or consume them with full fidelity:

Publishing typed CloudEvents

var cloudEvent = new CloudEvent<Order>(order)
{
    Source = new Uri("urn:service:checkout"),
    Type = "com.myapp.order.created",
    Subject = $"orders/{order.Id}",
    Time = DateTimeOffset.UtcNow
};

await pubsubClient.PublishEventAsync("pubsub", "orders", cloudEvent);

Consuming CloudEvent headers in handlers

public Task<TopicResponseAction> HandleAsync(Order message, TopicContext context, CancellationToken cancellationToken)
{
    if (context.CloudEvent is not null)
    {
        var correlationId = context.CloudEvent.TraceId;
        var eventSource = context.CloudEvent.Source;
        logger.LogInformation("Processing CloudEvent {Id} from {Source}", context.CloudEvent.Id, eventSource);
    }

    return Task.FromResult(TopicResponseAction.Success);
}

Native AOT and trimming considerations

The Dapr.Messaging.Generators source generator produces dispatch logic and subscriber registries at build time, avoiding reflection for handler discovery and registration. Generated subscriber dispatchers currently deserialize messages with runtime System.Text.Json metadata. Native AOT and trimming scenarios therefore require explicit validation with the target SDK version; configuring DaprMessagingOptions.JsonSerializerOptions does not currently replace the generated subscriber deserializer.

Resiliency and error handling

Backpressure management in streaming mode

When using DeliveryMode.Streaming or dynamic streaming subscriptions, the Dapr Messaging SDK maintains an internal backpressure queue. Messages remain held in the Dapr sidecar runtime until your handler is ready to process them.

You can tune the queue size and cleanup timeout in DaprSubscriptionOptions:

var options = new DaprSubscriptionOptions(
    new MessageHandlingPolicy(
        TimeoutDuration: TimeSpan.FromSeconds(15), 
        DefaultResponseAction: TopicResponseAction.Retry))
{
    MaximumQueuedMessages = 1000,                    // Max backlog held in-memory
    MaximumCleanupTimeout = TimeSpan.FromSeconds(30), // Max wait to flush ACKs on shutdown
    ErrorHandler = async (exception) =>
    {
        // Invoked on sidecar streaming or connection faults
        logger.LogError(exception, "Streaming subscription encountered an error");
    }
};

Graceful shutdown

When your application shuts down, streaming subscriptions stop pulling new messages and flush pending acknowledgments to the Dapr sidecar within the configured MaximumCleanupTimeout, ensuring zero message loss during deployments or scale-downs.

Next steps

• How-To: Publish events with IDaprPublishSubscribeClient
• How-To: Author subscriptions and handle topic messages
• Tutorial: Dapr.Messaging by example
• Dapr Pub/Sub overview
• Dapr components specification

4 - Tutorial: Dapr.Messaging by example

This tutorial uses the runnable examples in the .NET SDK repository to demonstrate the main capabilities of Dapr.Messaging. Each part includes the implementation pattern and the corresponding unit and integration testing approach.

The examples use .NET 8, .NET 9, or .NET 10 and require Dapr runtime 1.18 or later. The unit tests run without Dapr, Docker, or a message broker. The integration tests use Dapr.Testcontainers to run a real Dapr sidecar and Redis pub/sub component, which is the recommended way to verify delivery behavior in an application.

Tutorial content

• Part 1: Publish events: Typed events, CloudEvent metadata, bulk publishing, and raw payloads.
• Part 2: Declarative streaming subscriptions: Source-generated handlers with explicit success, retry, and drop acknowledgements.
• Part 3: Content-based routing and dead-letter topics: CEL matching rules, priorities, metadata, and dead-letter handling.
• Part 4: Bulk subscriptions: High-throughput delivery with bounded batch size and wait duration.
• Part 5: gRPC AppCallback push: Sidecar-initiated gRPC delivery for applications that expose a gRPC port.
• Part 6: HTTP push subscriptions: Generated /dapr/subscribe discovery and custom HTTP routes.
• Part 7: Dynamic streaming subscriptions: Imperative runtime subscriptions with supervision, cancellation, and reconnect behavior.

Testing pattern

Keep message-handling decisions in ordinary C# handlers so they can be tested directly. The example unit tests construct a TopicContext, invoke HandleAsync, and assert the returned TopicResponseAction for valid, invalid, and transient inputs.

Add integration tests for the infrastructure boundary. The examples create a PubSubHarness, configure the application with the harness’s dynamically assigned Dapr endpoints, publish real events, and wait for observable processing results. Use bounded timeouts, IAsyncLifetime cleanup, and a test log sink rather than sleeps alone when validating asynchronous delivery.

Prerequisites

• .NET SDK 8, 9, or 10
• Dapr CLI and a Dapr environment
• Dapr runtime 1.18 or later for streaming subscriptions
• Docker or another container runtime for integration tests
• The Dapr.Messaging package

When using the SDK repository directly, the examples reference the local SDK projects. In an application, install the published Dapr.Messaging package instead.

4.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

4.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

4.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.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

4.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

4.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

4.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