EasyCore.RedisStreams 8.0.0

🔴 EasyCore.RedisStreams

EasyCore.RedisStreams is a Redis Streams infrastructure client for .NET 8. Built on StackExchange.Redis, it provides DI registration, connect, consumer-group subscribe/ack, and a unified entry format (RedisHeader / type / payload). It is not tied to IEvent / EventBus — use it standalone, or via the EasyCore.EventBus.RedisStreams adapter.

🌍 Language

• Chinese: README.md
• English (this document)

📚 Table of Contents

• 1. Positioning
• 2. Relation to EventBus
• 3. Requirements
• 4. Installation
• 5. Quick Start
• 6. Message Format (RedisStreamMessageFormat)
• 7. API Overview
• 8. Options (RedisStreamsOptions)
• 9. FAQ
• 10. License

1. 🎯 Positioning

1.1 Design Principles

2. Relation to EventBus

┌─────────────────────────────┐
│  App code (JSON payload)    │
└──────────────┬──────────────┘
               │ IRedisStreamsClient
               ▼
┌─────────────────────────────┐
│ EasyCore.RedisStreams (this)│  ← infrastructure client
└──────────────┬──────────────┘
               │ StackExchange.Redis
               ▼
            Redis

┌─────────────────────────────┐
│ EasyCore.EventBus.RedisStreams │  ← adapter: IEvent / Handler
│   └── uses this client         │
└─────────────────────────────┘

3. ⚙ Requirements

4. 📥 Installation

dotnet add package EasyCore.RedisStreams

5. ⚡ Quick Start

5️⃣.1️⃣ Register DI

using EasyCore.RedisStreams;

builder.Services.EasyCoreRedisStreams(o =>
{
    o.EndPoints = new List<string> { "localhost:6379" };
    o.User = null;           // ACL user (optional)
    o.Password = null;       // password (optional)
    o.ConnectTimeout = 10;   // seconds
    o.SyncTimeout = 10;      // seconds
    o.DefaultDatabase = 0;
    o.ConsumerGroup = "RedisGroup";
    o.AppName = "MyApp";
});

5️⃣.2️⃣ Publish

public sealed class NotifyPublisher
{
    private readonly IRedisStreamsClient _client;

    public NotifyPublisher(IRedisStreamsClient client) => _client = client;

    public async Task PublishAsync(string userId, CancellationToken ct = default)
    {
        await _client.ConnectAsync(ct);

        var payload = $$"""{"userId":"{{userId}}","channel":"sms"}""";
        var header = new RedisStreamHeader
        {
            RetryCount = 0,
            RetryInterval = 5
        };

        await _client.PublishAsync(
            streamKey: "notify:sms",
            typeName: "SmsNotify",
            payloadJson: payload,
            header: header,
            cancellationToken: ct);
    }
}

5️⃣.3️⃣ Subscribe and Acknowledge

public sealed class NotifyConsumer : BackgroundService
{
    private readonly IRedisStreamsClient _client;

    public NotifyConsumer(IRedisStreamsClient client) => _client = client;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await _client.ConnectAsync(stoppingToken);

        await _client.SubscribeAsync(
            streamKeys: new[] { "notify:sms" },
            handler: async (msg, ct) =>
            {
                // msg.TypeName / msg.PayloadJson / msg.Header
                // handle business…

                await _client.AcknowledgeAsync(msg.StreamKey, msg.MessageId, ct);
            },
            cancellationToken: stoppingToken);
    }
}

RedisStreamsDeliveredMessage fields: StreamKey, MessageId, TypeName, PayloadJson, Header.

6. Message Format (RedisStreamMessageFormat)

Each stream entry uses fixed field names:

RedisStreamHeader:

When header is null, publish writes a default header.

7. 🧩 API Overview

IRedisStreamsClient implements IAsyncDisposable — dispose on shutdown.

8. Options (RedisStreamsOptions)

ToConfigurationOptions() converts these settings to StackExchange.Redis.ConfigurationOptions.

9. ❓ FAQ

Q: This package vs EasyCore.EventBus.RedisStreams?
A: Use the EventBus adapter for IEvent, handlers, and retries. Use this package for stream-key + JSON I/O.

Q: Why fixed RedisHeader / type / payload?
A: So multiple services (and the EventBus adapter) can share the same stream layout.

Q: Is ConnectTimeout in milliseconds?
A: No — Options use seconds; values are multiplied by 1000 for StackExchange.Redis.

Q: Do I need to create the consumer group myself?
A: The client prepares the group during subscribe. Ensure Redis supports Streams and the key is consumable.

Q: Cluster / sentinel?
A: Put endpoints in EndPoints with credentials as needed; advanced topologies follow StackExchange.Redis docs.

10. 📄 License

MIT OR Apache-2.0 — see the repository LICENSE if present, or the NuGet package metadata.

🤝 Contributing

Issues and PRs are welcome. After changing this package, verify EasyCore.EventBus.RedisStreams and related demos.