NetEvolve.Pulse.Dapr 0.84.106

NetEvolve.Pulse.Dapr

Dapr pub/sub transport for the Pulse outbox pattern. Publishes outbox messages to Dapr topics via DaprClient, enabling reliable event delivery to any message broker supported by Dapr—Redis, Kafka, Azure Service Bus, RabbitMQ, and more—without changing application code.

Features

• Dapr pub/sub: Publish outbox messages to any Dapr-supported message broker
• CloudEvents: The raw JSON payload is forwarded as CloudEvent data via DaprClient.PublishByteEventAsync, with a stable CloudEvent id and type derived from the outbox message
• Health checks: Delegates to DaprClient.CheckHealthAsync for readiness probing
• Configurable topic resolution: Map event types to topic names by registering a custom ITopicNameResolver
• Broker-agnostic: Switch brokers by changing the Dapr component configuration — no code changes required

Installation

NuGet Package Manager

Install-Package NetEvolve.Pulse.Dapr

.NET CLI

dotnet add package NetEvolve.Pulse.Dapr

PackageReference

<PackageReference Include="NetEvolve.Pulse.Dapr" Version="x.x.x" />

Quick Start

1. Add the Dapr client package

dotnet add package Dapr.AspNetCore

2. Register services

using Microsoft.Extensions.DependencyInjection;
using NetEvolve.Pulse;

var services = new ServiceCollection();

// DaprClient must be registered before UseDaprTransport
services.AddDaprClient();

services.AddPulse(config => config
    .AddOutbox(
        options => options.Schema = "pulse",
        processorOptions => processorOptions.BatchSize = 100)
    .UseDaprTransport(options =>
    {
        options.PubSubName = "pubsub"; // Dapr pub/sub component name
    }));

3. Store events via IEventOutbox

Use IEventOutbox to store events reliably. The outbox processor picks them up and publishes each one to the configured Dapr topic:

public class OrderService
{
    private readonly IEventOutbox _outbox;

    public OrderService(IEventOutbox outbox) => _outbox = outbox;

    public async Task CreateOrderAsync(CreateOrderRequest request, CancellationToken ct)
    {
        // ... business logic ...

        // Stored reliably; published via Dapr when the processor runs
        await _outbox.StoreAsync(new OrderCreatedEvent
        {
            OrderId = Guid.NewGuid(),
            CustomerId = request.CustomerId
        }, ct);
    }
}

Transaction Integration

For reliable at-least-once delivery guarantees, store outbox events within the same database transaction as your business data. Pair the Dapr transport with a persistence provider that supports transaction enlistment:

public class OrderService
{
    private readonly ApplicationDbContext _context;
    private readonly IEventOutbox _outbox;

    public OrderService(ApplicationDbContext context, IEventOutbox outbox)
    {
        _context = context;
        _outbox = outbox;
    }

    public async Task CreateOrderAsync(CreateOrderRequest request, CancellationToken ct)
    {
        // Begin transaction
        await using var transaction = await _context.Database.BeginTransactionAsync(ct);

        try
        {
            // Business operation
            var order = new Order { CustomerId = request.CustomerId, Total = request.Total };
            _context.Orders.Add(order);
            await _context.SaveChangesAsync(ct);

            // Store event in outbox (same transaction)
            await _outbox.StoreAsync(new OrderCreatedEvent
            {
                OrderId = order.Id,
                CustomerId = order.CustomerId
            }, ct);

            // Commit both business data and event atomically
            await transaction.CommitAsync(ct);
        }
        catch
        {
            // Rollback discards both business data AND the outbox event
            await transaction.RollbackAsync(ct);
            throw;
        }
    }
}

Subscriber Integration

Dapr subscribers receive the published events as CloudEvents. The transport overrides these CloudEvent attributes through publish metadata:

Use the id to deduplicate redelivered events in idempotent subscribers. Because the type is the assembly-qualified name, it includes the assembly version. Match on it with care in subscription routing rules. Trace attributes (traceid, traceparent, tracestate) are left to the Dapr sidecar.

Use Dapr.AspNetCore to subscribe to topics in ASP.NET Core:

// Program.cs — enable Dapr subscriber routing
app.MapSubscribeHandler();

// OrdersController.cs
using Dapr;
using Microsoft.AspNetCore.Mvc;

[ApiController]
[Route("[controller]")]
public class OrdersController : ControllerBase
{
    [Topic("pubsub", "OrderCreated")]
    [HttpPost("order-created")]
    public async Task<IActionResult> OnOrderCreated(
        [FromBody] OrderCreatedEvent evt,
        CancellationToken ct)
    {
        // Handle the event
        return Ok();
    }
}

Or use a declarative subscription component:

apiVersion: dapr.io/v1alpha1
kind: Subscription
metadata:
  name: order-created-subscription
spec:
  pubsubname: pubsub
  topic: OrderCreated
  route: /orders/order-created

Configuration

DaprMessageTransportOptions

Topic Name Resolution

Topic names are resolved by the registered ITopicNameResolver. By default, the Type.Name of OutboxMessage.EventType (a System.Type) is used as the topic name.

Override the resolver for custom naming strategies:

public class FullNameTopicNameResolver : ITopicNameResolver
{
    // Use the namespace-qualified type name as topic
    public string Resolve(OutboxMessage message) => message.EventType.FullName ?? message.EventType.Name;
}

services.AddSingleton<ITopicNameResolver, FullNameTopicNameResolver>();
services.AddPulse(config => config
    .AddOutbox()
    .UseDaprTransport(options => options.PubSubName = "servicebus-pubsub"));

Dapr Component Examples

Redis Streams (local development)

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

Azure Service Bus

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: pubsub
spec:
  type: pubsub.azure.servicebus.topics
  version: v1
  metadata:
    - name: connectionString
      value: "Endpoint=sb://..."

RabbitMQ

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: pubsub
spec:
  type: pubsub.rabbitmq
  version: v1
  metadata:
    - name: host
      value: "amqp://guest:guest@localhost:5672"

Broker Switching

Because all broker configuration lives in the Dapr component YAML, switching brokers requires no code changes:

// This registration never changes, regardless of the broker
services.AddPulse(config => config
    .AddOutbox()
    .UseDaprTransport(options => options.PubSubName = "pubsub"));

To switch from Redis to Azure Service Bus, update the component YAML and redeploy — application code stays identical.

How It Works

1. Your application stores events in the outbox via IEventOutbox.StoreAsync within a database transaction.
2. The Pulse background processor polls the outbox for pending messages.
3. For each message, DaprMessageTransport publishes the stored JSON payload bytes unchanged via DaprClient.PublishByteEventAsync (content type application/json) with the resolved pub/sub component name and topic.
4. Dapr delivers the CloudEvent to subscribers on the configured broker.
5. On success, the message is marked as processed; on failure, it remains pending for the next poll cycle.

Performance Considerations

Batch Processing

Configure batch size and polling interval based on your throughput requirements:

.AddOutbox(configureProcessorOptions: options =>
{
    options.BatchSize = 100;                         // Messages per poll cycle
    options.PollingInterval = TimeSpan.FromSeconds(1);
})

Singleton Transport

DaprMessageTransport is registered as a singleton. DaprClient is thread-safe and designed for concurrent use, so no additional synchronization is required.

Requirements

• .NET 8.0, .NET 9.0, or .NET 10.0
• Dapr runtime 1.13+
• Dapr.AspNetCore (or Dapr.Client) with a registered DaprClient in the DI container
• Microsoft.Extensions.DependencyInjection for service registration
• Microsoft.Extensions.Hosting for the background processor

Related Packages

• NetEvolve.Pulse - Core mediator and outbox abstractions
• NetEvolve.Pulse.Extensibility - Core contracts and abstractions
• NetEvolve.Pulse.EntityFramework - Entity Framework Core persistence provider
• NetEvolve.Pulse.SqlServer - SQL Server ADO.NET persistence provider
• NetEvolve.Pulse.Polly - Polly v8 resilience policies integration

Documentation

For complete documentation, please visit the official documentation.

Contributing

Contributions are welcome! Please read the Contributing Guidelines before submitting a pull request.

Support

• Issues: Report bugs or request features on GitHub Issues
• Documentation: Read the full documentation at https://github.com/dailydevops/pulse

License

This project is licensed under the MIT License - see the LICENSE file for details.