Skip to content

Enable Redis unit tests #32601

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
May 13, 2021
Merged

Enable Redis unit tests #32601

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9c39409
Enable Redis unit tests
BrennanConroy Nov 19, 2020
c4467f4
fixups
BrennanConroy Apr 30, 2021
e64251f
update
BrennanConroy May 11, 2021
47694e3
Apply suggestions from code review
BrennanConroy May 13, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions src/SignalR/server/Specification.Tests/src/HubLifetimeManagerTestBase.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,12 +47,12 @@ public async Task SendAllAsyncWritesToAllConnectionsOutput()

await manager.SendAllAsync("Hello", new object[] { "World" }).DefaultTimeout();

var message = Assert.IsType<InvocationMessage>(client1.TryRead());
var message = Assert.IsType<InvocationMessage>(await client1.ReadAsync().DefaultTimeout());
Assert.Equal("Hello", message.Target);
Assert.Single(message.Arguments);
Assert.Equal("World", (string)message.Arguments[0]);

message = Assert.IsType<InvocationMessage>(client2.TryRead());
message = Assert.IsType<InvocationMessage>(await client2.ReadAsync().DefaultTimeout());
Assert.Equal("Hello", message.Target);
Assert.Single(message.Arguments);
Assert.Equal("World", (string)message.Arguments[0]);
Expand Down Expand Up @@ -80,7 +80,7 @@ public async Task SendAllAsyncDoesNotWriteToDisconnectedConnectionsOutput()

await manager.SendAllAsync("Hello", new object[] { "World" }).DefaultTimeout();

var message = Assert.IsType<InvocationMessage>(client1.TryRead());
var message = Assert.IsType<InvocationMessage>(await client1.ReadAsync().DefaultTimeout());
Assert.Equal("Hello", message.Target);
Assert.Single(message.Arguments);
Assert.Equal("World", (string)message.Arguments[0]);
Expand Down Expand Up @@ -110,7 +110,7 @@ public async Task SendGroupAsyncWritesToAllConnectionsInGroupOutput()

await manager.SendGroupAsync("group", "Hello", new object[] { "World" }).DefaultTimeout();

var message = Assert.IsType<InvocationMessage>(client1.TryRead());
var message = Assert.IsType<InvocationMessage>(await client1.ReadAsync().DefaultTimeout());
Assert.Equal("Hello", message.Target);
Assert.Single(message.Arguments);
Assert.Equal("World", (string)message.Arguments[0]);
Expand Down Expand Up @@ -141,7 +141,7 @@ public async Task SendGroupExceptAsyncDoesNotWriteToExcludedConnections()

await manager.SendGroupExceptAsync("group1", "Hello", new object[] { "World" }, new[] { connection2.ConnectionId }).DefaultTimeout();

var message = Assert.IsType<InvocationMessage>(client1.TryRead());
var message = Assert.IsType<InvocationMessage>(await client1.ReadAsync().DefaultTimeout());
Assert.Equal("Hello", message.Target);
Assert.Single(message.Arguments);
Assert.Equal("World", (string)message.Arguments[0]);
Expand All @@ -166,7 +166,7 @@ public async Task SendConnectionAsyncWritesToConnectionOutput()

await manager.SendConnectionAsync(connection.ConnectionId, "Hello", new object[] { "World" }).DefaultTimeout();

var message = Assert.IsType<InvocationMessage>(client.TryRead());
var message = Assert.IsType<InvocationMessage>(await client.ReadAsync().DefaultTimeout());
Assert.Equal("Hello", message.Target);
Assert.Single(message.Arguments);
Assert.Equal("World", (string)message.Arguments[0]);
Expand Down
24 changes: 21 additions & 3 deletions src/SignalR/server/StackExchangeRedis/test/RedisHubLifetimeManagerTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
using System.Threading.Tasks;
using Microsoft.AspNetCore.SignalR.Internal;
using Microsoft.AspNetCore.SignalR.Protocol;
using Microsoft.AspNetCore.SignalR.Specification.Tests;
using Microsoft.AspNetCore.SignalR.Tests;
using Microsoft.AspNetCore.Testing;
using Microsoft.Extensions.Logging.Abstractions;
Expand All @@ -14,9 +15,10 @@

namespace Microsoft.AspNetCore.SignalR.StackExchangeRedis.Tests
{
// Add ScaleoutHubLifetimeManagerTests<TestRedisServer> back after https://github.com/aspnet/SignalR/issues/3088
public class RedisHubLifetimeManagerTests
public class RedisHubLifetimeManagerTests : ScaleoutHubLifetimeManagerTests<TestRedisServer>
{
private TestRedisServer _server;

public class TestObject
{
public string TestProperty { get; set; }
Expand All @@ -37,7 +39,7 @@ private RedisHubLifetimeManager<Hub> CreateLifetimeManager(TestRedisServer serve
}, NullLogger<DefaultHubProtocolResolver>.Instance));
}

[Fact(Skip = "https://github.com/aspnet/SignalR/issues/3088")]
[Fact]
public async Task CamelCasedJsonIsPreservedAcrossRedisBoundary()
{
var server = new TestRedisServer();
Expand Down Expand Up @@ -80,5 +82,21 @@ public async Task CamelCasedJsonIsPreservedAcrossRedisBoundary()
});
}
}

public override TestRedisServer CreateBackplane()
{
return new TestRedisServer();
}

public override HubLifetimeManager<Hub> CreateNewHubLifetimeManager()
{
_server = new TestRedisServer();
return CreateLifetimeManager(_server);
}

public override HubLifetimeManager<Hub> CreateNewHubLifetimeManager(TestRedisServer backplane)
{
return CreateLifetimeManager(backplane);
}
}
}
50 changes: 36 additions & 14 deletions src/SignalR/server/StackExchangeRedis/test/TestConnectionMultiplexer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using StackExchange.Redis;
using StackExchange.Redis.Profiling;
Expand Down Expand Up @@ -73,11 +75,11 @@ public event EventHandler<HashSlotMovedEventArgs> HashSlotMoved
remove { }
}

private readonly ISubscriber _subscriber;
private readonly TestRedisServer _server;

public TestConnectionMultiplexer(TestRedisServer server)
{
_subscriber = new TestSubscriber(server);
_server = server;
}

public void BeginProfiling(object forContext)
Expand Down Expand Up @@ -167,7 +169,7 @@ public string GetStormLog()

public ISubscriber GetSubscriber(object asyncState = null)
{
return _subscriber;
return new TestSubscriber(_server);
}

public int HashSlot(RedisKey key)
Expand Down Expand Up @@ -223,14 +225,14 @@ public void ExportConfiguration(Stream destination, ExportOptions options = (Exp

public class TestRedisServer
{
private readonly ConcurrentDictionary<RedisChannel, List<Action<RedisChannel, RedisValue>>> _subscriptions =
new ConcurrentDictionary<RedisChannel, List<Action<RedisChannel, RedisValue>>>();
private readonly ConcurrentDictionary<RedisChannel, List<(int, Action<RedisChannel, RedisValue>)>> _subscriptions =
new ConcurrentDictionary<RedisChannel, List<(int, Action<RedisChannel, RedisValue>)>>();

public long Publish(RedisChannel channel, RedisValue message, CommandFlags flags = CommandFlags.None)
{
if (_subscriptions.TryGetValue(channel, out var handlers))
{
foreach (var handler in handlers)
foreach (var (_, handler) in handlers)
{
handler(channel, message);
}
Expand All @@ -239,26 +241,37 @@ public long Publish(RedisChannel channel, RedisValue message, CommandFlags flags
return handlers != null ? handlers.Count : 0;
}

public void Subscribe(RedisChannel channel, Action<RedisChannel, RedisValue> handler, CommandFlags flags = CommandFlags.None)
public void Subscribe(ChannelMessageQueue messageQueue, int subscriberId, CommandFlags flags = CommandFlags.None)
{
_subscriptions.AddOrUpdate(channel, _ => new List<Action<RedisChannel, RedisValue>> { handler }, (_, list) =>
Action<RedisChannel, RedisValue> handler = (channel, value) =>
{
// Workaround for https://github.com/StackExchange/StackExchange.Redis/issues/969
// ChannelMessageQueue isn't mockable currently, this works around that by using private reflection
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should link to the issue both places we do the crazy private reflection.

typeof(ChannelMessageQueue).GetMethod("Write", BindingFlags.NonPublic | BindingFlags.Instance)
.Invoke(messageQueue, new object[] { channel, value });
};

_subscriptions.AddOrUpdate(messageQueue.Channel, _ => new List<(int, Action<RedisChannel, RedisValue>)> { (subscriberId, handler) }, (_, list) =>
{
list.Add(handler);
list.Add((subscriberId, handler));
return list;
});
}

public void Unsubscribe(RedisChannel channel, Action<RedisChannel, RedisValue> handler = null, CommandFlags flags = CommandFlags.None)
public void Unsubscribe(RedisChannel channel, int subscriberId, CommandFlags flags = CommandFlags.None)
{
if (_subscriptions.TryGetValue(channel, out var list))
{
list.Remove(handler);
list.RemoveAll((item) => item.Item1 == subscriberId);
}
}
}

public class TestSubscriber : ISubscriber
{
private static int StaticId;

private readonly int _id;
private readonly TestRedisServer _server;
public ConnectionMultiplexer Multiplexer => throw new NotImplementedException();

Expand All @@ -267,6 +280,7 @@ public class TestSubscriber : ISubscriber
public TestSubscriber(TestRedisServer server)
{
_server = server;
_id = Interlocked.Increment(ref StaticId);
}

public EndPoint IdentifyEndpoint(RedisChannel channel, CommandFlags flags = CommandFlags.None)
Expand Down Expand Up @@ -307,7 +321,7 @@ public async Task<long> PublishAsync(RedisChannel channel, RedisValue message, C

public void Subscribe(RedisChannel channel, Action<RedisChannel, RedisValue> handler, CommandFlags flags = CommandFlags.None)
{
_server.Subscribe(channel, handler, flags);
throw new NotImplementedException();
}

public Task SubscribeAsync(RedisChannel channel, Action<RedisChannel, RedisValue> handler, CommandFlags flags = CommandFlags.None)
Expand All @@ -328,7 +342,7 @@ public bool TryWait(Task task)

public void Unsubscribe(RedisChannel channel, Action<RedisChannel, RedisValue> handler = null, CommandFlags flags = CommandFlags.None)
{
_server.Unsubscribe(channel, handler, flags);
_server.Unsubscribe(channel, _id, flags);
}

public void UnsubscribeAll(CommandFlags flags = CommandFlags.None)
Expand Down Expand Up @@ -364,7 +378,15 @@ public void WaitAll(params Task[] tasks)

public ChannelMessageQueue Subscribe(RedisChannel channel, CommandFlags flags = CommandFlags.None)
{
throw new NotImplementedException();
// Workaround for https://github.com/StackExchange/StackExchange.Redis/issues/969
var redisSubscriberType = typeof(RedisChannel).Assembly.GetType("StackExchange.Redis.RedisSubscriber");
var ctor = typeof(ChannelMessageQueue).GetConstructor(BindingFlags.Instance | BindingFlags.NonPublic,
binder: null,
new Type[] { typeof(RedisChannel).MakeByRefType(), redisSubscriberType }, modifiers: null);

var queue = (ChannelMessageQueue)ctor.Invoke(new object[] { channel, null });
_server.Subscribe(queue, _id);
return queue;
}

public Task<ChannelMessageQueue> SubscribeAsync(RedisChannel channel, CommandFlags flags = CommandFlags.None)
Expand Down