MemQueue 1.0.2

MemQueue

In-memory Kafka-like message queue and event bus for .NET 8. Zero external infrastructure. Process-local pub/sub with topic partitions, consumer groups, offset tracking, retention, and backpressure.

MemQueue brings the Kafka programming model (topics, partitions, consumer groups, offsets) into your .NET process — no broker, no disk, no network. Ideal for in-process event sourcing, work distribution, and domain events.

Quick Links

• Installation
• Minimal Example
• Features
• Configuration Reference
• Source Generator
• FAQ
• Comparison
• Architecture
• Running Tests & Benchmarks
• License

Installation

.NET CLI

dotnet add package MemQueue
dotnet add package MemQueue.SourceGenerators

PackageReference

<PackageReference Include="MemQueue" Version="1.0.2" />
<PackageReference Include="MemQueue.SourceGenerators" Version="1.0.2" />

Package Overview

Minimal Example

A complete produce/consume loop in under 50 lines:

using MemQueue.Abstractions;
using MemQueue.DependencyInjection;
using MemQueue.Models;
using Microsoft.Extensions.DependencyInjection;

// 1. Define a message
public record OrderCreated(Guid OrderId, string Product, int Quantity) : MessageBase;

// 2. Setup
var services = new ServiceCollection();
services.AddMemQueue(builder => builder
    .AddTopic("orders", topic => topic
        .WithPartitions(2)
        .WithBufferCapacity(1_000)));
await using var provider = services.BuildServiceProvider();
var bus = provider.GetRequiredService<IMessageBus>();

// 3. Produce
await bus.ProduceAsync((TopicId)"orders", new OrderCreated(Guid.NewGuid(), "Widget", 3));

// 4. Subscribe (consumer group)
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
await bus.SubscribeAsync<OrderCreated>(
    (TopicId)"orders",
    (ConsumerGroupId)"processors",
    async (msg, ctx, ct) =>
    {
        Console.WriteLine($"Processed: {msg.Product} x{msg.Quantity} [P{ctx.Partition} @O{ctx.Offset}]");
        await ValueTask.CompletedTask;
    },
    cts.Token);

Features

• Topics & Partitions — Kafka-style topic/partition model with configurable partition count
• Consumer Groups — Multiple consumers share load; partitions assigned via rebalance strategies
• Offset Management — Manual commit, auto-commit, seek (beginning/end/arbitrary)
• Retention Policies — Trim by message count, age, or both; background cleanup via PeriodicTimer
• Overflow Control — Overwrite oldest, block with backpressure, or reject (drop newest)
• Backpressure Monitoring — High/low watermark events, slow consumer detection, producer block count
• Dead Letter Queue — Capture failed message deliveries with error metadata
• Domain Event Bus — Lightweight fire-and-forget in-process events (at-most-once)
• Ordering Modes — None, per-partition, or per-key (key-hash routing via FNV-1a)
• Source Generator — [Subscribe] attribute generates IHostedService boilerplate automatically
• DI Integration — AddMemQueue() builder with fluent topic configuration
• Topic Statistics — Per-partition head/tail offsets, message count, buffer utilization

Configuration Reference

Topic Options

services.AddMemQueue(builder => builder
    .AddTopic("my-topic", topic => topic
        .WithPartitions(8)                           // partition count (default: 1)
        .WithBufferCapacity(10_000)                  // ring buffer size per partition (default: 1024)
        .WithRetention(RetentionPolicy.ByCount.Of(5000))
        .WithOverflowPolicy(OverflowPolicy.Block)    // OverwriteOldest | Block | DropNewest
        .WithBackpressure(bp => {
            bp.HighWatermarkPercent = 0.8;            // alert at 80% full
            bp.LowWatermarkPercent = 0.3;             // recover at 30%
            bp.SlowConsumerStrategy = SlowConsumerStrategy.Warn;
            bp.SlowConsumerLagThreshold = 500;
        })
        .WithOrdering(OrderingMode.PerKey)           // None | PerPartition | PerKey
    )
);

Global Options

services.AddMemQueue(builder => builder
    .SetDefaultOrdering(OrderingMode.PerPartition)
    .UseKeyHashPartitioner()    // or UseRoundRobinPartitioner() (default)
    .UseRangeRebalancer()       // or UseRoundRobinRebalancer()
);

Consumer Group Options

Overflow Policies

Retention Policies

Source Generator

The MemQueue.SourceGenerators package provides the [Subscribe] attribute for declarative subscription registration.

using MemQueue;
using MemQueue.Abstractions;
using MemQueue.Models;

[Subscribe("orders", GroupId = "order-processors", AutoCommit = true)]
public class OrderHandler : IMessageHandler<OrderCreated>
{
    public async ValueTask HandleAsync(OrderCreated message, MessageContext context, CancellationToken ct)
    {
        Console.WriteLine($"Processing order {message.OrderId}");
        await ValueTask.CompletedTask;
    }
}

Register all [Subscribe]-decorated handlers:

services.AddMemQueueSubscribers();

The source generator produces:

• AddMemQueueSubscribers() extension method
• IHostedService implementations per subscriber class
• Scoped DI resolution for each handler

FAQ

What delivery guarantees does MemQueue provide?

Two modes:

• IMessageBus — At-least-once delivery. Messages persist in the ring buffer until consumed and committed. Suitable for task queues, event sourcing, and work distribution.
• IDomainEventBus — At-most-once delivery. Fire-and-forget in-process events. If a handler throws, the event is lost. Suitable for loose coupling and notifications.

Must all messages inherit from MessageBase?

Yes. MessageBase is an abstract record that all messages must extend. This allows MemQueue to track metadata (timestamp, key) uniformly. Define messages as public record MyMessage(...) : MessageBase;.

How do subscriptions work with cancellation?

SubscribeAsync accepts a CancellationToken. When cancelled, the background consumption loop stops gracefully. For graceful shutdown in ASP.NET Core, use IHostApplicationLifetime.ApplicationStopping.

Should I use auto-commit or manual commit?

• Auto-commit (true) — Recommended for idempotent handlers. Offset is committed after the handler returns successfully.
• Manual commit (false) — Use when you need to commit after downstream side effects (e.g., database write). Call await ctx.CommitAsync(ct) when ready.

How do I choose an overflow policy?

• OverwriteOldest (default) — Best for metrics/logs where data freshness matters more than completeness.
• Block — Use when zero data loss is required. Producers wait until consumers free space.
• DropNewest — Use when you prefer to reject new work rather than lose old work.

Is MemQueue thread-safe?

Yes. All core components use appropriate synchronization: Interlocked for offset counters, SemaphoreSlim for blocking writes, ReaderWriterLockSlim for consumer group mutations, and ConcurrentDictionary for registries.

Can I replay messages?

Yes. Use bus.CreateConsumer<T>(topic) to create a manual consumer, then call SeekToBeginningAsync, SeekToEndAsync, or SeekAsync(partition, offset) to position the cursor.

Comparison

MemQueue vs System.Threading.Channels

When to use Channels: Simple producer/consumer with bounded capacity. Lower overhead for basic scenarios.

When to use MemQueue: Kafka-style semantics (consumer groups, offset management, partitioning) in-process.

MemQueue vs Apache Kafka

When to use Kafka: Multi-process/multi-service communication, durable event streaming, cross-datacenter replication.

When to use MemQueue: Single-process scenarios where you want the Kafka programming model without infrastructure overhead.

MemQueue vs MediatR / In-process event bus

When to use MediatR: CQRS patterns, request/response, decoupling request handling from controllers.

When to use MemQueue: High-throughput async message processing, event sourcing, work queues with consumer group semantics.

Architecture

┌─────────────────────────────────────────────────────┐
│                        Bus                          │
│         (IMessageBus + IDomainEventBus)              │
├──────────┬──────────┬───────────┬───────────────────┤
│Producer<T│Consumer<T│ Subscripti│     EventBus      │
│    >     │    >     │onManager  │  (fire-and-forget)│
├──────────┴──────────┴───────────┴───────────────────┤
│                  ClientFactory                      │
├─────────────────────────────────────────────────────┤
│               GroupCoordinator                      │
│    (consumer group registry, offset commits)        │
├─────────────────────────────────────────────────────┤
│                  TopicManager                       │
│    (topic/partition lifecycle, store lookups)       │
├─────────────────────────────────────────────────────┤
│  PartitionLog<T>  │  PartitionLog<T>  │  ...        │
│  (typed facade)   │  (typed facade)   │             │
├───────────────────┼───────────────────┤             │
│  AppendOnlyLog    │  AppendOnlyLog    │  ...        │
│  (ring buffer)    │  (ring buffer)    │             │
└───────────────────┴───────────────────┴─────────────┘
         │                           │
  BackpressureCoordinator    RetentionManager
  (watermarks, slow consumer)  (background trim)

Core Components

Delivery Guarantees

Thread Safety

• AppendOnlyLog uses Interlocked for offset counters, SemaphoreSlim for blocking writes, and AsyncReaderWriterLock for retention reads
• ConsumerGroup uses ReaderWriterLockSlim for member/assignment mutations
• GroupCoordinator uses ConcurrentDictionary for group registry
• TopicManager uses ConcurrentDictionary for topic registry

Performance Characteristics

• Zero-allocation partitioning — FNV-1a hash for key routing (no SHA256 allocation)
• Lock-free offset counters — Interlocked.Increment for head offset
• Async notification — TaskCompletionSource-based pulse (no polling)
• Ring buffer — Fixed-size StoredEntry?[] with modular indexing, no resizing
• Partition fan-out — Task.WhenAny across all assigned partitions for fast consume

Backpressure Monitoring

var monitor = bus.BackpressureMonitor;

monitor.HighWatermarkReached += (topic, partition, utilization) =>
    Console.WriteLine($"WARNING: {topic}/{partition} at {utilization:P0}");

monitor.LowWatermarkRecovered += (topic, partition, utilization) =>
    Console.WriteLine($"RECOVERED: {topic}/{partition} at {utilization:P0}");

monitor.SlowConsumerDetected += (topic, partition, groupId, lag) =>
    Console.WriteLine($"SLOW CONSUMER: {groupId} on {topic}/{partition} lag={lag}");

var util = monitor.GetUtilization("orders", 0);
var lag = monitor.GetConsumerLag("orders", 0);
var blocked = monitor.GetBlockedProducerCount("orders", 0);

Dead Letter Queue

var dlq = new InMemoryDeadLetterQueue(maxErrorsPerTopic: 100);
var errors = dlq.GetErrors("orders");
foreach (var (error, message) in errors)
{
    Console.WriteLine($"Failed: {error.ErrorType} at offset {error.Offset}: {error.ErrorMessage}");
}

Topic Statistics

var topicManager = provider.GetRequiredService<ITopicManager>();
var stats = topicManager.GetTopicStatistics((TopicId)"orders");

Console.WriteLine($"Topic: {stats.Topic} | Partitions: {stats.PartitionCount} | Messages: {stats.TotalMessages}");

foreach (var (pid, p) in stats.Partitions)
{
    Console.WriteLine($"  P{pid}: head={p.HeadOffset} tail={p.TailOffset} count={p.MessageCount} util={p.Utilization:P1}");
}

Running Tests & Benchmarks

# Run tests
dotnet test tests/MemQueue.Tests/

# Run benchmarks
dotnet run --project tests/MemQueue.Benchmarks/

License

This project is licensed under the MIT License.