CrispyWaffle.Redis

CrispyWaffle.Redis connects CrispyWaffle to Redis. It has four parts:

• A cache repository for CacheManager.
• A log provider that publishes log messages to Redis Pub/Sub channels.
• A telemetry client that stores hits, events and metrics in Redis.
• Helpers for master/replica setups and quick counter operations.

Use it when several processes or machines need a shared cache, when you want log messages streamed to listeners in real time, or when you want lightweight counters stored in Redis.

How it works

The connector

Everything in this package goes through a RedisConnector. The connector builds a StackExchange.Redis.Extensions RedisConfiguration with these settings:

• AbortOnConnectFail = false and AllowAdmin = true.
• A server enumeration strategy of All servers, Any role, and Throw on unreachable servers.
• A host list parsed from a comma-separated string of host[:port] entries. The port defaults to 6379 when it's missing.

From that configuration, the connector creates a RedisConnectionPoolManager and a RedisClient.

Every connector needs an ISerializer from StackExchange.Redis.Extensions.Core. That serializer converts cached values to and from bytes. The package doesn't include an implementation, so either reference a serializer package such as StackExchange.Redis.Extensions.System.Text.Json or write your own (see Getting started).

You can create a connector in several ways:

• From an IConnection (Host, Port and Credentials.Password are used; the username is ignored).
• From a host and a port.
• From a raw host list.
• From a MasterSlaveConfiguration, which combines two connections.
• From an existing IRedisClient and IRedisConnectionPoolManager.

Register the connector as a singleton, because it owns the connection pool.

QueuePrefix defaults to "crispy-waffle". The Pub/Sub log provider uses it to name channels. The overloads that take a queuePrefix throw ArgumentNullException when it's null.

Cache repository

RedisCacheRepository implements ICacheRepository, so you can add it to CacheManager alongside other repositories.

• The ICacheRepository methods always use database 0 (IRedisClient.Db0) and the connector's serializer.
• Sub-key entries are stored as Redis hashes. SetAsync(value, key, subKey) reads the whole hash, sets one field, and writes the hash back with FireAndForget.
• SetToDatabase, GetFromDatabase, TryGetFromDatabase and RemoveFromDatabase work with any database number and call the connector serializer directly.

Errors depend on ShouldPropagateExceptions, which defaults to false:

• When it's false, failures are logged instead of thrown. Exceptions whose message contains "timeout" go to LogConsumer.Trace; all others go to LogConsumer.Handle.
• OperationCanceledException and argument-validation exceptions are always rethrown.
• GetAsync throws InvalidOperationException when the key is missing. It also throws InvalidOperationException after logging a suppressed error, so a missing key and a failed call look the same to the caller.
• The TryGetAsync overloads return (false, default) in both of those cases instead.

Dispose() disposes the underlying RedisConnector. If several components share one connector, don't dispose the repository until all of them are done.

Pub/Sub log provider

PubSubRedisLogProvider is an ILogProvider. Each logging call that passes the level check builds a LogMessage, serializes it to JSON, and adds it to an in-memory ConcurrentQueue<string>. The LogMessage carries:

• The date, host name and process ID.
• The local and remote IP addresses.
• The operation, user agent, thread ID and thread name.
• A new GUID as the message Id.

The constructor starts a dedicated worker thread named "Message queue Redis log provider worker". The thread waits one second and then loops. Each pass drains the queue and hands every message to the configured IPropagationStrategy. When the queue is empty and the CancellationToken passed to the constructor has been cancelled, the worker exits.

The worker is a foreground thread and doesn't sleep between passes. Cancel the token when your application shuts down, or the thread keeps the process alive.

If publishing a message throws, the worker logs the exception through LogConsumer.Debug and moves on to the next message.

The provider's log level starts at no levels, so it records nothing until you call SetLevel.

Each propagation strategy publishes two messages:

1. The queue name to the channel {QueuePrefix}-queues, so listeners can discover which channels are active.
2. The serialized log message to {QueuePrefix}-{queue name}.

Telemetry client

RedisTelemetryClient implements ITelemetryClient:

• Hits are string counters in database HitDatabase.
• Metrics are hash counters in database MetricDatabase, with one hash per metric and one field per variation.
• Events are serialized strings stored in the database whose number is the event's Category, under the event's Name.
• Exceptions and dependencies are counted in the hashes Exceptions_{suffix} and Dependencies_{suffix} in MetricDatabase.

Writes use CommandFlags.FireAndForget, and reads use CommandFlags.PreferReplica.

Getting started

Start Redis locally. The repository's docker-compose.yml runs redis:6.2-alpine on port 6379 with the password myP@ssw0rd.

Then create a serializer, a connection and a connector, and register them:

using System.Text.Json;
using System.Threading;
using CrispyWaffle.Cache;
using CrispyWaffle.Composition;
using CrispyWaffle.Configuration;
using CrispyWaffle.Log;
using CrispyWaffle.Redis.Cache;
using CrispyWaffle.Redis.Log;
using CrispyWaffle.Redis.Log.PropagationStrategy;
using CrispyWaffle.Redis.Telemetry;
using CrispyWaffle.Redis.Utils.Communications;
using CrispyWaffle.Telemetry;
using StackExchange.Redis.Extensions.Core;

public sealed class SystemTextJsonRedisSerializer : ISerializer
{
    private readonly JsonSerializerOptions _options = new(JsonSerializerDefaults.Web);

    public byte[] Serialize<T>(T item) => JsonSerializer.SerializeToUtf8Bytes(item, _options);

    public T Deserialize<T>(byte[] serializedObject) =>
        JsonSerializer.Deserialize<T>(serializedObject, _options);
}

public static class RedisBootstrap
{
    public static void Configure(CancellationToken shutdownToken)
    {
        var connection = new Connection
        {
            Host = "localhost",
            Port = 6379,
            Credentials = { Password = "myP@ssw0rd" },
        };

        var connector = new RedisConnector(
            connection,
            new SystemTextJsonRedisSerializer(),
            "my-app"
        );

        // One connector per process: it owns the connection pool.
        ServiceLocator.Register(connector);

        // Cache
        CacheManager.AddRepository(new RedisCacheRepository(connector));

        // Logging
        LogConsumer
            .AddProvider(
                new PubSubRedisLogProvider(connector, new GeneralPropagation(), shutdownToken)
            )
            .SetLevel(LogLevel.Production);

        // Telemetry
        TelemetryAnalytics.AddClient(new RedisTelemetryClient(connector, "my-app"));
    }
}

After that, await CacheManager.SetAsync("value", "my-key"), LogConsumer.Info("...") and TelemetryAnalytics.TrackHit("home") all reach Redis.

Reference

RedisConnector

public class RedisConnector : IDisposable owns the Redis connection pool and the IRedisClient that every other type in this package uses. It has the [ConnectionName("Redis")] attribute.

RedisCacheRepository

public class RedisCacheRepository : ICacheRepository, IDisposable is a cache repository backed by Redis. It reads and writes database 0, except for the *Database methods.

PubSubRedisLogProvider

public class PubSubRedisLogProvider : ILogProvider queues log messages and publishes them to Redis Pub/Sub channels from a background thread.

IPropagationStrategy

public interface IPropagationStrategy decides which Redis channel a log message is published to. Implement it to route messages differently.

GeneralPropagation

public sealed class GeneralPropagation : IPropagationStrategy publishes every message to one shared channel. The class is in the file GeneralPropagationStrategy.cs.

OperationPropagationStrategy

public sealed class OperationPropagationStrategy : IPropagationStrategy publishes messages to a channel named after the current operation.

ProcessPropagation

public sealed class ProcessPropagation : IPropagationStrategy publishes messages to a channel named after the current process ID. The class is in the file ProcessPropagationStrategy.cs.

RedisTelemetryClient

public sealed class RedisTelemetryClient : ITelemetryClient stores telemetry counters and events in Redis.

MasterSlaveConfiguration

public sealed class MasterSlaveConfiguration combines a primary connection and a replica connection into one host list for RedisConnector. It has [ConnectionName("RedisMaster", Order = 0)] and [ConnectionName("RedisSlave", Order = 1)].

RedisExtensions

public static class RedisExtensions has shortcuts that run against the RedisConnector registered in ServiceLocator.

The connector is resolved once, when the class is first used. Register RedisConnector before you call any of these methods.

Examples

Register through the service locator

Register the connector and repository as singletons, then let CacheManager resolve the repository.

using CrispyWaffle.Cache;
using CrispyWaffle.Composition;
using CrispyWaffle.Configuration;
using CrispyWaffle.Redis.Cache;
using CrispyWaffle.Redis.Utils.Communications;

ServiceLocator.Register(
    () =>
        new RedisConnector(
            new Connection
            {
                Host = "localhost",
                Port = 6379,
                Credentials = { Password = "myP@ssw0rd" },
            },
            new SystemTextJsonRedisSerializer()
        ),
    Lifetime.Singleton
);

ServiceLocator.Register(
    () => new RedisCacheRepository(ServiceLocator.Resolve<RedisConnector>()),
    Lifetime.Singleton
);

CacheManager.AddRepository<RedisCacheRepository>();

Store and read sub-keys

var repository = ServiceLocator.Resolve<RedisCacheRepository>();

await repository.SetAsync("Alice", "user:42", "name");
await repository.SetAsync("alice@example.com", "user:42", "email");

var (found, name) = await repository.TryGetAsync<string>("user:42", "name");

Connect to a primary and a replica

using CrispyWaffle.Configuration;
using CrispyWaffle.Redis.Utils;
using CrispyWaffle.Redis.Utils.Communications;

var master = new Connection { Host = "redis-primary", Port = 6379, Credentials = { Password = "secret" } };
var slave = new Connection { Host = "redis-replica", Port = 6379 };

var connector = new RedisConnector(
    new MasterSlaveConfiguration(master, slave),
    new SystemTextJsonRedisSerializer()
);

Listen to log messages from another process

The GeneralPropagation strategy publishes to {QueuePrefix}-general. Any Redis client can subscribe to that channel.

using StackExchange.Redis;

connector.Subscriber.Subscribe(
    RedisChannel.Literal($"{connector.QueuePrefix}-general"),
    (channel, message) => Console.WriteLine(message)
);

Track a telemetry event

using CrispyWaffle.Telemetry;

public sealed class CheckoutEvent : ITelemetryEvent<CheckoutData>
{
    public string Name => "checkout:last";

    public int Category => 2; // Redis database number

    public CheckoutData Event { get; init; }
}

public sealed class CheckoutData
{
    public decimal Total { get; set; }
}

TelemetryAnalytics.TrackEvent(new CheckoutEvent { Event = new CheckoutData { Total = 99.90m } }, TimeSpan.FromDays(7));

Rate-limit with a temporary counter

using CrispyWaffle.Redis.Utils;

var calls = RedisExtensions.TemporaryIncrease($"rate:{clientId}", TimeSpan.FromMinutes(1));
if (calls > 100)
{
    throw new InvalidOperationException("Too many requests.");
}

See also

• Caching
• Logging
• Telemetry
• Service locator
• CrispyWaffle.Configuration