[SignalR] Pass new App to Transport Pipe to App layer

src/Servers/Connections.Abstractions/src/Features/IStatefulReconnectFeature.cs

@@ -0,0 +1,34 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System;
+using System.Collections.Generic;
+using System.IO.Pipelines;
+using System.Linq;
+#if NET8_0_OR_GREATER
+using System.Runtime.Versioning;
+#endif
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Microsoft.AspNetCore.Connections.Abstractions;
+
+/// <summary>
+/// Provides access to connection reconnect operations.
+/// </summary>
+/// <remarks>This feature is experimental.</remarks>
+#if NET8_0_OR_GREATER
+[RequiresPreviewFeatures("IStatefulReconnectFeature is a preview interface")]
+#endif
+public interface IStatefulReconnectFeature
+{
+    /// <summary>
+    /// Called when a connection reconnects. The new <see cref="PipeWriter"/> that application code should write to is passed in.
+    /// </summary>
+    public void OnReconnected(Func<PipeWriter, Task> notifyOnReconnect);
+
+    /// <summary>
+    /// Allows disabling the reconnect feature so a reconnecting connection will not be allowed anymore.
+    /// </summary>
+    void DisableReconnect();
+}

src/Servers/Connections.Abstractions/src/PublicAPI/net462/PublicAPI.Unshipped.txt

@@ -1,7 +1,7 @@
 #nullable enable
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.get -> System.Action!
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.set -> void
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature.DisableReconnect() -> void
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature.OnReconnected(System.Func<System.IO.Pipelines.PipeWriter!, System.Threading.Tasks.Task!>! notifyOnReconnect) -> void
 Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature
 Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature.Tags.get -> System.Collections.Generic.ICollection<System.Collections.Generic.KeyValuePair<string!, object?>>!
 Microsoft.AspNetCore.Connections.Features.IConnectionNamedPipeFeature

src/Servers/Connections.Abstractions/src/PublicAPI/net8.0/PublicAPI.Unshipped.txt

@@ -1,7 +1,7 @@
 #nullable enable
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.get -> System.Action!
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.set -> void
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature.DisableReconnect() -> void
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature.OnReconnected(System.Func<System.IO.Pipelines.PipeWriter!, System.Threading.Tasks.Task!>! notifyOnReconnect) -> void
 Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature
 Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature.Tags.get -> System.Collections.Generic.ICollection<System.Collections.Generic.KeyValuePair<string!, object?>>!
 Microsoft.AspNetCore.Connections.Features.IConnectionNamedPipeFeature

src/Servers/Connections.Abstractions/src/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt

@@ -1,7 +1,7 @@
 #nullable enable
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.get -> System.Action!
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.set -> void
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature.DisableReconnect() -> void
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature.OnReconnected(System.Func<System.IO.Pipelines.PipeWriter!, System.Threading.Tasks.Task!>! notifyOnReconnect) -> void
 Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature
 Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature.Tags.get -> System.Collections.Generic.ICollection<System.Collections.Generic.KeyValuePair<string!, object?>>!
 Microsoft.AspNetCore.Connections.Features.IConnectionNamedPipeFeature

src/Servers/Connections.Abstractions/src/PublicAPI/netstandard2.1/PublicAPI.Unshipped.txt

@@ -1,7 +1,7 @@
 #nullable enable
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.get -> System.Action!
-Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.set -> void
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature.DisableReconnect() -> void
+Microsoft.AspNetCore.Connections.Abstractions.IStatefulReconnectFeature.OnReconnected(System.Func<System.IO.Pipelines.PipeWriter!, System.Threading.Tasks.Task!>! notifyOnReconnect) -> void
 Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature
 Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature.Tags.get -> System.Collections.Generic.ICollection<System.Collections.Generic.KeyValuePair<string!, object?>>!
 Microsoft.AspNetCore.Connections.Features.IConnectionNamedPipeFeature

src/SignalR/clients/csharp/Client.Core/src/HubConnection.cs

@@ -579,6 +579,13 @@ private async Task StopAsyncCore(bool disposing)
                         TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnFaulted,
                         TaskScheduler.Default);
                 }
+
+#pragma warning disable CA2252 // This API requires opting into preview features
+                if (connectionState.Connection.Features.Get<IStatefulReconnectFeature>() is IStatefulReconnectFeature feature)
+                {
+                    feature.DisableReconnect();
+                }
+#pragma warning restore CA2252 // This API requires opting into preview features
             }
             else
             {
@@ -1088,6 +1095,14 @@ private async Task SendWithLock(ConnectionState expectedConnectionState, HubMess
                 {
                     Log.ReceivedCloseWithError(_logger, close.Error);
                 }
+
+#pragma warning disable CA2252 // This API requires opting into preview features
+                if (connectionState.Connection.Features.Get<IStatefulReconnectFeature>() is IStatefulReconnectFeature feature)
+                {
+                    feature.DisableReconnect();
+                }
+#pragma warning restore CA2252 // This API requires opting into preview features
+
                 return close;
             case PingMessage _:
                 Log.ReceivedPing(_logger);
@@ -1900,14 +1915,16 @@ public ConnectionState(ConnectionContext connection, HubConnection hubConnection
             _logger = _hubConnection._logger;
             _hasInherentKeepAlive = connection.Features.Get<IConnectionInherentKeepAliveFeature>()?.HasInherentKeepAlive ?? false;
 
-            if (Connection.Features.Get<IReconnectFeature>() is IReconnectFeature feature)
+#pragma warning disable CA2252 // This API requires opting into preview features
+            if (Connection.Features.Get<IStatefulReconnectFeature>() is IStatefulReconnectFeature feature)
             {
                 _messageBuffer = new MessageBuffer(connection, hubConnection._protocol,
                     _hubConnection._serviceProvider.GetService<IOptions<HubConnectionOptions>>()?.Value.StatefulReconnectBufferSize
                         ?? DefaultStatefulReconnectBufferSize);
 
-                feature.NotifyOnReconnect = _messageBuffer.Resend;
+                feature.OnReconnected(_messageBuffer.ResendAsync);
             }
+#pragma warning restore CA2252 // This API requires opting into preview features
         }
 
         public string GetNextId() => (++_nextInvocationId).ToString(CultureInfo.InvariantCulture);

src/SignalR/clients/csharp/Client/test/UnitTests/HubConnectionTests.cs

@@ -7,6 +7,7 @@
 using System.Net.WebSockets;
 using System.Threading.Channels;
 using Microsoft.AspNetCore.Connections;
+using Microsoft.AspNetCore.Connections.Abstractions;
 using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Http.Connections.Client;
 using Microsoft.AspNetCore.SignalR.Protocol;
@@ -878,6 +879,70 @@ public async Task HubConnectionIsMockable()
         mockConnection.Verify(c => c.StopAsync(It.IsAny<CancellationToken>()), Times.Once);
     }
 
+    [Fact]
+    public async Task DisableReconnectCalledWhenCloseMessageReceived()
+    {
+        var builder = new HubConnectionBuilder().WithUrl("http://example.com");
+        var innerConnection = new TestConnection();
+        var reconnectFeature = new TestReconnectFeature();
+#pragma warning disable CA2252 // This API requires opting into preview features
+        innerConnection.Features.Set<IStatefulReconnectFeature>(reconnectFeature);
+#pragma warning restore CA2252 // This API requires opting into preview features
+
+        var delegateConnectionFactory = new DelegateConnectionFactory(
+            endPoint => innerConnection.StartAsync());
+        builder.Services.AddSingleton<IConnectionFactory>(delegateConnectionFactory);
+
+        var hubConnection = builder.Build();
+        var closedEventTcs = new TaskCompletionSource<Exception>();
+        hubConnection.Closed += e =>
+        {
+            closedEventTcs.SetResult(e);
+            return Task.CompletedTask;
+        };
+
+        await hubConnection.StartAsync().DefaultTimeout();
+
+        await innerConnection.ReceiveJsonMessage(new { type = HubProtocolConstants.CloseMessageType });
+
+        var exception = await closedEventTcs.Task.DefaultTimeout();
+        Assert.Null(exception);
+
+        await reconnectFeature.DisableReconnectCalled.DefaultTimeout();
+    }
+
+    [Fact]
+    public async Task DisableReconnectCalledWhenSendingCloseMessage()
+    {
+        var builder = new HubConnectionBuilder().WithUrl("http://example.com");
+        var innerConnection = new TestConnection();
+        var reconnectFeature = new TestReconnectFeature();
+#pragma warning disable CA2252 // This API requires opting into preview features
+        innerConnection.Features.Set<IStatefulReconnectFeature>(reconnectFeature);
+#pragma warning restore CA2252 // This API requires opting into preview features
+
+        var delegateConnectionFactory = new DelegateConnectionFactory(
+            endPoint => innerConnection.StartAsync());
+        builder.Services.AddSingleton<IConnectionFactory>(delegateConnectionFactory);
+
+        var hubConnection = builder.Build();
+        var closedEventTcs = new TaskCompletionSource<Exception>();
+        hubConnection.Closed += e =>
+        {
+            closedEventTcs.SetResult(e);
+            return Task.CompletedTask;
+        };
+
+        await hubConnection.StartAsync().DefaultTimeout();
+
+        await hubConnection.StopAsync().DefaultTimeout();
+
+        var exception = await closedEventTcs.Task.DefaultTimeout();
+        Assert.Null(exception);
+
+        await reconnectFeature.DisableReconnectCalled.DefaultTimeout();
+    }
+
     private class SampleObject
     {
         public SampleObject(string foo, int bar)
@@ -962,4 +1027,24 @@ public ReadOnlyMemory<byte> GetMessageBytes(HubMessage message)
             return HubProtocolExtensions.GetMessageBytes(this, message);
         }
     }
+
+#pragma warning disable CA2252 // This API requires opting into preview features
+    private sealed class TestReconnectFeature : IStatefulReconnectFeature
+#pragma warning restore CA2252 // This API requires opting into preview features
+    {
+        private TaskCompletionSource _disableReconnect = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public Task DisableReconnectCalled => _disableReconnect.Task;
+
+#pragma warning disable CA2252 // This API requires opting into preview features
+        public void OnReconnected(Func<PipeWriter, Task> notifyOnReconnected) { }
+#pragma warning restore CA2252 // This API requires opting into preview features
+
+#pragma warning disable CA2252 // This API requires opting into preview features
+        public void DisableReconnect()
+#pragma warning restore CA2252 // This API requires opting into preview features
+        {
+            _disableReconnect.TrySetResult();
+        }
+    }
 }

src/SignalR/clients/csharp/Http.Connections.Client/src/HttpConnection.cs

@@ -531,9 +531,11 @@ private async Task StartTransport(Uri connectUrl, HttpTransportType transportTyp
         // We successfully started, set the transport properties (we don't want to set these until the transport is definitely running).
         _transport = transport;
 
-        if (useAck && _transport is IReconnectFeature reconnectFeature)
+        if (useAck && _transport is IStatefulReconnectFeature reconnectFeature)
         {
+#pragma warning disable CA2252 // This API requires opting into preview features
             Features.Set(reconnectFeature);
+#pragma warning restore CA2252 // This API requires opting into preview features
         }
 
         Log.TransportStarted(_logger, transportType);