[SignalR] Seamless Reconnect

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

@@ -0,0 +1,24 @@
+// 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.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Microsoft.AspNetCore.Connections.Abstractions;
+
+/// <summary>
+/// 
+/// </summary>
+public interface IReconnectFeature
+{
+    /// <summary>
+    /// 
+    /// </summary>
+    public Action NotifyOnReconnect { get; set; }
+
+    // TODO
+    // void DisableReconnect();
+}

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

@@ -1,4 +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.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,4 +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.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,4 +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.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,4 +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.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.Log.cs

@@ -325,5 +325,14 @@ public static void ErrorHandshakeTimedOut(ILogger logger, TimeSpan handshakeTime
 
         [LoggerMessage(89, LogLevel.Trace, "Error sending Completion message for stream '{StreamId}'.", EventName = "ErrorSendingStreamCompletion")]
         public static partial void ErrorSendingStreamCompletion(ILogger logger, string streamId, Exception exception);
+
+        [LoggerMessage(90, LogLevel.Trace, "Dropping {MessageType} with ID '{InvocationId}'.", EventName = "DroppingMessage")]
+        public static partial void DroppingMessage(ILogger logger, string messageType, string? invocationId);
+
+        [LoggerMessage(91, LogLevel.Trace, "Received AckMessage with Sequence ID '{SequenceId}'.", EventName = "ReceivedAckMessage")]
+        public static partial void ReceivedAckMessage(ILogger logger, long sequenceId);
+
+        [LoggerMessage(92, LogLevel.Trace, "Received SequenceMessage with Sequence ID '{SequenceId}'.", EventName = "ReceivedSequenceMessage")]
+        public static partial void ReceivedSequenceMessage(ILogger logger, long sequenceId);
     }
 }

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

@@ -8,6 +8,7 @@
 using System.Diagnostics.CodeAnalysis;
 using System.Globalization;
 using System.IO;
+using System.IO.Pipelines;
 using System.Linq;
 using System.Net;
 using System.Reflection;
@@ -16,6 +17,7 @@
 using System.Threading.Channels;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
+using Microsoft.AspNetCore.Connections.Abstractions;
 using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Internal;
 using Microsoft.AspNetCore.Shared;
@@ -946,11 +948,19 @@ private async Task InvokeStreamCore(ConnectionState connectionState, string meth
     private async Task SendHubMessage(ConnectionState connectionState, HubMessage hubMessage, CancellationToken cancellationToken = default)
     {
         _state.AssertConnectionValid();
-        _protocol.WriteMessage(hubMessage, connectionState.Connection.Transport.Output);
 
         Log.SendingMessage(_logger, hubMessage);
 
-        await connectionState.Connection.Transport.Output.FlushAsync(cancellationToken).ConfigureAwait(false);
+        if (connectionState.UsingAcks())
+        {
+            await connectionState.WriteAsync(new SerializedHubMessage(hubMessage), cancellationToken).ConfigureAwait(false);
+        }
+        else
+        {
+            _protocol.WriteMessage(hubMessage, connectionState.Connection.Transport.Output);
+
+            await connectionState.Connection.Transport.Output.FlushAsync(cancellationToken).ConfigureAwait(false);
+        }
         Log.MessageSent(_logger, hubMessage);
 
         // We've sent a message, so don't ping for a while
@@ -1004,6 +1014,11 @@ private async Task SendWithLock(ConnectionState expectedConnectionState, HubMess
         Log.ResettingKeepAliveTimer(_logger);
         connectionState.ResetTimeout();
 
+        if (!connectionState.ShouldProcessMessage(message))
+        {
+            return null;
+        }
+
         InvocationRequest? irq;
         switch (message)
         {
@@ -1055,6 +1070,14 @@ private async Task SendWithLock(ConnectionState expectedConnectionState, HubMess
                 Log.ReceivedPing(_logger);
                 // timeout is reset above, on receiving any message
                 break;
+            case AckMessage ackMessage:
+                Log.ReceivedAckMessage(_logger, ackMessage.SequenceId);
+                connectionState.Ack(ackMessage);
+                break;
+            case SequenceMessage sequenceMessage:
+                Log.ReceivedSequenceMessage(_logger, sequenceMessage.SequenceId);
+                connectionState.ResetSequence(sequenceMessage);
+                break;
             default:
                 throw new InvalidOperationException($"Unexpected message type: {message.GetType().FullName}");
         }
@@ -1235,6 +1258,7 @@ private async Task HandshakeAsync(ConnectionState startingConnectionState, Cance
                                 }
 
                                 Log.HandshakeComplete(_logger);
+
                                 break;
                             }
                         }
@@ -1813,6 +1837,7 @@ private sealed class ConnectionState : IInvocationBinder
         private readonly HubConnection _hubConnection;
         private readonly ILogger _logger;
         private readonly bool _hasInherentKeepAlive;
+        private readonly MessageBuffer? _messageBuffer;
 
         private readonly object _lock = new object();
         private readonly Dictionary<string, InvocationRequest> _pendingCalls = new Dictionary<string, InvocationRequest>(StringComparer.Ordinal);
@@ -1850,6 +1875,13 @@ public ConnectionState(ConnectionContext connection, HubConnection hubConnection
 
             _logger = _hubConnection._logger;
             _hasInherentKeepAlive = connection.Features.Get<IConnectionInherentKeepAliveFeature>()?.HasInherentKeepAlive ?? false;
+
+            if (Connection.Features.Get<IReconnectFeature>() is IReconnectFeature feature)
+            {
+                _messageBuffer = new MessageBuffer(connection, hubConnection._protocol);
+
+                feature.NotifyOnReconnect = _messageBuffer.Resend;
+            }
         }
 
         public string GetNextId() => (++_nextInvocationId).ToString(CultureInfo.InvariantCulture);
@@ -1935,6 +1967,8 @@ private async Task StopAsyncCore()
         {
             Log.Stopping(_logger);
 
+            _messageBuffer?.Dispose();
+
             // Complete our write pipe, which should cause everything to shut down
             Log.TerminatingReceiveLoop(_logger);
             Connection.Transport.Input.CancelPendingRead();
@@ -1966,6 +2000,44 @@ public async Task TimerLoop(TimerAwaitable timer)
             }
         }
 
+        public ValueTask<FlushResult> WriteAsync(SerializedHubMessage message, CancellationToken cancellationToken)
+        {
+            Debug.Assert(_messageBuffer is not null);
+            return _messageBuffer.WriteAsync(message, cancellationToken);
+        }
+
+        public bool ShouldProcessMessage(HubMessage message)
+        {
+            if (UsingAcks())
+            {
+                if (!_messageBuffer.ShouldProcessMessage(message))
+                {
+                    Log.DroppingMessage(_logger, ((HubInvocationMessage)message).GetType().Name, ((HubInvocationMessage)message).InvocationId);
+                    return false;
+                }
+            }
+            return true;
+        }
+
+        public void Ack(AckMessage ackMessage)
+        {
+            if (UsingAcks())
+            {
+                _messageBuffer.Ack(ackMessage);
+            }
+        }
+
+        public void ResetSequence(SequenceMessage sequenceMessage)
+        {
+            if (UsingAcks())
+            {
+                _messageBuffer.ResetSequence(sequenceMessage);
+            }
+        }
+
+        [MemberNotNullWhen(true, nameof(_messageBuffer))]
+        public bool UsingAcks() => _messageBuffer is not null;
+
         public void ResetSendPing()
         {
             Volatile.Write(ref _nextActivationSendPing, (DateTime.UtcNow + _hubConnection.KeepAliveInterval).Ticks);