Enable Redis unit tests (dotnet#32601)

BrennanConroy · web-flow · commit d8906c8523f0 · 2021-05-13T05:07:52.000Z
diff --git a/src/SignalR/server/Specification.Tests/src/HubLifetimeManagerTestBase.cs b/src/SignalR/server/Specification.Tests/src/HubLifetimeManagerTestBase.cs
@@ -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]);
@@ -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]);
@@ -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]);
@@ -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]);
@@ -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]);
diff --git a/src/SignalR/server/StackExchangeRedis/test/RedisHubLifetimeManagerTests.cs b/src/SignalR/server/StackExchangeRedis/test/RedisHubLifetimeManagerTests.cs
@@ -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;
@@ -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; }
@@ -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();
@@ -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);
+        }
     }
 }
diff --git a/src/SignalR/server/StackExchangeRedis/test/TestConnectionMultiplexer.cs b/src/SignalR/server/StackExchangeRedis/test/TestConnectionMultiplexer.cs
@@ -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;
@@ -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)
@@ -167,7 +169,7 @@ public string GetStormLog()
 
         public ISubscriber GetSubscriber(object asyncState = null)
         {
-            return _subscriber;
+            return new TestSubscriber(_server);
         }
 
         public int HashSlot(RedisKey key)
@@ -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);
                 }
@@ -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
+                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();
 
@@ -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)
@@ -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)
@@ -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)
@@ -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)

Original file line number	Diff line number	Diff line change
`@@ -4,6 +4,7 @@`
`4`	`4`	`using System.Threading.Tasks;`
`5`	`5`	`using Microsoft.AspNetCore.SignalR.Internal;`
`6`	`6`	`using Microsoft.AspNetCore.SignalR.Protocol;`
	`7`	`+using Microsoft.AspNetCore.SignalR.Specification.Tests;`
`7`	`8`	`using Microsoft.AspNetCore.SignalR.Tests;`
`8`	`9`	`using Microsoft.AspNetCore.Testing;`
`9`	`10`	`using Microsoft.Extensions.Logging.Abstractions;`
`@@ -14,9 +15,10 @@`
`14`	`15`
`15`	`16`	`namespace Microsoft.AspNetCore.SignalR.StackExchangeRedis.Tests`
`16`	`17`	`{`
`17`		`- // Add ScaleoutHubLifetimeManagerTests<TestRedisServer> back after https://github.com/aspnet/SignalR/issues/3088`
`18`		`- public class RedisHubLifetimeManagerTests`
	`18`	`+ public class RedisHubLifetimeManagerTests : ScaleoutHubLifetimeManagerTests<TestRedisServer>`
`19`	`19`	`{`
	`20`	`+ private TestRedisServer _server;`
	`21`	`+`
`20`	`22`	`public class TestObject`
`21`	`23`	`{`
`22`	`24`	`public string TestProperty { get; set; }`
`@@ -37,7 +39,7 @@ private RedisHubLifetimeManager<Hub> CreateLifetimeManager(TestRedisServer serve`
`37`	`39`	`}, NullLogger<DefaultHubProtocolResolver>.Instance));`
`38`	`40`	`}`
`39`	`41`
`40`		`- [Fact(Skip = "https://github.com/aspnet/SignalR/issues/3088")]`
	`42`	`+ [Fact]`
`41`	`43`	`public async Task CamelCasedJsonIsPreservedAcrossRedisBoundary()`
`42`	`44`	`{`
`43`	`45`	`var server = new TestRedisServer();`
`@@ -80,5 +82,21 @@ public async Task CamelCasedJsonIsPreservedAcrossRedisBoundary()`
`80`	`82`	`});`
`81`	`83`	`}`
`82`	`84`	`}`
	`85`	`+`
	`86`	`+ public override TestRedisServer CreateBackplane()`
	`87`	`+ {`
	`88`	`+ return new TestRedisServer();`
	`89`	`+ }`
	`90`	`+`
	`91`	`+ public override HubLifetimeManager<Hub> CreateNewHubLifetimeManager()`
	`92`	`+ {`
	`93`	`+ _server = new TestRedisServer();`
	`94`	`+ return CreateLifetimeManager(_server);`
	`95`	`+ }`
	`96`	`+`
	`97`	`+ public override HubLifetimeManager<Hub> CreateNewHubLifetimeManager(TestRedisServer backplane)`
	`98`	`+ {`
	`99`	`+ return CreateLifetimeManager(backplane);`
	`100`	`+ }`
`83`	`101`	`}`
`84`	`102`	`}`
Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,8 @@`
`6`	`6`	`using System.Collections.Generic;`
`7`	`7`	`using System.IO;`
`8`	`8`	`using System.Net;`
	`9`	`+using System.Reflection;`
	`10`	`+using System.Threading;`
`9`	`11`	`using System.Threading.Tasks;`
`10`	`12`	`using StackExchange.Redis;`
`11`	`13`	`using StackExchange.Redis.Profiling;`
`@@ -73,11 +75,11 @@ public event EventHandler<HashSlotMovedEventArgs> HashSlotMoved`
`73`	`75`	`remove { }`
`74`	`76`	`}`
`75`	`77`
`76`		`- private readonly ISubscriber _subscriber;`
	`78`	`+ private readonly TestRedisServer _server;`
`77`	`79`
`78`	`80`	`public TestConnectionMultiplexer(TestRedisServer server)`
`79`	`81`	`{`
`80`		`- _subscriber = new TestSubscriber(server);`
	`82`	`+ _server = server;`
`81`	`83`	`}`
`82`	`84`
`83`	`85`	`public void BeginProfiling(object forContext)`
`@@ -167,7 +169,7 @@ public string GetStormLog()`
`167`	`169`
`168`	`170`	`public ISubscriber GetSubscriber(object asyncState = null)`
`169`	`171`	`{`
`170`		`- return _subscriber;`
	`172`	`+ return new TestSubscriber(_server);`
`171`	`173`	`}`
`172`	`174`
`173`	`175`	`public int HashSlot(RedisKey key)`
`@@ -223,14 +225,14 @@ public void ExportConfiguration(Stream destination, ExportOptions options = (Exp`
`223`	`225`
`224`	`226`	`public class TestRedisServer`
`225`	`227`	`{`
`226`		`- private readonly ConcurrentDictionary<RedisChannel, List<Action<RedisChannel, RedisValue>>> _subscriptions =`
`227`		`- new ConcurrentDictionary<RedisChannel, List<Action<RedisChannel, RedisValue>>>();`
	`228`	`+ private readonly ConcurrentDictionary<RedisChannel, List<(int, Action<RedisChannel, RedisValue>)>> _subscriptions =`
	`229`	`+ new ConcurrentDictionary<RedisChannel, List<(int, Action<RedisChannel, RedisValue>)>>();`
`228`	`230`
`229`	`231`	`public long Publish(RedisChannel channel, RedisValue message, CommandFlags flags = CommandFlags.None)`
`230`	`232`	`{`
`231`	`233`	`if (_subscriptions.TryGetValue(channel, out var handlers))`
`232`	`234`	`{`
`233`		`- foreach (var handler in handlers)`
	`235`	`+ foreach (var (_, handler) in handlers)`
`234`	`236`	`{`
`235`	`237`	`handler(channel, message);`
`236`	`238`	`}`
`@@ -239,26 +241,37 @@ public long Publish(RedisChannel channel, RedisValue message, CommandFlags flags`
`239`	`241`	`return handlers != null ? handlers.Count : 0;`
`240`	`242`	`}`
`241`	`243`
`242`		`- public void Subscribe(RedisChannel channel, Action<RedisChannel, RedisValue> handler, CommandFlags flags = CommandFlags.None)`
	`244`	`+ public void Subscribe(ChannelMessageQueue messageQueue, int subscriberId, CommandFlags flags = CommandFlags.None)`
`243`	`245`	`{`
`244`		`- _subscriptions.AddOrUpdate(channel, _ => new List<Action<RedisChannel, RedisValue>> { handler }, (_, list) =>`
	`246`	`+ Action<RedisChannel, RedisValue> handler = (channel, value) =>`
	`247`	`+ {`
	`248`	`+ // Workaround for https://github.com/StackExchange/StackExchange.Redis/issues/969`
	`249`	`+ // ChannelMessageQueue isn't mockable currently, this works around that by using private reflection`
	`250`	`+ typeof(ChannelMessageQueue).GetMethod("Write", BindingFlags.NonPublic \| BindingFlags.Instance)`
	`251`	`+ .Invoke(messageQueue, new object[] { channel, value });`
	`252`	`+ };`
	`253`	`+`
	`254`	`+ _subscriptions.AddOrUpdate(messageQueue.Channel, _ => new List<(int, Action<RedisChannel, RedisValue>)> { (subscriberId, handler) }, (_, list) =>`
`245`	`255`	`{`
`246`		`- list.Add(handler);`
	`256`	`+ list.Add((subscriberId, handler));`
`247`	`257`	`return list;`
`248`	`258`	`});`
`249`	`259`	`}`
`250`	`260`
`251`		`- public void Unsubscribe(RedisChannel channel, Action<RedisChannel, RedisValue> handler = null, CommandFlags flags = CommandFlags.None)`
	`261`	`+ public void Unsubscribe(RedisChannel channel, int subscriberId, CommandFlags flags = CommandFlags.None)`
`252`	`262`	`{`
`253`	`263`	`if (_subscriptions.TryGetValue(channel, out var list))`
`254`	`264`	`{`
`255`		`- list.Remove(handler);`
	`265`	`+ list.RemoveAll((item) => item.Item1 == subscriberId);`
`256`	`266`	`}`
`257`	`267`	`}`
`258`	`268`	`}`
`259`	`269`
`260`	`270`	`public class TestSubscriber : ISubscriber`
`261`	`271`	`{`
	`272`	`+ private static int StaticId;`
	`273`	`+`
	`274`	`+ private readonly int _id;`
`262`	`275`	`private readonly TestRedisServer _server;`
`263`	`276`	`public ConnectionMultiplexer Multiplexer => throw new NotImplementedException();`
`264`	`277`
`@@ -267,6 +280,7 @@ public class TestSubscriber : ISubscriber`
`267`	`280`	`public TestSubscriber(TestRedisServer server)`
`268`	`281`	`{`
`269`	`282`	`_server = server;`
	`283`	`+ _id = Interlocked.Increment(ref StaticId);`
`270`	`284`	`}`
`271`	`285`
`272`	`286`	`public EndPoint IdentifyEndpoint(RedisChannel channel, CommandFlags flags = CommandFlags.None)`
`@@ -307,7 +321,7 @@ public async Task<long> PublishAsync(RedisChannel channel, RedisValue message, C`
`307`	`321`
`308`	`322`	`public void Subscribe(RedisChannel channel, Action<RedisChannel, RedisValue> handler, CommandFlags flags = CommandFlags.None)`
`309`	`323`	`{`
`310`		`- _server.Subscribe(channel, handler, flags);`
	`324`	`+ throw new NotImplementedException();`
`311`	`325`	`}`
`312`	`326`
`313`	`327`	`public Task SubscribeAsync(RedisChannel channel, Action<RedisChannel, RedisValue> handler, CommandFlags flags = CommandFlags.None)`
`@@ -328,7 +342,7 @@ public bool TryWait(Task task)`
`328`	`342`
`329`	`343`	`public void Unsubscribe(RedisChannel channel, Action<RedisChannel, RedisValue> handler = null, CommandFlags flags = CommandFlags.None)`
`330`	`344`	`{`
`331`		`- _server.Unsubscribe(channel, handler, flags);`
	`345`	`+ _server.Unsubscribe(channel, _id, flags);`
`332`	`346`	`}`
`333`	`347`
`334`	`348`	`public void UnsubscribeAll(CommandFlags flags = CommandFlags.None)`
`@@ -364,7 +378,15 @@ public void WaitAll(params Task[] tasks)`
`364`	`378`
`365`	`379`	`public ChannelMessageQueue Subscribe(RedisChannel channel, CommandFlags flags = CommandFlags.None)`
`366`	`380`	`{`
`367`		`- throw new NotImplementedException();`
	`381`	`+ // Workaround for https://github.com/StackExchange/StackExchange.Redis/issues/969`
	`382`	`+ var redisSubscriberType = typeof(RedisChannel).Assembly.GetType("StackExchange.Redis.RedisSubscriber");`
	`383`	`+ var ctor = typeof(ChannelMessageQueue).GetConstructor(BindingFlags.Instance \| BindingFlags.NonPublic,`
	`384`	`+ binder: null,`
	`385`	`+ new Type[] { typeof(RedisChannel).MakeByRefType(), redisSubscriberType }, modifiers: null);`
	`386`	`+`
	`387`	`+ var queue = (ChannelMessageQueue)ctor.Invoke(new object[] { channel, null });`
	`388`	`+ _server.Subscribe(queue, _id);`
	`389`	`+ return queue;`
`368`	`390`	`}`
`369`	`391`
`370`	`392`	`public Task<ChannelMessageQueue> SubscribeAsync(RedisChannel channel, CommandFlags flags = CommandFlags.None)`