Permalink
Browse files

Fixed regression in reconnecting clients after the shutdown token is …

…triggered.

- Create a linked cancellation token that represents a timedout connection.
- Added a unit test.
Fixes #449.
  • Loading branch information...
davidfowl committed Jun 15, 2012
1 parent f25bf2a commit c7b1d7419e4d9c63b0cc0a9ba0bc08908a5beebc
@@ -10,8 +10,10 @@
namespace SignalR.Hosting.Memory
{
- public class MemoryHost : RoutingHost, IHttpClient
+ public class MemoryHost : RoutingHost, IHttpClient, IDisposable
{
+ private readonly CancellationTokenSource _shutDownToken = new CancellationTokenSource();
+
public MemoryHost()
: this(new DefaultDependencyResolver())
{
@@ -37,9 +39,9 @@ Task<IClientResponse> IHttpClient.PostAsync(string url, Action<IClientRequest> p
private Task<IClientResponse> ProcessRequest(string url, Action<IClientRequest> prepareRequest, Dictionary<string, string> postData)
{
var uri = new Uri(url);
-
PersistentConnection connection;
- if (TryGetConnection(uri.LocalPath, out connection))
+
+ if (!_shutDownToken.IsCancellationRequested && TryGetConnection(uri.LocalPath, out connection))
{
var tcs = new TaskCompletionSource<IClientResponse>();
var clientTokenSource = new CancellationTokenSource();
@@ -49,6 +51,7 @@ private Task<IClientResponse> ProcessRequest(string url, Action<IClientRequest>
Response response = null;
response = new Response(clientTokenSource.Token, () => tcs.TrySetResult(response));
var hostContext = new HostContext(request, response);
+ hostContext.Items[HostConstants.ShutdownToken] = _shutDownToken.Token;
connection.Initialize(DependencyResolver);
@@ -75,5 +78,10 @@ private Task<IClientResponse> ProcessRequest(string url, Action<IClientRequest>
return TaskAsyncHelper.FromError<IClientResponse>(new InvalidOperationException("Not a valid end point"));
}
+
+ public void Dispose()
+ {
+ _shutDownToken.Cancel(throwOnFirstException: false);
@kenegozi

kenegozi Jun 15, 2012

love the named parameter usage for clarity. more people should adopt that

+ }
}
}
@@ -85,6 +85,26 @@ public void SendRaisesOnReceivedFromAllEvents()
public class OnReconnectedAsync
{
+ [Fact]
+ public void ReconnectFiresAfterHostShutDown()
+ {
+ var host = new MemoryHost();
+ var conn = new MyReconnect();
+ host.DependencyResolver.Register(typeof(MyReconnect), () => conn);
+ host.MapConnection<MyReconnect>("/endpoint");
+
+ var connection = new Client.Connection("http://foo/endpoint");
+ connection.Start(host).Wait();
+
+ host.Dispose();
+
+ Thread.Sleep(TimeSpan.FromSeconds(5));
+
+ Assert.Equal(Client.ConnectionState.Reconnecting, connection.State);
+
+ connection.Stop();
+ }
+
[Fact]
public void ReconnectFiresAfterTimeOutSSE()
{
@@ -14,7 +14,6 @@ public ForeverTransport(HostContext context, IDependencyResolver resolver)
resolver.Resolve<IJsonSerializer>(),
resolver.Resolve<ITransportHeartBeat>())
{
-
}
public ForeverTransport(HostContext context, IJsonSerializer jsonSerializer, ITransportHeartBeat heartBeat)
@@ -181,13 +180,13 @@ private Task ProcessMessages(ITransportConnection connection, Action postReceive
private void ProcessMessagesImpl(TaskCompletionSource<object> taskCompletetionSource, ITransportConnection connection, Action postReceive = null)
{
- if (!IsTimedOut && !IsDisconnected && IsAlive && !HostShutdownToken.IsCancellationRequested)
+ if (!IsTimedOut && !IsDisconnected && IsAlive && !ConnectionEndToken.IsCancellationRequested)
{
// ResponseTask will either subscribe and wait for a signal then return new messages,
// or return immediately with messages that were pending
var receiveAsyncTask = LastMessageId == null
- ? connection.ReceiveAsync(TimeoutToken)
- : connection.ReceiveAsync(LastMessageId, TimeoutToken);
+ ? connection.ReceiveAsync(ConnectionEndToken)
+ : connection.ReceiveAsync(LastMessageId, ConnectionEndToken);
if (postReceive != null)
{
@@ -202,8 +202,8 @@ private Task ProcessReceiveRequest(ITransportConnection connection, Action postR
// ReceiveAsync() will async wait until a message arrives then return
var receiveTask = IsConnectRequest ?
- connection.ReceiveAsync(TimeoutToken) :
- connection.ReceiveAsync(MessageId, TimeoutToken);
+ connection.ReceiveAsync(ConnectionEndToken) :
+ connection.ReceiveAsync(MessageId, ConnectionEndToken);
if (postReceive != null)
{
@@ -17,6 +17,7 @@ public abstract class TransportDisconnectBase : ITrackingConnection
protected int _isDisconnected;
private readonly CancellationTokenSource _timeoutTokenSource;
private readonly CancellationToken _hostShutdownToken;
+ private readonly CancellationTokenSource _connectionEndToken;
public TransportDisconnectBase(HostContext context, IJsonSerializer jsonSerializer, ITransportHeartBeat heartBeat)
{
@@ -25,6 +26,9 @@ public TransportDisconnectBase(HostContext context, IJsonSerializer jsonSerializ
_heartBeat = heartBeat;
_timeoutTokenSource = new CancellationTokenSource();
_hostShutdownToken = context.HostShutdownToken();
+
+ // Create a token that represents the end of this connection's life
+ _connectionEndToken = CancellationTokenSource.CreateLinkedTokenSource(_timeoutTokenSource.Token, _hostShutdownToken);
}
public string ConnectionId
@@ -62,19 +66,11 @@ public virtual bool IsAlive
get { return _context.Response.IsClientConnected; }
}
- protected CancellationToken TimeoutToken
- {
- get
- {
- return _timeoutTokenSource.Token;
- }
- }
-
- protected CancellationToken HostShutdownToken
+ protected CancellationToken ConnectionEndToken
{
get
{
- return _hostShutdownToken;
+ return _connectionEndToken.Token;
}
}

0 comments on commit c7b1d74

Please sign in to comment.