Skip to main content

Apache Pulsar Transport

Excalibur.Dispatch.Transport.Pulsar registers the transport primitives for Apache Pulsar: a keyed ITransportSender and ITransportReceiver that send, receive, and acknowledge messages against a Pulsar broker over the DotPulsar client.

Scope of this package

This package provides the keyed transport primitives (ITransportSender/ITransportReceiver for direct byte-level send/receive) plus a keyed IMessageBus publisher — so you can publish typed dispatch messages (actions, events, documents) to Pulsar through the pipeline's serialization. Full pipeline-integrated consumption (a receive-side adapter that dispatches inbound Pulsar messages end-to-end through middleware and routing) is not part of this package. Request/reply is not natively supported by Pulsar.

If you need full pipeline-integrated send and receive today, use one of the pipeline-integrated transports (Kafka, RabbitMQ, Azure Service Bus, AWS SQS, Google Pub/Sub).

Before You Start​

  • .NET 10.0
  • A running Apache Pulsar broker (local or hosted)
  • Familiarity with transports

Installation​

dotnet add package Excalibur.Dispatch.Transport.Pulsar

Dependencies: Excalibur.Dispatch.Abstractions, Excalibur.Dispatch.Transport.Abstractions, DotPulsar

Registration​

Register a named transport with AddPulsarTransport. The name is used for multi-transport routing and to resolve the keyed sender/receiver.

using Microsoft.Extensions.DependencyInjection;

services.AddPulsarTransport("events", pulsar =>
{
    pulsar.ServiceUrl("pulsar://localhost:6650")
          .Topic("orders")
          .SubscriptionName("order-processors")
          .SubscriptionType(PulsarSubscriptionType.Shared);
});

To register with the default name (pulsar), omit the name argument:

services.AddPulsarTransport(pulsar =>
{
    pulsar.ServiceUrl("pulsar://localhost:6650")
          .Topic("orders")
          .SubscriptionName("order-processors");
});

Builder options​

MethodPurpose
ServiceUrl(string)The Pulsar broker service URL (e.g. pulsar://localhost:6650).
Topic(string)The topic to produce to and consume from.
SubscriptionName(string)The consumer subscription name.
SubscriptionType(PulsarSubscriptionType)The subscription mode (see below). Defaults to Shared.

Subscription types​

PulsarSubscriptionType maps to Pulsar's native subscription modes:

ValueSemantics
Shared (default)Competing consumers — messages are delivered round-robin across all consumers on the subscription. Matches consumer-group transport semantics.
ExclusiveOnly one consumer may attach to the subscription.
FailoverOne active consumer with standby consumers that take over on failure.
KeySharedMessages with the same key are delivered to the same consumer.

Options are validated at startup (ValidateOnStart), so a missing service URL or topic fails fast with OptionsValidationException.

Using the sender and receiver​

The sender and receiver are registered as keyed singletons under the transport name. Resolve them with GetRequiredKeyedService:

using Excalibur.Dispatch.Transport;

var sender = provider.GetRequiredKeyedService<ITransportSender>("events");
var receiver = provider.GetRequiredKeyedService<ITransportReceiver>("events");

// Send
await sender.SendAsync(
    new TransportMessage
    {
        Body = JsonSerializer.SerializeToUtf8Bytes(order),
        ContentType = "application/json",
        MessageType = "OrderPlaced",
    },
    cancellationToken);

// Receive and acknowledge
var messages = await receiver.ReceiveAsync(maxMessages: 10, cancellationToken);
foreach (var message in messages)
{
    Process(message);
    await receiver.AcknowledgeAsync(message, cancellationToken);
}

Publishing typed dispatch messages​

AddPulsarTransport also registers a keyed IMessageBus under the transport name, so you can publish typed dispatch messages (actions, events, documents) through the framework's serialization rather than hand-building TransportMessage bodies:

using Excalibur.Dispatch.Transport;

var bus = provider.GetRequiredKeyedService<IMessageBus>("events");

await bus.PublishAsync(new OrderPlaced(orderId), context, cancellationToken);

PublishAsync has overloads for IDispatchAction, IDispatchEvent, and IDispatchDocument. The message is serialized with the configured IEventSerializer and produced to the configured topic.

Payload-size guard​

The receiver enforces a payload-size limit measured before deserialization. It defaults to the standard transport-ingress bound and is tuned via PulsarOptions.Receive.MaxPayloadBytes — set it to null to opt out. A non-positive value is rejected at startup.

What's Next​