Merge pull request #239 from BalassaMarton/messagerouter-connectionfa…

…ctory

Messaging: Added IConnectionFactory for correctly implementing disposable pattern

src/messaging/dotnet/Directory.Build.props

@@ -4,5 +4,6 @@
 	<PropertyGroup>
 		<ImplicitUsings>enable</ImplicitUsings>
 		<Nullable>enable</Nullable>
+		<NoWarn>CS1066;CS1591</NoWarn>
 	</PropertyGroup>
 </Project>

src/messaging/dotnet/src/Client/Client/Abstractions/IConnection.cs

@@ -14,8 +14,6 @@
 
 namespace MorganStanley.ComposeUI.Messaging.Client.Abstractions;
 
-// TODO: Add IConnectionFactory abstraction to adhere to DI best practices
-
 /// <summary>
 /// Represents a connection that can communicate with the server.
 /// </summary>

src/messaging/dotnet/src/Client/Client/Abstractions/IConnectionFactory.cs

@@ -0,0 +1,25 @@
+// Morgan Stanley makes this available to you under the Apache License,
+// Version 2.0 (the "License"). You may obtain a copy of the License at
+// 
+//      http://www.apache.org/licenses/LICENSE-2.0.
+// 
+// See the NOTICE file distributed with this work for additional information
+// regarding copyright ownership. Unless required by applicable law or agreed
+// to in writing, software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+// or implied. See the License for the specific language governing permissions
+// and limitations under the License.
+
+namespace MorganStanley.ComposeUI.Messaging.Client.Abstractions;
+
+/// <summary>
+/// Creates instances of <see cref="IConnection"/>
+/// </summary>
+public interface IConnectionFactory
+{
+    /// <summary>
+    /// Creates a new <see cref="IConnection"/> instance.
+    /// </summary>
+    /// <returns></returns>
+    IConnection CreateConnection();
+}

src/messaging/dotnet/src/Client/Client/MessageRouterClient.cs

@@ -25,11 +25,11 @@ namespace MorganStanley.ComposeUI.Messaging.Client;
 internal sealed class MessageRouterClient : IMessageRouter
 {
     public MessageRouterClient(
-        IConnection connection,
+        IConnectionFactory connectionFactory,
         MessageRouterOptions options,
         ILogger<MessageRouterClient>? logger = null)
     {
-        _connection = connection;
+        _connection = connectionFactory.CreateConnection();
         _options = options;
         _logger = logger ?? NullLogger<MessageRouterClient>.Instance;
     }

src/messaging/dotnet/src/Client/Client/WebSocket/MessageRouterBuilderWebSocketExtensions.cs

@@ -31,7 +31,7 @@ public static MessageRouterBuilder UseWebSocket(
         MessageRouterWebSocketOptions options)
     {
         builder.ServiceCollection.AddSingleton<IOptions<MessageRouterWebSocketOptions>>(options);
-        builder.ServiceCollection.AddTransient<IConnection, WebSocketConnection>();
+        builder.ServiceCollection.AddSingleton<IConnectionFactory, WebSocketConnectionFactory>();
         return builder;
     }
 }

src/messaging/dotnet/src/Client/Client/WebSocket/WebSocketConnectionFactory.cs

@@ -0,0 +1,31 @@
+// Morgan Stanley makes this available to you under the Apache License,
+// Version 2.0 (the "License"). You may obtain a copy of the License at
+// 
+//      http://www.apache.org/licenses/LICENSE-2.0.
+// 
+// See the NOTICE file distributed with this work for additional information
+// regarding copyright ownership. Unless required by applicable law or agreed
+// to in writing, software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+// or implied. See the License for the specific language governing permissions
+// and limitations under the License.
+
+using Microsoft.Extensions.DependencyInjection;
+using MorganStanley.ComposeUI.Messaging.Client.Abstractions;
+
+namespace MorganStanley.ComposeUI.Messaging.Client.WebSocket;
+
+internal class WebSocketConnectionFactory : IConnectionFactory
+{
+    public WebSocketConnectionFactory(IServiceProvider serviceProvider)
+    {
+        _serviceProvider = serviceProvider;
+    }
+
+    public IConnection CreateConnection()
+    {
+        return ActivatorUtilities.CreateInstance<WebSocketConnection>(_serviceProvider);
+    }
+
+    private readonly IServiceProvider _serviceProvider;
+}

src/messaging/dotnet/src/Core/Protocol/Json/MessageBufferConverter.cs

@@ -28,13 +28,15 @@ internal class MessageBufferConverter : JsonConverter<MessageBuffer>
                 return null;
 
             case JsonTokenType.String:
-                {
-                    var length = reader.HasValueSequence ? checked((int)reader.ValueSequence.Length) : reader.ValueSpan.Length;
-                    var buffer = MessageBuffer.GetBuffer(length);
-                    length = reader.CopyString(buffer);
+            {
+                var length = reader.HasValueSequence
+                    ? checked((int) reader.ValueSequence.Length)
+                    : reader.ValueSpan.Length;
+                var buffer = MessageBuffer.GetBuffer(length);
+                length = reader.CopyString(buffer);
 
-                    return new MessageBuffer(buffer, length);
-                }
+                return new MessageBuffer(buffer, length);
+            }
         }
 
         throw new JsonException();
@@ -44,4 +46,4 @@ public override void Write(Utf8JsonWriter writer, MessageBuffer value, JsonSeria
     {
         writer.WriteStringValue(value.GetSpan());
     }
-}
+}

src/messaging/dotnet/src/Host/Client/Internal/InProcessConnection.cs

@@ -35,14 +35,19 @@ ValueTask<Message> IClientConnection.ReceiveAsync(CancellationToken cancellation
         return _clientToServer.Reader.ReadAsync(cancellationToken);
     }
 
-    public ValueTask DisposeAsync()
+    public ValueTask CloseAsync()
     {
         _clientToServer.Writer.TryComplete();
         _serverToClient.Writer.TryComplete();
-        
+
         return ValueTask.CompletedTask;
     }
 
+    public ValueTask DisposeAsync()
+    {
+        return CloseAsync();
+    }
+
     ValueTask IConnection.ConnectAsync(CancellationToken cancellationToken = default)
     {
         return _server.ClientConnected(this);

src/messaging/dotnet/src/Host/Client/Internal/InProcessConnectionFactory.cs

@@ -0,0 +1,31 @@
+// Morgan Stanley makes this available to you under the Apache License,
+// Version 2.0 (the "License"). You may obtain a copy of the License at
+// 
+//      http://www.apache.org/licenses/LICENSE-2.0.
+// 
+// See the NOTICE file distributed with this work for additional information
+// regarding copyright ownership. Unless required by applicable law or agreed
+// to in writing, software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+// or implied. See the License for the specific language governing permissions
+// and limitations under the License.
+
+using MorganStanley.ComposeUI.Messaging.Client.Abstractions;
+using MorganStanley.ComposeUI.Messaging.Server;
+
+namespace MorganStanley.ComposeUI.Messaging.Client.Internal;
+
+internal sealed class InProcessConnectionFactory : IConnectionFactory
+{
+    private readonly IMessageRouterServer _server;
+
+    public InProcessConnectionFactory(IMessageRouterServer server)
+    {
+        _server = server;
+    }
+
+    public IConnection CreateConnection()
+    {
+        return new InProcessConnection(_server);
+    }
+}

src/messaging/dotnet/src/Host/Client/Internal/MessageRouterBuilderInProcessExtensions.cs

@@ -28,7 +28,7 @@ public static class MessageRouterBuilderInProcessExtensions
     /// <returns></returns>
     public static MessageRouterBuilder UseServer(this MessageRouterBuilder builder)
     {
-        builder.ServiceCollection.AddTransient<IConnection, InProcessConnection>();
+        builder.ServiceCollection.AddSingleton<IConnectionFactory, InProcessConnectionFactory>();
 
         return builder;
     }

src/messaging/dotnet/src/Server/Server/Abstractions/IClientConnection.cs

@@ -17,7 +17,7 @@ namespace MorganStanley.ComposeUI.Messaging.Server.Abstractions;
 /// <summary>
 /// Abstraction of a client connected to the Message Router server.
 /// </summary>
-public interface IClientConnection : IAsyncDisposable
+public interface IClientConnection
 {
     /// <summary>
     /// Sends a message to the client.
@@ -37,4 +37,10 @@ public interface IClientConnection : IAsyncDisposable
     /// <exception cref="MessageRouterException">With name <see cref="MessageRouterErrors.ConnectionClosed"/> if the connection was closed by either party while sending the request</exception>
     /// <exception cref="MessageRouterException">With name <see cref="MessageRouterErrors.ConnectionAborted"/> if the connection was closed due to an unexpected error</exception>
     ValueTask<Message> ReceiveAsync(CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Invoked by the server to notify the object of a disconnection.
+    /// </summary>
+    /// <returns></returns>
+    ValueTask CloseAsync();
 }

src/messaging/dotnet/src/Server/Server/MessageRouterServer.cs

@@ -36,8 +36,7 @@ public async ValueTask DisposeAsync()
     {
         _stopTokenSource.Cancel();
 
-        // TODO: Don't dispose objects that were created by someone else. Signal disconnection using a dedicated method.
-        await Task.WhenAll(_clients.Values.Select(client => client.Connection.DisposeAsync().AsTask()));
+        await Task.WhenAll(_clients.Values.Select(client => client.Connection.CloseAsync().AsTask()));
     }
 
     public ValueTask ClientConnected(IClientConnection connection)

src/messaging/dotnet/src/Server/Server/WebSocket/WebSocketConnection.cs

@@ -82,11 +82,16 @@ public async ValueTask<Message> ReceiveAsync(CancellationToken cancellationToken
         }
     }
 
-    public ValueTask DisposeAsync()
+    public ValueTask CloseAsync()
     {
         _stopTokenSource.Cancel();
 
-        return default;
+        return ValueTask.CompletedTask;
+    }
+
+    public ValueTask DisposeAsync()
+    {
+        return CloseAsync();
     }
 
     public async Task HandleWebSocketRequest(

src/messaging/dotnet/test/Client.Tests/Client/MessageRouterClient.Tests.cs

@@ -11,6 +11,7 @@
 // and limitations under the License.
 
 using System.Linq.Expressions;
+using MorganStanley.ComposeUI.Messaging.Client.Abstractions;
 using MorganStanley.ComposeUI.Messaging.Protocol;
 using MorganStanley.ComposeUI.Messaging.Protocol.Messages;
 using MorganStanley.ComposeUI.Messaging.TestUtils;
@@ -682,6 +683,9 @@ public Task DisposeAsync()
 
     private MessageRouterClient CreateMessageRouter(MessageRouterOptions? options = null)
     {
-        return new MessageRouterClient(_connectionMock.Object, options ?? DefaultOptions);
+        var connectionFactory = new Mock<IConnectionFactory>();
+        connectionFactory.Setup(_ => _.CreateConnection()).Returns(_connectionMock.Object);
+
+        return new MessageRouterClient(connectionFactory.Object, options ?? DefaultOptions);
     }
 }

src/messaging/dotnet/test/Server.Tests/Server/MessageRouterServer.Tests.cs

@@ -13,6 +13,7 @@
 using System.Linq.Expressions;
 using MorganStanley.ComposeUI.Messaging.Protocol;
 using MorganStanley.ComposeUI.Messaging.Protocol.Messages;
+using MorganStanley.ComposeUI.Messaging.Server.Abstractions;
 using MorganStanley.ComposeUI.Messaging.TestUtils;
 using TaskExtensions = MorganStanley.ComposeUI.Testing.TaskExtensions;
 
@@ -347,14 +348,32 @@ await client.SendToServer(
             Times.Once);
     }
 
+    [Fact]
+    public async Task When_disposed_it_calls_CloseAsync_on_active_connections()
+    {
+        var server = CreateServer();
+        var connection = new Mock<IClientConnection>();
+
+        connection.SetupSequence(_ => _.ReceiveAsync(It.IsAny<CancellationToken>()))
+            .Returns(new ValueTask<Message>(
+                new ConnectRequest()))
+            .Returns(new ValueTask<Message>(
+                Task.Delay(1000).ContinueWith(_ => (Message)new PublishMessage {Topic = "dummy"})));
+
+        await server.ClientConnected(connection.Object);
+        await TaskExtensions.WaitForBackgroundTasksAsync();
+        await server.DisposeAsync();
+
+        connection.Verify(_ => _.CloseAsync(), Times.Once);
+    }
+
     private MessageRouterServer CreateServer(IAccessTokenValidator? accessTokenValidator = null) =>
         new MessageRouterServer(new MessageRouterServerDependencies(accessTokenValidator));
 
     private MockClientConnection CreateClient() => new MockClientConnection();
 
     private async Task<MockClientConnection> CreateAndConnectClient(MessageRouterServer server)
     {
-        var connectResponseReceived = new TaskCompletionSource<ConnectResponse>();
         var client = CreateClient();
         await server.ClientConnected(client.Object);
         await client.Connect();

src/messaging/dotnet/test/Server.Tests/TestUtils/MockClientConnection.cs

@@ -26,7 +26,7 @@ public MockClientConnection()
         Setup(_ => _.ReceiveAsync(It.IsAny<CancellationToken>()))
             .Returns((CancellationToken ct) => _sendChannel.Reader.ReadAsync(ct));
 
-        Setup(_ => _.DisposeAsync())
+        Setup(_ => _.CloseAsync())
             .Callback(
                 () => { _sendChannel.Writer.TryComplete(); });
     }