DotPulsar.Client.Api 0.1.17-preview

DotPulsar.Client.Api

Standalone prototype of the old Apache Pulsar service-facing library style, backed by the official DotPulsar client.

This project intentionally ignores dashboard/metrics for now. The goal is a simple library that services can use directly in the familiar way:

• configure Pulsar connection values
• create tenants, namespaces, and topics through Pulsar admin APIs
• publish messages
• subscribe with raw async handlers
• subscribe with batch semaphore async handlers

Timeouts

Timeouts are applied only to main I/O boundaries:

• producer create: ProducerConfig.CreateProducerTimeout
• producer send: ProducerConfig.SendTimeout
• consumer create: ConsumerConfig.CreateConsumerTimeout
• receive/move-next: ConsumerConfig.ReceiveOperationTimeout
• acknowledge: ConsumerConfig.AckTimeout
• negative acknowledge/redelivery: ConsumerConfig.NegativeAckTimeout
• dispose: ConsumerConfig.DisposeTimeout / ProducerConfig.DisposeTimeout
• business processing: ConsumerConfig.MessageProcessingTimeout
• batch task join: ConsumerConfig.BatchProcessingTimeout

Internal helper methods are not wrapped one by one.

Registration

using ApachePulsar.Business.Bootstrapper;
using ApachePulsar.Common.Static;

PulsarConfigurations.PulsarConnectionString = "pulsar://localhost:6650";
PulsarConfigurations.PulsarAdminApi = "http://localhost:8080/admin/v2";
PulsarConfigurations.Token = null;

services.AddDotPulsarLibrary();

Resource Creation

Tenant, namespace, and topic creation are still done through admin APIs, not through DotPulsar:

await _tenantService.CreateTenant(new TenantConfig
{
    Tenant = "public",
    AllowedClusters = "standalone"
});

await _namespaceService.CreateNamespace(new NamespaceConfig
{
    Tenant = "public",
    Namespace = "default"
});

await _topicService.CreateTopic(new TopicConfig
{
    Tenant = "public",
    Namespace = "default",
    TopicName = "my-topic",
    Partitions = 0
});

Handler Shape

public sealed class MyHandler : BatchSemaphorePulsarHandlerAsync<MyMessage>
{
    protected override ConsumerConfig HandlerConfig => new()
    {
        TopicConfig = new TopicConfig
        {
            Tenant = "public",
            Namespace = "default",
            TopicName = "my-topic"
        },
        SubscriptionName = "my-sub"
    };

    protected override Task ConsumeAsync(MyMessage? obj, CancellationToken cancellationToken)
    {
        return Task.CompletedTask;
    }
}

Current Scope

Implemented:

• DotPulsar client connection
• token auth
• producer publish
• raw consumer handler
• batch semaphore consumer handler
• admin API resource services
• boundary timeouts

Not implemented yet:

• metrics/dashboard
• duplicate tracking
• heartbeat UI
• full config parity with the old package