Merge branch 'dev' into tefa/handshake-response-set-connection-id

test/Microsoft.Azure.SignalR.AspNet.Tests/ServiceConnectionTests.cs

@@ -10,6 +10,7 @@
 using Microsoft.AspNet.SignalR;
 using Microsoft.AspNet.SignalR.Transports;
 using Microsoft.Azure.SignalR.Protocol;
+using Microsoft.Azure.SignalR.Tests;
 using Microsoft.Azure.SignalR.Tests.Common;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Primitives;
@@ -34,13 +35,14 @@ public async Task ServiceConnectionDispatchTest()
 
             var clientConnectionManager = new TestClientConnectionManager();
             using var proxy = new TestServiceConnectionProxy(clientConnectionManager, loggerFactory: loggerFactory);
+
             // start the server connection
             await proxy.StartServiceAsync().OrTimeout();
 
             var clientConnection = Guid.NewGuid().ToString("N");
 
             // Application layer sends OpenConnectionMessage
-            var openConnectionMessage = new OpenConnectionMessage(clientConnection, new Claim[0], null, "?transport=webSockets");
+            var openConnectionMessage = new OpenConnectionMessage(clientConnection, [], null, "?transport=webSockets");
             var task = clientConnectionManager.WaitForClientConnectAsync(clientConnection).OrTimeout();
             await proxy.WriteMessageAsync(openConnectionMessage);
             await task;
@@ -80,14 +82,16 @@ public async Task ServiceConnectionDispatchGroupMessagesTest()
             hubConfig.Resolver.Register(typeof(IClientConnectionManagerAspNet), () => ccm);
             DispatcherHelper.PrepareAndGetDispatcher(new TestAppBuilder(), hubConfig, new ServiceOptions { ConnectionString = ConnectionString }, appName, loggerFactory);
             using var proxy = new TestServiceConnectionProxy(ccm, loggerFactory: loggerFactory);
+
             // start the server connection
             await proxy.StartServiceAsync().OrTimeout();
 
             var clientConnection = Guid.NewGuid().ToString("N");
 
             var connectTask = scm.WaitForTransportOutputMessageAsync(typeof(GroupBroadcastDataMessage)).OrTimeout();
+
             // Application layer sends OpenConnectionMessage
-            var openConnectionMessage = new OpenConnectionMessage(clientConnection, new Claim[0], null, $"?transport=webSockets&connectionToken=conn1&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
+            var openConnectionMessage = new OpenConnectionMessage(clientConnection, [], null, $"?transport=webSockets&connectionToken=conn1&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
             await proxy.WriteMessageAsync(openConnectionMessage);
 
             var connectMessage = (await connectTask) as GroupBroadcastDataMessage;
@@ -148,7 +152,7 @@ public async Task ServiceConnectionDispatchGroupMessagesTest()
     [Fact]
     public async Task ServiceConnectionWithErrorConnectHub()
     {
-        using (StartVerifiableLog(out var loggerFactory, LogLevel.Warning, expectedErrors: c=>true, logChecker:
+        using (StartVerifiableLog(out var loggerFactory, LogLevel.Warning, expectedErrors: c => true, logChecker:
             logs =>
             {
                 Assert.Equal(2, logs.Count);
@@ -166,6 +170,7 @@ public async Task ServiceConnectionWithErrorConnectHub()
             hubConfig.Resolver.Register(typeof(IClientConnectionManagerAspNet), () => ccm);
             DispatcherHelper.PrepareAndGetDispatcher(new TestAppBuilder(), hubConfig, new ServiceOptions { ConnectionString = ConnectionString }, appName, loggerFactory);
             using var proxy = new TestServiceConnectionProxy(ccm, loggerFactory: loggerFactory);
+
             // start the server connection
             await proxy.StartServiceAsync().OrTimeout();
 
@@ -174,7 +179,7 @@ public async Task ServiceConnectionWithErrorConnectHub()
             var connectTask = proxy.WaitForOutgoingMessageAsync(clientConnection).OrTimeout();
 
             // Application layer sends OpenConnectionMessage
-            var openConnectionMessage = new OpenConnectionMessage(clientConnection, new Claim[0], null, $"?transport=webSockets&connectionToken=conn1&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
+            var openConnectionMessage = new OpenConnectionMessage(clientConnection, [], null, $"?transport=webSockets&connectionToken=conn1&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
             await proxy.WriteMessageAsync(openConnectionMessage);
 
             // other messages are just ignored because OnConnected failed
@@ -193,7 +198,7 @@ public async Task ServiceConnectionWithErrorConnectHub()
         }
     }
 
-    [Fact]
+    [RetryFact]
     public async Task ServiceConnectionWithErrorDisconnectHub()
     {
         using (StartVerifiableLog(out var loggerFactory, LogLevel.Debug, expectedErrors: c => true, logChecker:
@@ -207,17 +212,19 @@ public async Task ServiceConnectionWithErrorDisconnectHub()
             var ccm = new ClientConnectionManager(hubConfig, loggerFactory);
             hubConfig.Resolver.Register(typeof(IClientConnectionManagerAspNet), () => ccm);
             DispatcherHelper.PrepareAndGetDispatcher(new TestAppBuilder(), hubConfig,
-                new ServiceOptions {ConnectionString = ConnectionString}, appName, loggerFactory);
+                new ServiceOptions { ConnectionString = ConnectionString }, appName, loggerFactory);
             using var proxy = new TestServiceConnectionProxy(ccm, loggerFactory: loggerFactory);
+
             // start the server connection
             await proxy.StartServiceAsync().OrTimeout();
 
             var clientConnection = Guid.NewGuid().ToString("N");
 
             var connectTask = scm.WaitForTransportOutputMessageAsync(typeof(GroupBroadcastDataMessage))
                 .OrTimeout();
+
             // Application layer sends OpenConnectionMessage
-            var openConnectionMessage = new OpenConnectionMessage(clientConnection, new Claim[0], null,
+            var openConnectionMessage = new OpenConnectionMessage(clientConnection, [], null,
                 $"?transport=webSockets&connectionToken=conn1&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
             await proxy.WriteMessageAsync(openConnectionMessage);
 
@@ -266,14 +273,15 @@ public async Task ServiceConnectionDispatchOpenConnectionToUnauthorizedHubTest()
             var ccm = new ClientConnectionManager(hubConfig, loggerFactory);
             hubConfig.Resolver.Register(typeof(IClientConnectionManagerAspNet), () => ccm);
             using var proxy = new TestServiceConnectionProxy(ccm, loggerFactory: loggerFactory);
+
             // start the server connection
             await proxy.StartServiceAsync().OrTimeout();
 
             var connectionId = Guid.NewGuid().ToString("N");
             var connectTask = proxy.WaitForOutgoingMessageAsync(connectionId).OrTimeout();
 
             // Application layer sends OpenConnectionMessage to an authorized hub from anonymous user
-            var openConnectionMessage = new OpenConnectionMessage(connectionId, new Claim[0], null, "?transport=webSockets&connectionData=%5B%7B%22name%22%3A%22authchat%22%7D%5D");
+            var openConnectionMessage = new OpenConnectionMessage(connectionId, [], null, "?transport=webSockets&connectionData=%5B%7B%22name%22%3A%22authchat%22%7D%5D");
             await proxy.WriteMessageAsync(openConnectionMessage);
 
             var message = await connectTask;
@@ -301,6 +309,7 @@ public async Task ServiceConnectionWithNormalClientConnection()
             var ccm = new ClientConnectionManager(hubConfig, loggerFactory);
             hubConfig.Resolver.Register(typeof(IClientConnectionManagerAspNet), () => ccm);
             using var proxy = new TestServiceConnectionProxy(ccm, loggerFactory: loggerFactory);
+
             // start the server connection
             await proxy.StartServiceAsync().OrTimeout();
 
@@ -309,7 +318,7 @@ public async Task ServiceConnectionWithNormalClientConnection()
             var connectTask = proxy.WaitForOutgoingMessageAsync(connectionId).OrTimeout();
 
             // Application layer sends OpenConnectionMessage to an authorized hub from anonymous user
-            var openConnectionMessage = new OpenConnectionMessage(connectionId, new Claim[0], null, "?transport=webSockets&connectionData=%5B%7B%22name%22%3A%22authchat%22%7D%5D");
+            var openConnectionMessage = new OpenConnectionMessage(connectionId, [], null, "?transport=webSockets&connectionData=%5B%7B%22name%22%3A%22authchat%22%7D%5D");
             await proxy.WriteMessageAsync(openConnectionMessage);
 
             var message = await connectTask;
@@ -337,6 +346,7 @@ public async Task ServiceConnectionWithTransportLayerClosedShouldCleanupNormalCl
             DispatcherHelper.PrepareAndGetDispatcher(new TestAppBuilder(), hubConfig,
                 new ServiceOptions { ConnectionString = ConnectionString }, appName, loggerFactory);
             using var proxy = new TestServiceConnectionProxy(ccm, loggerFactory);
+
             // start the server connection
             var connectionTask = proxy.StartAsync();
             await proxy.ConnectionInitializedTask.OrTimeout();
@@ -345,8 +355,9 @@ public async Task ServiceConnectionWithTransportLayerClosedShouldCleanupNormalCl
 
             var connectTask = scm.WaitForTransportOutputMessageAsync(typeof(GroupBroadcastDataMessage))
                 .OrTimeout();
+
             // Application layer sends OpenConnectionMessage
-            var openConnectionMessage = new OpenConnectionMessage(clientConnection, new Claim[0], null,
+            var openConnectionMessage = new OpenConnectionMessage(clientConnection, [], null,
                 $"?transport=webSockets&connectionToken=conn1&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
             await proxy.WriteMessageAsync(openConnectionMessage);
 
@@ -393,6 +404,7 @@ public async Task ServiceConnectionWithTransportLayerClosedShouldCleanupEndlessC
             DispatcherHelper.PrepareAndGetDispatcher(new TestAppBuilder(), hubConfig,
                 new ServiceOptions { ConnectionString = ConnectionString }, appName, loggerFactory);
             using var proxy = new TestServiceConnectionProxy(ccm, loggerFactory);
+
             // start the server connection
             var connectionTask = proxy.StartAsync();
             await proxy.ConnectionInitializedTask.OrTimeout();
@@ -401,8 +413,9 @@ public async Task ServiceConnectionWithTransportLayerClosedShouldCleanupEndlessC
 
             var connectTask = scm.WaitForTransportOutputMessageAsync(typeof(GroupBroadcastDataMessage))
                 .OrTimeout();
+
             // Application layer sends OpenConnectionMessage
-            var openConnectionMessage = new OpenConnectionMessage(clientConnection, new Claim[0], null,
+            var openConnectionMessage = new OpenConnectionMessage(clientConnection, [], null,
                 $"?transport=webSockets&connectionToken=conn1&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
             await proxy.WriteMessageAsync(openConnectionMessage);
 
@@ -442,6 +455,7 @@ public async Task ServiceConnectionWithTransportLayerClosedShouldCleanupEndlessI
             DispatcherHelper.PrepareAndGetDispatcher(new TestAppBuilder(), hubConfig,
                 new ServiceOptions { ConnectionString = ConnectionString }, appName, loggerFactory);
             using var proxy = new TestServiceConnectionProxy(ccm, loggerFactory);
+
             // start the server connection
             var connectionTask = proxy.StartAsync();
             await proxy.ConnectionInitializedTask.OrTimeout();
@@ -450,8 +464,9 @@ public async Task ServiceConnectionWithTransportLayerClosedShouldCleanupEndlessI
 
             var connectTask = scm.WaitForTransportOutputMessageAsync(typeof(GroupBroadcastDataMessage))
                 .OrTimeout();
+
             // Application layer sends OpenConnectionMessage
-            var openConnectionMessage = new OpenConnectionMessage(clientConnection, new Claim[0], null,
+            var openConnectionMessage = new OpenConnectionMessage(clientConnection, [], null,
                 $"?transport=webSockets&connectionToken=conn1&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
             await proxy.WriteMessageAsync(openConnectionMessage);
 
@@ -500,6 +515,7 @@ public async Task ServiceConnectionWithOfflinePingWillTriggerDisconnectClients()
             hubConfig.Resolver.Register(typeof(IClientConnectionManagerAspNet), () => ccm);
             DispatcherHelper.PrepareAndGetDispatcher(new TestAppBuilder(), hubConfig, new ServiceOptions { ConnectionString = ConnectionString }, appName, loggerFactory);
             using var proxy = new TestServiceConnectionProxy(ccm, loggerFactory: loggerFactory);
+
             // prepare 2 clients with different instancesId connected
             var instanceId1 = Guid.NewGuid().ToString();
             var connectionId1 = Guid.NewGuid().ToString("N");
@@ -513,7 +529,7 @@ public async Task ServiceConnectionWithOfflinePingWillTriggerDisconnectClients()
 
             // Application layer sends OpenConnectionMessage for client1
             var connectTask = scm.WaitForTransportOutputMessageAsync(typeof(GroupBroadcastDataMessage)).OrTimeout();
-            var openConnectionMessage = new OpenConnectionMessage(connectionId1, new Claim[0], header1, $"?transport=webSockets&connectionToken=conn1&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
+            var openConnectionMessage = new OpenConnectionMessage(connectionId1, [], header1, $"?transport=webSockets&connectionToken=conn1&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
             await proxy.WriteMessageAsync(openConnectionMessage);
 
             // client1 is connected
@@ -528,7 +544,7 @@ public async Task ServiceConnectionWithOfflinePingWillTriggerDisconnectClients()
 
             // Application layer sends OpenConnectionMessage for client2
             connectTask = scm.WaitForTransportOutputMessageAsync(typeof(GroupBroadcastDataMessage)).OrTimeout();
-            openConnectionMessage = new OpenConnectionMessage(connectionId2, new Claim[0], header2, $"?transport=webSockets&connectionToken=conn2&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
+            openConnectionMessage = new OpenConnectionMessage(connectionId2, [], header2, $"?transport=webSockets&connectionToken=conn2&connectionData=%5B%7B%22name%22%3A%22{hub}%22%7D%5D");
             await proxy.WriteMessageAsync(openConnectionMessage);
 
             // client2 is connected
@@ -544,10 +560,10 @@ public async Task ServiceConnectionWithOfflinePingWillTriggerDisconnectClients()
             connectTask = scm.WaitForTransportOutputMessageAsync(typeof(GroupBroadcastDataMessage)).OrTimeout();
             await proxy.WriteMessageAsync(new PingMessage()
             {
-                Messages = new[] { "offline", instanceId1 }
+                Messages = ["offline", instanceId1]
             });
 
-            // Validate client1 disconnect 
+            // Validate client1 disconnect
             connectMessage = (await connectTask) as GroupBroadcastDataMessage;
             Assert.NotNull(connectMessage);
             Assert.Equal($"hg-{hub}.note", connectMessage.GroupName);
@@ -564,7 +580,9 @@ await proxy.WriteMessageAsync(new PingMessage()
     private sealed class HubResponseItem
     {
         public string H { get; set; }
+
         public string M { get; set; }
+
         public List<string> A { get; set; }
     }
 }

test/Microsoft.Azure.SignalR.Tests/ServiceConnectionContainerBaseTests.cs

@@ -204,14 +204,14 @@ internal async Task TestOffline(GracefulShutdownMode mode)
         };
         using var container = new TestServiceConnectionContainer(connections, factory: new SimpleTestServiceConnectionFactory());
 
-        foreach (SimpleTestServiceConnection c in connections)
+        foreach (var c in connections.Cast<SimpleTestServiceConnection>())
         {
             Assert.False(c.ConnectionOfflineTask.IsCompleted);
         }
 
         await container.OfflineAsync(mode, default);
 
-        foreach (SimpleTestServiceConnection c in connections)
+        foreach (var c in connections.Cast<SimpleTestServiceConnection>())
         {
             Assert.True(c.ConnectionOfflineTask.IsCompleted);
         }

test/Microsoft.Azure.SignalR.Tests/ServiceConnectionContainerTests.cs

@@ -3,19 +3,21 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Security.Claims;
 using System.Threading.Tasks;
 
 using Microsoft.Azure.SignalR.Protocol;
 using Microsoft.Azure.SignalR.Tests.Common;
+using Microsoft.Extensions.Logging.Abstractions;
 using Xunit;
 
 namespace Microsoft.Azure.SignalR.Tests;
 
 public class ServiceConnectionContainerTests
 {
     [Fact]
-    public async void TestCloseAsync()
+    public async Task TestCloseAsync()
     {
         var conn = new TestServiceConnectionForCloseAsync();
         var hub = new TestHubServiceEndpoint();
@@ -40,7 +42,7 @@ public void TestCloseAsyncWithoutStartAsync()
     }
 
     [Fact]
-    public async void TestCloseAsyncWithExceptionAndNoFinAck()
+    public async Task TestCloseAsyncWithExceptionAndNoFinAck()
     {
         var conn = new TestServiceConnectionForCloseAsync();
         var hub = new TestHubServiceEndpoint();
@@ -55,15 +57,161 @@ public async void TestCloseAsyncWithExceptionAndNoFinAck()
         // await AssertTask(container.CloseClientConnectionForTest(conn), TimeSpan.FromSeconds(5));
     }
 
-    private static async Task MockServiceAsync(TestServiceConnectionForCloseAsync conn)
+    [Fact]
+    public async Task TestServiceConnectionOffline()
+    {
+        var factory = new TestServiceConnectionFactory();
+        var hubServiceEndpoint = new HubServiceEndpoint("foo", null, new TestServiceEndpoint());
+
+        var container = new StrongServiceConnectionContainer(factory, 3, 3, hubServiceEndpoint, NullLogger.Instance);
+
+        Assert.True(factory.CreatedConnections.TryGetValue(hubServiceEndpoint, out var conns));
+        var connections = conns.Select(x => (TestServiceConnection)x).ToArray();
+
+        foreach (var connection in connections)
+        {
+            connection.SetStatus(ServiceConnectionStatus.Connected);
+        }
+
+        // write 100 messages.
+        for (var i = 0; i < 100; i++)
+        {
+            var message = new ConnectionDataMessage(i.ToString(), new byte[12]);
+            await container.WriteAsync(message);
+        }
+
+        var messageCount = new Dictionary<string, int>();
+        foreach (var connection in connections)
+        {
+            Assert.NotEmpty(connection.ReceivedMessages);
+            messageCount.TryAdd(connection.ConnectionId, connection.ReceivedMessages.Count);
+        }
+
+        connections[0].SetStatus(ServiceConnectionStatus.Disconnected);
+
+        // write 100 more messages.
+        for (var i = 0; i < 100; i++)
+        {
+            var message = new ConnectionDataMessage(i.ToString(), new byte[12]);
+            await container.WriteAsync(message);
+        }
+
+        var index = 0;
+        foreach (var connection in connections)
+        {
+            if (index == 0)
+            {
+                Assert.Equal(messageCount[connection.ConnectionId], connection.ReceivedMessages.Count);
+            }
+            else
+            {
+                Assert.NotEqual(messageCount[connection.ConnectionId], connection.ReceivedMessages.Count);
+            }
+            index++;
+        }
+    }
+
+    [Fact]
+    public async Task TestServiceConnectionStickyWrites()
     {
-        IServiceProtocol proto = new ServiceProtocol();
+        var factory = new TestServiceConnectionFactory();
+        var hubServiceEndpoint = new HubServiceEndpoint("foo", null, new TestServiceEndpoint());
+
+        var container = new StrongServiceConnectionContainer(factory, 30, 30, hubServiceEndpoint, NullLogger.Instance);
+
+        Assert.True(factory.CreatedConnections.TryGetValue(hubServiceEndpoint, out var conns));
+        var connections = conns.Select(x => (TestServiceConnection)x);
+
+        foreach (var connection in connections)
+        {
+            connection.SetStatus(ServiceConnectionStatus.Connected);
+        }
+
+        // write 100000 messages.
+        for (var i = 0; i < 100000; i++)
+        {
+            var message = new ConnectionDataMessage(i.ToString(), new byte[12]);
+            await container.WriteAsync(message);
+        }
 
+        var messageCount = new Dictionary<string, int>();
+        foreach (var connection in connections)
+        {
+            Assert.NotEmpty(connection.ReceivedMessages);
+            messageCount.TryAdd(connection.ConnectionId, connection.ReceivedMessages.Count);
+        }
+
+        // write 100000 messages with the same connectionIds should double the message count for each service connection
+        for (var i = 0; i < 100000; i++)
+        {
+            var message = new ConnectionDataMessage(i.ToString(), new byte[12]);
+            await container.WriteAsync(message);
+        }
+
+        foreach (var connection in connections)
+        {
+            Assert.Equal(messageCount[connection.ConnectionId] * 2, connection.ReceivedMessages.Count);
+        }
+
+        // Offline half of the connections
+        foreach (var connection in connections.Skip(15))
+        {
+            connection.SetStatus(ServiceConnectionStatus.Disconnected);
+        }
+
+        var sub = connections.SkipLast(15).Sum(s => s.ReceivedMessages.Count) - connections.Skip(15).Sum(s => s.ReceivedMessages.Count);
+
+        // write 100000 messages with the same connectionIds does not throw
+        for (var i = 0; i < 100000; i++)
+        {
+            var message = new ConnectionDataMessage(i.ToString(), new byte[12]);
+            await container.WriteAsync(message);
+        }
+
+        // messages are all going through the connected connections
+        var disconnected = connections.Skip(15).Sum(s => s.ReceivedMessages.Count);
+        var connected = connections.SkipLast(15).Sum(s => s.ReceivedMessages.Count);
+        Assert.Equal(100000 + sub, connected - disconnected);
+    }
+
+    [Fact]
+    public async Task TestServiceConnectionStickyWritesWithScope()
+    {
+        // with scope enabled, the messages always go through the first picked connection
+        using var _ = new ClientConnectionScope();
+        var factory = new TestServiceConnectionFactory();
+        var hubServiceEndpoint = new HubServiceEndpoint("foo", null, new TestServiceEndpoint());
+
+        var container = new StrongServiceConnectionContainer(factory, 30, 30, hubServiceEndpoint, NullLogger.Instance);
+
+        Assert.True(factory.CreatedConnections.TryGetValue(hubServiceEndpoint, out var conns));
+        var connections = conns.Select(x => (TestServiceConnection)x);
+
+        foreach (var connection in connections)
+        {
+            connection.SetStatus(ServiceConnectionStatus.Connected);
+        }
+
+        // write 100000 messages.
+        for (var i = 0; i < 100000; i++)
+        {
+            var message = new ConnectionDataMessage(i.ToString(), new byte[12]);
+            await container.WriteAsync(message);
+        }
+
+        var selected = connections.Where(s => s.ReceivedMessages.Count > 0).ToArray();
+        Assert.Single(selected);
+
+        Assert.Equal(100000, selected[0].ReceivedMessages.Count);
+    }
+
+    private static async Task MockServiceAsync(TestServiceConnectionForCloseAsync conn)
+    {
         await conn.ConnectionCreated;
 
         // open 2 new connections (to create 2 new outgoing tasks
-        proto.WriteMessage(new OpenConnectionMessage(Guid.NewGuid().ToString(), Array.Empty<Claim>()), conn.Application.Output);
-        proto.WriteMessage(new OpenConnectionMessage(Guid.NewGuid().ToString(), Array.Empty<Claim>()), conn.Application.Output);
+        new ServiceProtocol().WriteMessage(new OpenConnectionMessage(Guid.NewGuid().ToString(), Array.Empty<Claim>()), conn.Application.Output);
+        new ServiceProtocol().WriteMessage(new OpenConnectionMessage(Guid.NewGuid().ToString(), Array.Empty<Claim>()), conn.Application.Output);
         await conn.Application.Output.FlushAsync();
 
         while (true)
@@ -74,12 +222,12 @@ private static async Task MockServiceAsync(TestServiceConnectionForCloseAsync co
             try
             {
                 // write back a FinAck after receiving a Fin
-                if (proto.TryParseMessage(ref buffer, out var message))
+                if (new ServiceProtocol().TryParseMessage(ref buffer, out var message))
                 {
                     if (RuntimeServicePingMessage.IsFin(message))
                     {
                         var pong = RuntimeServicePingMessage.GetFinAckPingMessage();
-                        proto.WriteMessage(pong, conn.Application.Output);
+                        new ServiceProtocol().WriteMessage(pong, conn.Application.Output);
                         await conn.Application.Output.FlushAsync();
                         break;
                     }
@@ -102,17 +250,15 @@ private static PingMessage BuildPingMessage(string key, string val)
 
     private static async Task MockServiceAsyncWithException(TestServiceConnectionForCloseAsync conn)
     {
-        IServiceProtocol proto = new ServiceProtocol();
-
         await conn.ConnectionCreated;
 
         // open 2 new connections (to create 2 new outgoing tasks
-        proto.WriteMessage(new OpenConnectionMessage(Guid.NewGuid().ToString(), Array.Empty<Claim>()), conn.Application.Output);
-        proto.WriteMessage(new OpenConnectionMessage(Guid.NewGuid().ToString(), Array.Empty<Claim>()), conn.Application.Output);
+        new ServiceProtocol().WriteMessage(new OpenConnectionMessage(Guid.NewGuid().ToString(), Array.Empty<Claim>()), conn.Application.Output);
+        new ServiceProtocol().WriteMessage(new OpenConnectionMessage(Guid.NewGuid().ToString(), Array.Empty<Claim>()), conn.Application.Output);
         await conn.Application.Output.FlushAsync();
 
         await Task.Delay(TimeSpan.FromSeconds(1));
-        proto.WriteMessage(BuildPingMessage("_exception", "1"), conn.Application.Output);
+        new ServiceProtocol().WriteMessage(BuildPingMessage("_exception", "1"), conn.Application.Output);
         await conn.Application.Output.FlushAsync();
     }
 }