HoneyDrunk.Transport.StorageQueue 0.3.0

HoneyDrunk.Transport

Reliable messaging and outbox infrastructure for the Hive - Transport unifies brokers, queues, and event buses under one contract ensuring delivery, order, and idempotence. It powers communication between Nodes—Data, Pulse, Vault, and beyond—so every message finds its way.

Signal Quote: "Every message finds its way."

📦 What Is This?

HoneyDrunk.Transport is the messaging backbone of HoneyDrunk.OS ("the Hive"). It provides a transport-agnostic abstraction layer over different message brokers with built-in resilience, observability, and exactly-once semantics:

• ✅ Transport Abstraction - Unified ITransportPublisher and ITransportConsumer over Azure Service Bus, Azure Storage Queue, and InMemory
• ✅ Middleware Pipeline - Onion-style processing with logging, telemetry, correlation, and retry
• ✅ Envelope Pattern - Immutable ITransportEnvelope with correlation/causation tracking
• ✅ Transactional Outbox - Exactly-once processing with database transactions
• ✅ Kernel Integration - Uses TimeProvider and IGridContext from HoneyDrunk.Kernel for deterministic timestamps and distributed context
• ✅ Observability - OpenTelemetry spans and pluggable ITransportMetrics
• ✅ Blob Fallback for Service Bus - Persist failed publishes to Azure Blob Storage for later replay

🚀 Quick Start

Installation

<ItemGroup>
  <PackageReference Include="HoneyDrunk.Transport" Version="0.1.0" />
  <PackageReference Include="HoneyDrunk.Transport.AzureServiceBus" Version="0.1.0" />
  <PackageReference Include="HoneyDrunk.Transport.StorageQueue" Version="0.1.0" />
  <PackageReference Include="HoneyDrunk.Transport.InMemory" Version="0.1.0" />
</ItemGroup>

Configure in Program.cs

using HoneyDrunk.Kernel.DependencyInjection;
using HoneyDrunk.Transport.DependencyInjection;

var builder = WebApplication.CreateBuilder(args);

// 1. Register Kernel node
builder.Services.AddHoneyDrunkCoreNode(nodeDescriptor);

// 2. Register Transport core
builder.Services.AddHoneyDrunkTransportCore(options =>
{
    options.EnableTelemetry = true;
    options.EnableLogging = true;
    options.EnableCorrelation = true;
});

// 3. Choose a transport

// Azure Service Bus
builder.Services.AddHoneyDrunkServiceBusTransport(options =>
{
    options.FullyQualifiedNamespace = "mynamespace.servicebus.windows.net";
    options.Address = "orders";
    options.EntityType = ServiceBusEntityType.Topic;
    options.SubscriptionName = "order-processor";
    options.MaxConcurrency = 10;
    options.PrefetchCount = 20;

    options.ServiceBusRetry.Mode = ServiceBusRetryMode.Exponential;
    options.ServiceBusRetry.MaxRetries = 3;
});

// OR Azure Storage Queue
builder.Services
    .AddHoneyDrunkTransportStorageQueue(
        builder.Configuration["StorageQueue:ConnectionString"]!,
        "orders")
    .WithMaxDequeueCount(5)
    .WithConcurrency(10);

// 4. Register message handlers
builder.Services.AddMessageHandler<OrderCreatedEvent, OrderCreatedHandler>();

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

📖 Usage Examples

Publishing Messages

public class OrderService(
    ITransportPublisher publisher,
    EnvelopeFactory envelopeFactory,
    IMessageSerializer serializer,
    IGridContext gridContext)
{
    public async Task CreateOrderAsync(CreateOrderCommand command, CancellationToken ct)
    {
        // Create order...
        
        // Publish event
        var @event = new OrderCreatedEvent { OrderId = orderId, Total = total };
        var payload = serializer.Serialize(@event);
        var envelope = envelopeFactory.CreateEnvelopeWithGridContext<OrderCreatedEvent>(
            payload, gridContext);
        
        await publisher.PublishAsync(
            envelope,
            EndpointAddress.Create("orders", "orders-topic"),
            ct);
    }
}

Handling Messages

public class OrderCreatedHandler : IMessageHandler<OrderCreatedEvent>
{
    private readonly ILogger<OrderCreatedHandler> _logger;
    
    public OrderCreatedHandler(ILogger<OrderCreatedHandler> logger)
    {
        _logger = logger;
    }
    
    public async Task<MessageProcessingResult> HandleAsync(
        OrderCreatedEvent message,
        MessageContext context,
        CancellationToken cancellationToken)
    {
        var grid = context.GridContext;
        
        _logger.LogInformation(
            "Processing order {OrderId} with CorrelationId {CorrelationId} on Node {NodeId}",
            message.OrderId,
            grid?.CorrelationId,
            grid?.NodeId);
        
        await SendConfirmationEmailAsync(message.OrderId, cancellationToken);
        return MessageProcessingResult.Success;
    }
}

Transactional Outbox

public class OrderService(
    IOutboxStore outboxStore,
    EnvelopeFactory factory,
    IMessageSerializer serializer,
    IDbContext dbContext)
{
    public async Task CreateOrderAsync(CreateOrderCommand command, CancellationToken ct)
    {
        await using var transaction = await dbContext.BeginTransactionAsync(ct);
        
        try
        {
            // Save order to database
            var order = new Order { /* ... */ };
            await dbContext.Orders.AddAsync(order, ct);
            
            // Save message to outbox (same transaction)
            var payload = serializer.Serialize(new OrderCreatedEvent { OrderId = order.Id });
            var envelope = factory.CreateEnvelope<OrderCreatedEvent>(payload);
            var destination = EndpointAddress.Create("orders", "orders-topic");
            
            await outboxStore.SaveAsync(destination, envelope, ct);
            
            await dbContext.SaveChangesAsync(ct);
            await transaction.CommitAsync(ct);
            
            // DefaultOutboxDispatcher publishes from outbox in background
        }
        catch
        {
            await transaction.RollbackAsync(ct);
            throw;
        }
    }
}

🎯 Features

🔍 Core Components

🔗 Kernel Integration

HoneyDrunk.Transport extends HoneyDrunk.Kernel with messaging primitives:

🚀 Available Transports

🧪 Testing

Use InMemory transport and DI for tests:

var services = new ServiceCollection();
services.AddHoneyDrunkCoreNode(TestNodeDescriptor);
services.AddHoneyDrunkTransportCore()
    .AddHoneyDrunkInMemoryTransport();

services.AddMessageHandler<OrderCreatedEvent>((msg, ctx, ct) =>
{
    // Assert in handler
    return Task.FromResult(MessageProcessingResult.Success);
});

await using var provider = services.BuildServiceProvider();

var broker = provider.GetRequiredService<InMemoryBroker>();
var publisher = provider.GetRequiredService<ITransportPublisher>();
var pipeline = provider.GetRequiredService<IMessagePipeline>();

// Use broker for broker-level tests, pipeline for pipeline-level tests

See Testing.md for complete patterns including unit tests, integration tests, and test helpers.

📚 Documentation

🛠️ Repository Layout

HoneyDrunk.Transport/
├── HoneyDrunk.Transport/                    # Core abstractions & pipeline
│   ├── Abstractions/                        # Contracts & interfaces
│   ├── Pipeline/                            # Middleware execution engine
│   ├── Configuration/                       # Options & settings
│   ├── Context/                             # Grid context integration
│   ├── Primitives/                          # Envelope & factory
│   ├── Outbox/                              # Transactional outbox
│   ├── Runtime/                             # ITransportRuntime host
│   ├── Health/                              # Health contributors
│   ├── Metrics/                             # ITransportMetrics
│   ├── Telemetry/                           # OpenTelemetry integration
│   └── DependencyInjection/                 # DI registration
├── HoneyDrunk.Transport.AzureServiceBus/    # Azure Service Bus provider
├── HoneyDrunk.Transport.StorageQueue/       # Azure Storage Queue provider
├── HoneyDrunk.Transport.InMemory/           # In-memory provider
├── HoneyDrunk.Transport.Tests/              # Test project
└── docs/                                    # Documentation

⚖️ Storage Queue vs Service Bus

Choose Storage Queue for cost-effective, high-volume, simple queue scenarios.
Choose Service Bus for enterprise messaging with topics, sessions, or transactions.

📄 License

MIT