Skip to content

Commit

Permalink
Support constructor injection with IConnectionMultiplexerFactory
Browse files Browse the repository at this point in the history
  • Loading branch information
mayuki committed May 22, 2023
1 parent 33da55b commit 0b41e9a
Show file tree
Hide file tree
Showing 3 changed files with 89 additions and 5 deletions.
4 changes: 1 addition & 3 deletions src/MessagePipe.Redis/MessagePipeRedisOptions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,11 @@ public interface IConnectionMultiplexerFactory

public sealed class MessagePipeRedisOptions
{
public IConnectionMultiplexerFactory ConnectionMultiplexerFactory { get; }
public IRedisSerializer RedisSerializer { get; set; }

public MessagePipeRedisOptions(IConnectionMultiplexerFactory connectionMultiplexerFactory)
public MessagePipeRedisOptions()
{
this.RedisSerializer = new MessagePackRedisSerializer();
this.ConnectionMultiplexerFactory = connectionMultiplexerFactory;
}
}
}
21 changes: 19 additions & 2 deletions src/MessagePipe.Redis/ServiceCollectionRedisExtensions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -20,17 +20,34 @@ public static IServiceCollection AddMessagePipeRedis(this IServiceCollection ser
return AddMessagePipeRedis(services, connectionMultiplexerFactory, _ => { });
}

public static IServiceCollection AddMessagePipeRedis<T>(this IServiceCollection services)
where T : class, IConnectionMultiplexerFactory
{
return AddMessagePipeRedis<T>(services, _ => { });
}

public static IServiceCollection AddMessagePipeRedis<T>(this IServiceCollection services, Action<MessagePipeRedisOptions> configure)
where T : class, IConnectionMultiplexerFactory
{
return AddMessagePipeRedis(services, ServiceDescriptor.Singleton<IConnectionMultiplexerFactory, T>(), configure);
}

public static IServiceCollection AddMessagePipeRedis(this IServiceCollection services, IConnectionMultiplexer connectionMultiplexer, Action<MessagePipeRedisOptions> configure)
{
return AddMessagePipeRedis(services, new SingleConnectionMultiplexerFactory(connectionMultiplexer), configure);
}

public static IServiceCollection AddMessagePipeRedis(this IServiceCollection services, IConnectionMultiplexerFactory connectionMultiplexerFactory, Action<MessagePipeRedisOptions> configure)
{
var options = new MessagePipeRedisOptions(connectionMultiplexerFactory);
return AddMessagePipeRedis(services, ServiceDescriptor.Singleton(connectionMultiplexerFactory), configure);
}

static IServiceCollection AddMessagePipeRedis(IServiceCollection services, ServiceDescriptor connectionMultiplexerServiceDesc, Action<MessagePipeRedisOptions> configure)
{
var options = new MessagePipeRedisOptions();
configure(options);
services.AddSingleton(options); // add as singleton instance
services.AddSingleton<IConnectionMultiplexerFactory>(options.ConnectionMultiplexerFactory);
services.Add(connectionMultiplexerServiceDesc);
services.AddSingleton<IRedisSerializer>(options.RedisSerializer);

services.Add(typeof(IDistributedPublisher<,>), typeof(RedisPublisher<,>), InstanceLifetime.Singleton);
Expand Down
69 changes: 69 additions & 0 deletions tests/MessagePipe.Redis.Tests/ConnectionMultiplexerFactoryTest.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using FluentAssertions;
using MessagePipe.Tests;
using Microsoft.Extensions.DependencyInjection;
using StackExchange.Redis;
using Xunit;

namespace MessagePipe.Redis.Tests;

public class ConnectionMultiplexerFactoryTest
{
[Fact]
public async Task CreateFromInstance()
{
var connection = TestHelper.GetLocalConnectionMultiplexer();
var services = new ServiceCollection();
services.AddMessagePipe();
services.AddMessagePipeRedis(connection);
var sp = services.BuildServiceProvider();

var publisher = sp.GetRequiredService<IDistributedPublisher<string, string>>();
var subscriber = sp.GetRequiredService<IDistributedSubscriber<string, string>>();

var results = new List<string>();
await using var _ = await subscriber.SubscribeAsync("Foo", x => results.Add(x));

await publisher.PublishAsync("Foo", "Bar");
await Task.Delay(250);
results.FirstOrDefault().Should().Be("Bar");
}

[Fact]
public async Task CreateFromGenericType()
{
var connection = TestHelper.GetLocalConnectionMultiplexer();
var services = new ServiceCollection();
services.AddSingleton<IConnectionMultiplexer>(connection);
services.AddMessagePipe();
services.AddMessagePipeRedis<TestConnectionMultiplexerFactory>();
var sp = services.BuildServiceProvider();

var publisher = sp.GetRequiredService<IDistributedPublisher<string, string>>();
var subscriber = sp.GetRequiredService<IDistributedSubscriber<string, string>>();

var results = new List<string>();
await using var _ = await subscriber.SubscribeAsync("Foo", x => results.Add(x));

await publisher.PublishAsync("Foo", "Bar");
await Task.Delay(250);
results.FirstOrDefault().Should().Be("Bar");
}

public class TestConnectionMultiplexerFactory : IConnectionMultiplexerFactory
{
readonly IConnectionMultiplexer connectionMultiplexer;

public TestConnectionMultiplexerFactory(IConnectionMultiplexer connectionMultiplexer)
{
this.connectionMultiplexer = connectionMultiplexer;
}

public IConnectionMultiplexer GetConnectionMultiplexer<TKey>(TKey key)
{
return connectionMultiplexer;
}
}
}

0 comments on commit 0b41e9a

Please sign in to comment.