Merge pull request #24 from Cysharp/feature/ShardedPubSub

Adds support for Redis Sharded Pub/Sub

Author: mayuki

Directory.Packages.props

@@ -12,7 +12,7 @@
     <PackageVersion Include="Microsoft.Extensions.TimeProvider.Testing" Version="8.6.0" />
     <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="17.8.0" />
     <PackageVersion Include="NATS.Net" Version="2.2.2" />
-    <PackageVersion Include="StackExchange.Redis" Version="2.7.33" />
+    <PackageVersion Include="StackExchange.Redis" Version="2.8.41" />
     <PackageVersion Include="Testcontainers.Nats" Version="3.8.0" />
     <PackageVersion Include="Testcontainers.Redis" Version="3.8.0" />
     <PackageVersion Include="xunit" Version="2.9.2" />

src/Multicaster.Distributed.Redis/RedisChannelFactory.cs

@@ -0,0 +1,21 @@
+using StackExchange.Redis;
+
+namespace Cysharp.Runtime.Multicast.Distributed.Redis;
+
+/// <summary>
+/// Provides factory methods for creating <see cref="RedisChannel"/> instances with predefined naming conventions.
+/// </summary>
+public static class RedisChannelFactory
+{
+    private const string Prefix = "Multicaster.Group?name=";
+
+    /// <summary>
+    /// Gets a factory that creates <see cref="RedisChannel"/> instances.
+    /// </summary>
+    public static Func<string, RedisChannel> Default { get; } = static (groupName) => RedisChannel.Literal($"{Prefix}{groupName}");
+
+    /// <summary>
+    /// Gets a factory that creates <see cref="RedisChannel"/> instances for use with Sharded Pub/Sub.
+    /// </summary>
+    public static Func<string, RedisChannel> Sharded { get; } = static (groupName) => RedisChannel.Sharded($"{Prefix}{groupName}");
+}

src/Multicaster.Distributed.Redis/RedisGroup.cs

@@ -28,13 +28,13 @@ internal class RedisGroup<TKey, T> : IMulticastAsyncGroup<TKey, T>, IMulticastSy
 
     internal string Name { get; }
 
-    public RedisGroup(string name, ISubscriber subscriber, IRemoteProxyFactory proxyFactory, IRemoteSerializer serializer, MessagePackSerializerOptions messagePackSerializerOptions, Action<RedisGroup<TKey, T>> onDisposeAction)
+    public RedisGroup(string name, Func<string, RedisChannel> redisChannelFactory, ISubscriber subscriber, IRemoteProxyFactory proxyFactory, IRemoteSerializer serializer, MessagePackSerializerOptions messagePackSerializerOptions, Action<RedisGroup<TKey, T>> onDisposeAction)
     {
         Name = name;
         _subscriber = subscriber;
         _proxyFactory = proxyFactory;
         _serializer = serializer;
-        _channel = new RedisChannel($"Multicaster.Group?name={name}", RedisChannel.PatternMode.Literal);
+        _channel = redisChannelFactory(name);
         _messagePackSerializerOptionsForKey = messagePackSerializerOptions;
         _onDisposeAction = onDisposeAction;
 
@@ -354,4 +354,4 @@ private void ThrowIfDisposed()
             throw new ObjectDisposedException(nameof(RedisGroup<TKey, T>));
         }
     }
-}
+}

src/Multicaster.Distributed.Redis/RedisGroupOptions.cs

@@ -0,0 +1,40 @@
+using MessagePack;
+using StackExchange.Redis;
+
+namespace Cysharp.Runtime.Multicast.Distributed.Redis;
+
+/// <summary>
+/// Represents configuration options for connecting to and interacting with a Redis group.
+/// </summary>
+/// <remarks>This class provides various settings to configure Redis connectivity, key serialization, and channel
+/// creation. Use these options to customize the behavior of Redis-based operations, such as specifying a connection
+/// string, providing a custom <see cref="ConnectionMultiplexer"/>, or defining a key prefix.</remarks>
+public class RedisGroupOptions
+{
+    /// <summary>
+    /// Gets or sets the connection string to connect to Redis. If <see cref="ConnectionMultiplexer"/> property is not set, this will be used.
+    /// </summary>
+    public string ConnectionString { get; set; } = "localhost:6379";
+
+    /// <summary>
+    /// Gets or sets a ConnectionMultiplexer instance to connect to Redis. If this is set, <see cref="ConnectionString"/> property will be ignored.
+    /// </summary>
+    public ConnectionMultiplexer? ConnectionMultiplexer { get; set; }
+
+    /// <summary>
+    /// Gets or sets a prefix for the Redis key.
+    /// </summary>
+    public string? Prefix { get; set; }
+
+    /// <summary>
+    /// Gets or sets a MessagePackSerializerOptions used for serializing the key.
+    /// </summary>
+    public MessagePackSerializerOptions? MessagePackSerializerOptionsForKey { get; set; }
+
+    /// <summary>
+    /// Gets or sets the factory method used to create <see cref="RedisChannel"/> instances. Default is <see cref="RedisChannelFactory.Default"/>.
+    /// </summary>
+    /// <remarks>Use this property to customize how <see cref="RedisChannel"/> instances are created,  such as
+    /// applying specific naming conventions or transformations to channel names.</remarks>
+    public Func<string, RedisChannel> ChannelFactory { get; set; } = RedisChannelFactory.Default;
+}

src/Multicaster.Distributed.Redis/RedisGroupProvider.cs

@@ -1,15 +1,23 @@
 using System.Collections.Concurrent;
+
 using Cysharp.Runtime.Multicast.Remoting;
 
 using MessagePack;
 using MessagePack.Formatters;
 using MessagePack.Resolvers;
 
 using Microsoft.Extensions.Options;
+
 using StackExchange.Redis;
 
 namespace Cysharp.Runtime.Multicast.Distributed.Redis;
 
+/// <summary>
+/// Provides functionality for managing multicast groups using Redis as the underlying communication mechanism.
+/// </summary>
+/// <remarks>This class allows the creation and management of asynchronous and synchronous multicast groups, where
+/// messages can be sent to multiple receivers. It uses Redis channels for communication and supports custom
+/// serialization and deserialization of keys and messages.</remarks>
 public class RedisGroupProvider : IMulticastGroupProvider, IDisposable
 {
     private readonly ConcurrentDictionary<(string Name, Type KeyType, Type ReceiverType), object> _groups = new();
@@ -19,11 +27,32 @@ public class RedisGroupProvider : IMulticastGroupProvider, IDisposable
     private readonly ISubscriber _subscriber;
     private readonly string? _prefix;
     private readonly MessagePackSerializerOptions _messagePackSerializerOptionsForKey;
+    private readonly Func<string, RedisChannel> _channelFactory;
 
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RedisGroupProvider"/> class with the specified proxy factory,
+    /// serializer, and configuration options.
+    /// </summary>
+    /// <param name="proxyFactory">The factory used to create remote proxies for interacting with Redis.</param>
+    /// <param name="serializer">The serializer used to serialize and deserialize data for Redis operations.</param>
+    /// <param name="options">The configuration options for the Redis group provider.</param>
     public RedisGroupProvider(IRemoteProxyFactory proxyFactory, IRemoteSerializer serializer, IOptions<RedisGroupOptions> options)
         : this(proxyFactory, serializer, options.Value)
     {}
 
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RedisGroupProvider"/> class, which provides functionality for
+    /// managing Redis-based group communication using a specified remote proxy factory, serializer, and configuration
+    /// options.
+    /// </summary>
+    /// <remarks>If the <see cref="RedisGroupOptions.ConnectionMultiplexer"/> property in <paramref
+    /// name="options"/> is <see langword="null"/>, a new connection multiplexer is created using the connection string
+    /// specified in <see cref="RedisGroupOptions.ConnectionString"/>. Otherwise, the provided connection multiplexer is
+    /// used.</remarks>
+    /// <param name="proxyFactory">The factory used to create remote proxies for communication. This parameter cannot be <see langword="null"/>.</param>
+    /// <param name="serializer">The serializer used to serialize and deserialize messages. This parameter cannot be <see langword="null"/>.</param>
+    /// <param name="options">The configuration options for the Redis group provider, including connection settings, channel factory, and
+    /// serialization options. This parameter cannot be <see langword="null"/>.</param>
     public RedisGroupProvider(IRemoteProxyFactory proxyFactory, IRemoteSerializer serializer, RedisGroupOptions options)
     {
         _proxyFactory = proxyFactory;
@@ -39,52 +68,33 @@ public RedisGroupProvider(IRemoteProxyFactory proxyFactory, IRemoteSerializer se
             _subscriber = options.ConnectionMultiplexer.GetSubscriber();
         }
         _prefix = options.Prefix;
+        _channelFactory = options.ChannelFactory;
 
         var messagePackSerializerOptions = (options.MessagePackSerializerOptionsForKey ?? MessagePackSerializer.DefaultOptions);
         _messagePackSerializerOptionsForKey = messagePackSerializerOptions.WithResolver(
             CompositeResolver.Create([NativeGuidFormatter.Instance], [messagePackSerializerOptions.Resolver])
         );
     }
 
+    /// <inheritdoc />
     public IMulticastAsyncGroup<TKey, TReceiver> GetOrAddGroup<TKey, TReceiver>(string name)
         where TKey : IEquatable<TKey>
-        => (IMulticastAsyncGroup<TKey, TReceiver>)_groups.GetOrAdd((name, typeof(TKey), typeof(TReceiver)), _ => new RedisGroup<TKey, TReceiver>(_prefix + name, _subscriber, _proxyFactory, _serializer, _messagePackSerializerOptionsForKey, Remove));
+        => (IMulticastAsyncGroup<TKey, TReceiver>)_groups.GetOrAdd((name, typeof(TKey), typeof(TReceiver)), _ => new RedisGroup<TKey, TReceiver>(_prefix + name, _channelFactory, _subscriber, _proxyFactory, _serializer, _messagePackSerializerOptionsForKey, Remove));
 
+    /// <inheritdoc />
     public IMulticastSyncGroup<TKey, TReceiver> GetOrAddSynchronousGroup<TKey, TReceiver>(string name)
         where TKey : IEquatable<TKey>
-        => (IMulticastSyncGroup<TKey, TReceiver>)_groups.GetOrAdd((name, typeof(TKey), typeof(TReceiver)), _ => new RedisGroup<TKey, TReceiver>(_prefix + name, _subscriber, _proxyFactory, _serializer, _messagePackSerializerOptionsForKey, Remove));
+        => (IMulticastSyncGroup<TKey, TReceiver>)_groups.GetOrAdd((name, typeof(TKey), typeof(TReceiver)), _ => new RedisGroup<TKey, TReceiver>(_prefix + name, _channelFactory, _subscriber, _proxyFactory, _serializer, _messagePackSerializerOptionsForKey, Remove));
 
     private void Remove<TKey, TReceiver>(RedisGroup<TKey, TReceiver> group)
         where TKey : IEquatable<TKey>
     {
         _groups.TryRemove((group.Name, typeof(TKey), typeof(TReceiver)), out _);
     }
 
+    /// <inheritdoc />
     public void Dispose()
     {
         _createdConnectionMultiplexer?.Dispose();
     }
 }
-
-public class RedisGroupOptions
-{
-    /// <summary>
-    /// Gets or sets the connection string to connect to Redis. If <see cref="ConnectionMultiplexer"/> property is not set, this will be used.
-    /// </summary>
-    public string ConnectionString { get; set; } = "localhost:6379";
-
-    /// <summary>
-    /// Gets or sets a ConnectionMultiplexer instance to connect to Redis. If this is set, <see cref="ConnectionString"/> property will be ignored.
-    /// </summary>
-    public ConnectionMultiplexer? ConnectionMultiplexer { get; set; }
-
-    /// <summary>
-    /// Gets or sets a prefix for the Redis key.
-    /// </summary>
-    public string? Prefix { get; set; }
-
-    /// <summary>
-    /// Gets or sets a MessagePackSerializerOptions used for serializing the key.
-    /// </summary>
-    public MessagePackSerializerOptions? MessagePackSerializerOptionsForKey { get; set; }
-}