Skip to content

Commit f56c242

Browse files
BrennanConroyBrennanConroy
authored
[SignalR] Seamless Reconnect (#48338)
1 parent 51293e3 commit f56c242

58 files changed

Lines changed: 1972 additions & 144 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

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

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
// Licensed to the .NET Foundation under one or more agreements.
2+
// The .NET Foundation licenses this file to you under the MIT license.
3+
4+
using System;
5+
using System.Collections.Generic;
6+
using System.Linq;
7+
using System.Text;
8+
using System.Threading.Tasks;
9+
10+
namespace Microsoft.AspNetCore.Connections.Abstractions;
11+
12+
/// <summary>
13+
///
14+
/// </summary>
15+
public interface IReconnectFeature
16+
{
17+
/// <summary>
18+
///
19+
/// </summary>
20+
public Action NotifyOnReconnect { get; set; }
21+
22+
// TODO
23+
// void DisableReconnect();
24+
}

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,7 @@
11
#nullable enable
2+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature
3+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.get -> System.Action!
4+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.set -> void
25
Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature
36
Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature.Tags.get -> System.Collections.Generic.ICollection<System.Collections.Generic.KeyValuePair<string!, object?>>!
47
Microsoft.AspNetCore.Connections.Features.IConnectionNamedPipeFeature

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,7 @@
11
#nullable enable
2+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature
3+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.get -> System.Action!
4+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.set -> void
25
Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature
36
Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature.Tags.get -> System.Collections.Generic.ICollection<System.Collections.Generic.KeyValuePair<string!, object?>>!
47
Microsoft.AspNetCore.Connections.Features.IConnectionNamedPipeFeature

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,7 @@
11
#nullable enable
2+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature
3+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.get -> System.Action!
4+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.set -> void
25
Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature
36
Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature.Tags.get -> System.Collections.Generic.ICollection<System.Collections.Generic.KeyValuePair<string!, object?>>!
47
Microsoft.AspNetCore.Connections.Features.IConnectionNamedPipeFeature

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,7 @@
11
#nullable enable
2+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature
3+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.get -> System.Action!
4+
Microsoft.AspNetCore.Connections.Abstractions.IReconnectFeature.NotifyOnReconnect.set -> void
25
Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature
36
Microsoft.AspNetCore.Connections.Features.IConnectionMetricsTagsFeature.Tags.get -> System.Collections.Generic.ICollection<System.Collections.Generic.KeyValuePair<string!, object?>>!
47
Microsoft.AspNetCore.Connections.Features.IConnectionNamedPipeFeature

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -325,5 +325,14 @@ public static void ErrorHandshakeTimedOut(ILogger logger, TimeSpan handshakeTime
325325

326326
[LoggerMessage(89, LogLevel.Trace, "Error sending Completion message for stream '{StreamId}'.", EventName = "ErrorSendingStreamCompletion")]
327327
public static partial void ErrorSendingStreamCompletion(ILogger logger, string streamId, Exception exception);
328+
329+
[LoggerMessage(90, LogLevel.Trace, "Dropping {MessageType} with ID '{InvocationId}'.", EventName = "DroppingMessage")]
330+
public static partial void DroppingMessage(ILogger logger, string messageType, string? invocationId);
331+
332+
[LoggerMessage(91, LogLevel.Trace, "Received AckMessage with Sequence ID '{SequenceId}'.", EventName = "ReceivedAckMessage")]
333+
public static partial void ReceivedAckMessage(ILogger logger, long sequenceId);
334+
335+
[LoggerMessage(92, LogLevel.Trace, "Received SequenceMessage with Sequence ID '{SequenceId}'.", EventName = "ReceivedSequenceMessage")]
336+
public static partial void ReceivedSequenceMessage(ILogger logger, long sequenceId);
328337
}
329338
}

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

Lines changed: 74 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
using System.Diagnostics.CodeAnalysis;
99
using System.Globalization;
1010
using System.IO;
11+
using System.IO.Pipelines;
1112
using System.Linq;
1213
using System.Net;
1314
using System.Reflection;
@@ -16,6 +17,7 @@
1617
using System.Threading.Channels;
1718
using System.Threading.Tasks;
1819
using Microsoft.AspNetCore.Connections;
20+
using Microsoft.AspNetCore.Connections.Abstractions;
1921
using Microsoft.AspNetCore.Connections.Features;
2022
using Microsoft.AspNetCore.Internal;
2123
using Microsoft.AspNetCore.Shared;
@@ -946,11 +948,19 @@ private async Task InvokeStreamCore(ConnectionState connectionState, string meth
946948
private async Task SendHubMessage(ConnectionState connectionState, HubMessage hubMessage, CancellationToken cancellationToken = default)
947949
{
948950
_state.AssertConnectionValid();
949-
_protocol.WriteMessage(hubMessage, connectionState.Connection.Transport.Output);
950951

951952
Log.SendingMessage(_logger, hubMessage);
952953

953-
await connectionState.Connection.Transport.Output.FlushAsync(cancellationToken).ConfigureAwait(false);
954+
if (connectionState.UsingAcks())
955+
{
956+
await connectionState.WriteAsync(new SerializedHubMessage(hubMessage), cancellationToken).ConfigureAwait(false);
957+
}
958+
else
959+
{
960+
_protocol.WriteMessage(hubMessage, connectionState.Connection.Transport.Output);
961+
962+
await connectionState.Connection.Transport.Output.FlushAsync(cancellationToken).ConfigureAwait(false);
963+
}
954964
Log.MessageSent(_logger, hubMessage);
955965

956966
// We've sent a message, so don't ping for a while
@@ -1004,6 +1014,11 @@ private async Task SendWithLock(ConnectionState expectedConnectionState, HubMess
10041014
Log.ResettingKeepAliveTimer(_logger);
10051015
connectionState.ResetTimeout();
10061016

1017+
if (!connectionState.ShouldProcessMessage(message))
1018+
{
1019+
return null;
1020+
}
1021+
10071022
InvocationRequest? irq;
10081023
switch (message)
10091024
{
@@ -1055,6 +1070,14 @@ private async Task SendWithLock(ConnectionState expectedConnectionState, HubMess
10551070
Log.ReceivedPing(_logger);
10561071
// timeout is reset above, on receiving any message
10571072
break;
1073+
case AckMessage ackMessage:
1074+
Log.ReceivedAckMessage(_logger, ackMessage.SequenceId);
1075+
connectionState.Ack(ackMessage);
1076+
break;
1077+
case SequenceMessage sequenceMessage:
1078+
Log.ReceivedSequenceMessage(_logger, sequenceMessage.SequenceId);
1079+
connectionState.ResetSequence(sequenceMessage);
1080+
break;
10581081
default:
10591082
throw new InvalidOperationException($"Unexpected message type: {message.GetType().FullName}");
10601083
}
@@ -1235,6 +1258,7 @@ private async Task HandshakeAsync(ConnectionState startingConnectionState, Cance
12351258
}
12361259

12371260
Log.HandshakeComplete(_logger);
1261+
12381262
break;
12391263
}
12401264
}
@@ -1813,6 +1837,7 @@ private sealed class ConnectionState : IInvocationBinder
18131837
private readonly HubConnection _hubConnection;
18141838
private readonly ILogger _logger;
18151839
private readonly bool _hasInherentKeepAlive;
1840+
private readonly MessageBuffer? _messageBuffer;
18161841

18171842
private readonly object _lock = new object();
18181843
private readonly Dictionary<string, InvocationRequest> _pendingCalls = new Dictionary<string, InvocationRequest>(StringComparer.Ordinal);
@@ -1850,6 +1875,13 @@ public ConnectionState(ConnectionContext connection, HubConnection hubConnection
18501875

18511876
_logger = _hubConnection._logger;
18521877
_hasInherentKeepAlive = connection.Features.Get<IConnectionInherentKeepAliveFeature>()?.HasInherentKeepAlive ?? false;
1878+
1879+
if (Connection.Features.Get<IReconnectFeature>() is IReconnectFeature feature)
1880+
{
1881+
_messageBuffer = new MessageBuffer(connection, hubConnection._protocol);
1882+
1883+
feature.NotifyOnReconnect = _messageBuffer.Resend;
1884+
}
18531885
}
18541886

18551887
public string GetNextId() => (++_nextInvocationId).ToString(CultureInfo.InvariantCulture);
@@ -1935,6 +1967,8 @@ private async Task StopAsyncCore()
19351967
{
19361968
Log.Stopping(_logger);
19371969

1970+
_messageBuffer?.Dispose();
1971+
19381972
// Complete our write pipe, which should cause everything to shut down
19391973
Log.TerminatingReceiveLoop(_logger);
19401974
Connection.Transport.Input.CancelPendingRead();
@@ -1966,6 +2000,44 @@ public async Task TimerLoop(TimerAwaitable timer)
19662000
}
19672001
}
19682002

2003+
public ValueTask<FlushResult> WriteAsync(SerializedHubMessage message, CancellationToken cancellationToken)
2004+
{
2005+
Debug.Assert(_messageBuffer is not null);
2006+
return _messageBuffer.WriteAsync(message, cancellationToken);
2007+
}
2008+
2009+
public bool ShouldProcessMessage(HubMessage message)
2010+
{
2011+
if (UsingAcks())
2012+
{
2013+
if (!_messageBuffer.ShouldProcessMessage(message))
2014+
{
2015+
Log.DroppingMessage(_logger, ((HubInvocationMessage)message).GetType().Name, ((HubInvocationMessage)message).InvocationId);
2016+
return false;
2017+
}
2018+
}
2019+
return true;
2020+
}
2021+
2022+
public void Ack(AckMessage ackMessage)
2023+
{
2024+
if (UsingAcks())
2025+
{
2026+
_messageBuffer.Ack(ackMessage);
2027+
}
2028+
}
2029+
2030+
public void ResetSequence(SequenceMessage sequenceMessage)
2031+
{
2032+
if (UsingAcks())
2033+
{
2034+
_messageBuffer.ResetSequence(sequenceMessage);
2035+
}
2036+
}
2037+
2038+
[MemberNotNullWhen(true, nameof(_messageBuffer))]
2039+
public bool UsingAcks() => _messageBuffer is not null;
2040+
19692041
public void ResetSendPing()
19702042
{
19712043
Volatile.Write(ref _nextActivationSendPing, (DateTime.UtcNow + _hubConnection.KeepAliveInterval).Ticks);

0 commit comments

Comments
 (0)