Merge branch 'client-disconnection' into develop

Elad Zelingher · Elad Zelingher · commit 3dcebd1523ed · 2015-07-04T18:56:32.000+03:00
diff --git a/src/net45/Default/WampSharp.Fleck/Fleck/FleckWampConnection.cs b/src/net45/Default/WampSharp.Fleck/Fleck/FleckWampConnection.cs
@@ -1,16 +1,25 @@
 ﻿using System;
+using System.Reactive;
+using System.Reactive.Linq;
+using System.Reactive.Threading.Tasks;
+using System.Threading.Tasks;
 using Fleck;
 using WampSharp.Core.Listener;
+using WampSharp.Logging;
 
 namespace WampSharp.Fleck
 {
     internal abstract class FleckWampConnection<TMessage> : AsyncWampConnection<TMessage>
     {
         protected IWebSocketConnection mWebSocketConnection;
-
-        public FleckWampConnection(IWebSocketConnection webSocketConnection)
+        private readonly byte[] mPingBuffer = new byte[8];
+        private readonly TimeSpan mAutoSendPingInterval;
+        
+        public FleckWampConnection(IWebSocketConnection webSocketConnection, 
+            TimeSpan? autoSendPingInterval = null)
         {
             mWebSocketConnection = webSocketConnection;
+            mAutoSendPingInterval = autoSendPingInterval ?? TimeSpan.FromSeconds(45);
             mWebSocketConnection.OnOpen = OnConnectionOpen;
             mWebSocketConnection.OnError = OnConnectionError;
             mWebSocketConnection.OnClose = OnConnectionClose;
@@ -19,6 +28,73 @@ public FleckWampConnection(IWebSocketConnection webSocketConnection)
         private void OnConnectionOpen()
         {
             RaiseConnectionOpen();
+
+            StartPing();
+        }
+
+#if NET40
+        private void StartPing()
+        {
+            Observable.Defer
+                (() => Observable.FromAsync
+                     (() =>
+                     {
+                         byte[] ticks = GetCurrentTicks();
+                         return mWebSocketConnection.SendPing(ticks);
+                     })
+                                 .Concat(Observable.Timer(mAutoSendPingInterval)
+                                                   .Select(y => Unit.Default))
+                )
+                      .Repeat()
+                      .ToTask()
+                      .ContinueWith(x =>
+                      {
+                          if (x.Exception != null)
+                          {
+                              mLogger.WarnException("Failed pinging remote peer", x.Exception);
+                          }
+                      });
+        }
+
+#elif NET45
+        private void StartPing()
+        {
+            Task.Run((Action)Ping);
+        }
+
+        // We currently detect only disconnected clients and not "zombies",
+        // i.e. clients that respond relatively slow.
+        private async void Ping()
+        {
+            while (IsConnected)
+            {
+                try
+                {
+                    byte[] ticks = GetCurrentTicks();
+                    await mWebSocketConnection.SendPing(ticks);
+                    await Task.Delay(mAutoSendPingInterval);
+                }
+                catch (Exception ex)
+                {
+                    mLogger.WarnException("Failed pinging remote peer", ex);
+                }
+            }
+        }
+#endif
+
+        private byte[] GetCurrentTicks()
+        {
+            DateTime now = DateTime.Now;
+            long ticks = now.Ticks;
+            long bytes = ticks;
+
+            for (int i = 0; i < 8; i++)
+            {
+                mPingBuffer[i] = (byte) bytes;
+                bytes = bytes >> 8;
+            }
+
+            return mPingBuffer;
         }
 
         private void OnConnectionError(Exception exception)
diff --git a/src/net45/WampSharp/Core/Listener/Connections/WampConnectionMonitor.cs b/src/net45/WampSharp/Core/Listener/Connections/WampConnectionMonitor.cs
@@ -1,7 +1,4 @@
 ﻿using System;
-using System.Reactive;
-using System.Reactive.Linq;
-using System.Reactive.Subjects;
 
 namespace WampSharp.Core.Listener
 {
@@ -15,16 +12,11 @@ internal class WampConnectionMonitor<TMessage> : IWampConnectionMonitor
         private readonly IWampConnection<TMessage> mConnection;
         private bool mConnected;
 
-        private readonly IEventPatternSource<EventArgs> mDisconnectionEventSource;
-        private readonly ISubject<EventPattern<EventArgs>> mDisconnectionSubject;
-
         public WampConnectionMonitor(IWampConnection<TMessage> connection)
         {
             mConnected = true;
             mConnection = connection;
-            mDisconnectionSubject = new ReplaySubject<EventPattern<EventArgs>>(1);
 
-            mDisconnectionEventSource = mDisconnectionSubject.ToEventPattern();
             mConnection.ConnectionError += OnConnectionError;
             mConnection.ConnectionClosed += OnConnectionClosed;
         }
@@ -59,19 +51,14 @@ private void OnConnectionClosed()
 
         private void RaiseConnectionClosed(object client, EventArgs empty)
         {
-            mDisconnectionSubject.OnNext(new EventPattern<EventArgs>(client, empty));
-        }
+            EventHandler connectionClosed = ConnectionClosed;
 
-        public event EventHandler ConnectionClosed
-        {
-            add
+            if (connectionClosed != null)
             {
-                mDisconnectionEventSource.OnNext += new EventHandler<EventArgs>(value);
-            }
-            remove
-            {
-                mDisconnectionEventSource.OnNext -= new EventHandler<EventArgs>(value);
+                connectionClosed(client, empty);
             }
         }
+
+        public event EventHandler ConnectionClosed;
     }
 }
diff --git a/src/net45/WampSharp/Core/Listener/WampListener.cs b/src/net45/WampSharp/Core/Listener/WampListener.cs
@@ -76,6 +76,7 @@ public virtual void Stop()
 
         protected virtual void OnConnectionException(IWampConnection<TMessage> connection, Exception exception)
         {
+            OnCloseConnection(connection);
         }
 
         protected virtual void OnCloseConnection(IWampConnection<TMessage> connection)
@@ -139,6 +140,7 @@ protected virtual void OnNewConnection(IWampConnection<TMessage> connection)
 
             connection.MessageArrived += OnNewMessage;
             connection.ConnectionOpen += OnConnectionOpen;
+            connection.ConnectionError += OnConnectionError;
             connection.ConnectionClosed += OnConnectionClose;
         }
 
@@ -159,11 +161,22 @@ private void OnConnectionOpen(object sender, EventArgs e)
             OnConnectionOpen(connection);
         }
 
+        private void OnConnectionError(object sender, WampConnectionErrorEventArgs e)
+        {
+            OnConnectionClosed(sender);
+        }
+
         private void OnConnectionClose(object sender, EventArgs e)
+        {
+            OnConnectionClosed(sender);
+        }
+
+        private void OnConnectionClosed(object sender)
         {
             IWampConnection<TMessage> connection = sender as IWampConnection<TMessage>;
             connection.ConnectionClosed -= OnConnectionClose;
             connection.MessageArrived -= OnNewMessage;
+            connection.ConnectionError -= OnConnectionError;
             OnCloseConnection(connection);
         }
     }
diff --git a/src/net45/WampSharp/WAMP2/V2/Rpc/Dealer/WampClientRouterCallbackAdapter.cs b/src/net45/WampSharp/WAMP2/V2/Rpc/Dealer/WampClientRouterCallbackAdapter.cs
@@ -16,9 +16,15 @@ public WampClientRouterCallbackAdapter(IWampRawRpcOperationClientCallback caller
         {
             mCaller = caller;
             mNotifier = mCaller as ICallbackDisconnectionNotifier;
+            mNotifier.Disconnected += OnDisconnected;
             mOptions = options;
         }
 
+        private void OnDisconnected(object sender, EventArgs e)
+        {
+            RaiseDisconnected();
+        }
+
         public void Result<TMessage>(IWampFormatter<TMessage> formatter, YieldOptions details)
         {
             ResultDetails resultDetails = GetResultDetails(details);
@@ -59,15 +65,15 @@ private ResultDetails GetResultDetails(YieldOptions details)
             return new ResultDetails {Progress = details.Progress};
         }
 
-        public event EventHandler Disconnected
+        public event EventHandler Disconnected;
+
+        private void RaiseDisconnected()
         {
-            add
-            {
-                mNotifier.Disconnected += value;
-            }
-            remove
+            EventHandler handler = Disconnected;
+            
+            if (handler != null)
             {
-                mNotifier.Disconnected -= value;                
+                handler(this, EventArgs.Empty);
             }
         }
     }
diff --git a/src/net45/WampSharp/WAMP2/V2/Rpc/Dealer/WampRpcOperationCallback.cs b/src/net45/WampSharp/WAMP2/V2/Rpc/Dealer/WampRpcOperationCallback.cs
@@ -24,6 +24,7 @@ public WampRpcOperationCallback(IWampCaller caller, long requestId)
             mRequestId = requestId;
 
             mMonitor = caller as IWampConnectionMonitor;
+            mMonitor.ConnectionClosed += OnConnectionClosed;
         }
 
         public void Result<TResult>(IWampFormatter<TResult> formatter, ResultDetails details)
@@ -56,15 +57,20 @@ public void Error<TResult>(IWampFormatter<TResult> formatter, TResult details, s
             mCaller.CallError(mRequestId, details, error, arguments.Cast<object>().ToArray(), argumentsKeywords);
         }
 
-        public event EventHandler Disconnected
+        public event EventHandler Disconnected;
+
+        private void OnConnectionClosed(object sender, EventArgs e)
         {
-            add
-            {
-                mMonitor.ConnectionClosed += value;
-            }
-            remove
+            RaiseDisconnected();
+        }
+
+        private void RaiseDisconnected()
+        {
+            EventHandler handler = Disconnected;
+            
+            if (handler != null)
             {
-                mMonitor.ConnectionClosed -= value;                
+                handler(this, EventArgs.Empty);
             }
         }
 

Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,4 @@`
`1`	`1`	`using System;`
`2`		`-using System.Reactive;`
`3`		`-using System.Reactive.Linq;`
`4`		`-using System.Reactive.Subjects;`
`5`	`2`
`6`	`3`	`namespace WampSharp.Core.Listener`
`7`	`4`	`{`
`@@ -15,16 +12,11 @@ internal class WampConnectionMonitor<TMessage> : IWampConnectionMonitor`
`15`	`12`	`private readonly IWampConnection<TMessage> mConnection;`
`16`	`13`	`private bool mConnected;`
`17`	`14`
`18`		`- private readonly IEventPatternSource<EventArgs> mDisconnectionEventSource;`
`19`		`- private readonly ISubject<EventPattern<EventArgs>> mDisconnectionSubject;`
`20`		`-`
`21`	`15`	`public WampConnectionMonitor(IWampConnection<TMessage> connection)`
`22`	`16`	`{`
`23`	`17`	`mConnected = true;`
`24`	`18`	`mConnection = connection;`
`25`		`- mDisconnectionSubject = new ReplaySubject<EventPattern<EventArgs>>(1);`
`26`	`19`
`27`		`- mDisconnectionEventSource = mDisconnectionSubject.ToEventPattern();`
`28`	`20`	`mConnection.ConnectionError += OnConnectionError;`
`29`	`21`	`mConnection.ConnectionClosed += OnConnectionClosed;`
`30`	`22`	`}`
`@@ -59,19 +51,14 @@ private void OnConnectionClosed()`
`59`	`51`
`60`	`52`	`private void RaiseConnectionClosed(object client, EventArgs empty)`
`61`	`53`	`{`
`62`		`- mDisconnectionSubject.OnNext(new EventPattern<EventArgs>(client, empty));`
`63`		`- }`
	`54`	`+ EventHandler connectionClosed = ConnectionClosed;`
`64`	`55`
`65`		`- public event EventHandler ConnectionClosed`
`66`		`- {`
`67`		`- add`
	`56`	`+ if (connectionClosed != null)`
`68`	`57`	`{`
`69`		`- mDisconnectionEventSource.OnNext += new EventHandler<EventArgs>(value);`
`70`		`- }`
`71`		`- remove`
`72`		`- {`
`73`		`- mDisconnectionEventSource.OnNext -= new EventHandler<EventArgs>(value);`
	`58`	`+ connectionClosed(client, empty);`
`74`	`59`	`}`
`75`	`60`	`}`
	`61`	`+`
	`62`	`+ public event EventHandler ConnectionClosed;`
`76`	`63`	`}`
`77`	`64`	`}`
Original file line number	Diff line number	Diff line change
`@@ -76,6 +76,7 @@ public virtual void Stop()`
`76`	`76`
`77`	`77`	`protected virtual void OnConnectionException(IWampConnection<TMessage> connection, Exception exception)`
`78`	`78`	`{`
	`79`	`+ OnCloseConnection(connection);`
`79`	`80`	`}`
`80`	`81`
`81`	`82`	`protected virtual void OnCloseConnection(IWampConnection<TMessage> connection)`
`@@ -139,6 +140,7 @@ protected virtual void OnNewConnection(IWampConnection<TMessage> connection)`
`139`	`140`
`140`	`141`	`connection.MessageArrived += OnNewMessage;`
`141`	`142`	`connection.ConnectionOpen += OnConnectionOpen;`
	`143`	`+ connection.ConnectionError += OnConnectionError;`
`142`	`144`	`connection.ConnectionClosed += OnConnectionClose;`
`143`	`145`	`}`
`144`	`146`
`@@ -159,11 +161,22 @@ private void OnConnectionOpen(object sender, EventArgs e)`
`159`	`161`	`OnConnectionOpen(connection);`
`160`	`162`	`}`
`161`	`163`
	`164`	`+ private void OnConnectionError(object sender, WampConnectionErrorEventArgs e)`
	`165`	`+ {`
	`166`	`+ OnConnectionClosed(sender);`
	`167`	`+ }`
	`168`	`+`
`162`	`169`	`private void OnConnectionClose(object sender, EventArgs e)`
	`170`	`+ {`
	`171`	`+ OnConnectionClosed(sender);`
	`172`	`+ }`
	`173`	`+`
	`174`	`+ private void OnConnectionClosed(object sender)`
`163`	`175`	`{`
`164`	`176`	`IWampConnection<TMessage> connection = sender as IWampConnection<TMessage>;`
`165`	`177`	`connection.ConnectionClosed -= OnConnectionClose;`
`166`	`178`	`connection.MessageArrived -= OnNewMessage;`
	`179`	`+ connection.ConnectionError -= OnConnectionError;`
`167`	`180`	`OnCloseConnection(connection);`
`168`	`181`	`}`
`169`	`182`	`}`
Original file line number	Diff line number	Diff line change
`@@ -16,9 +16,15 @@ public WampClientRouterCallbackAdapter(IWampRawRpcOperationClientCallback caller`
`16`	`16`	`{`
`17`	`17`	`mCaller = caller;`
`18`	`18`	`mNotifier = mCaller as ICallbackDisconnectionNotifier;`
	`19`	`+ mNotifier.Disconnected += OnDisconnected;`
`19`	`20`	`mOptions = options;`
`20`	`21`	`}`
`21`	`22`
	`23`	`+ private void OnDisconnected(object sender, EventArgs e)`
	`24`	`+ {`
	`25`	`+ RaiseDisconnected();`
	`26`	`+ }`
	`27`	`+`
`22`	`28`	`public void Result<TMessage>(IWampFormatter<TMessage> formatter, YieldOptions details)`
`23`	`29`	`{`
`24`	`30`	`ResultDetails resultDetails = GetResultDetails(details);`
`@@ -59,15 +65,15 @@ private ResultDetails GetResultDetails(YieldOptions details)`
`59`	`65`	`return new ResultDetails {Progress = details.Progress};`
`60`	`66`	`}`
`61`	`67`
`62`		`- public event EventHandler Disconnected`
	`68`	`+ public event EventHandler Disconnected;`
	`69`	`+`
	`70`	`+ private void RaiseDisconnected()`
`63`	`71`	`{`
`64`		`- add`
`65`		`- {`
`66`		`- mNotifier.Disconnected += value;`
`67`		`- }`
`68`		`- remove`
	`72`	`+ EventHandler handler = Disconnected;`
	`73`	`+`
	`74`	`+ if (handler != null)`
`69`	`75`	`{`
`70`		`- mNotifier.Disconnected -= value;`
	`76`	`+ handler(this, EventArgs.Empty);`
`71`	`77`	`}`
`72`	`78`	`}`
`73`	`79`	`}`
Original file line number	Diff line number	Diff line change
`@@ -24,6 +24,7 @@ public WampRpcOperationCallback(IWampCaller caller, long requestId)`
`24`	`24`	`mRequestId = requestId;`
`25`	`25`
`26`	`26`	`mMonitor = caller as IWampConnectionMonitor;`
	`27`	`+ mMonitor.ConnectionClosed += OnConnectionClosed;`
`27`	`28`	`}`
`28`	`29`
`29`	`30`	`public void Result<TResult>(IWampFormatter<TResult> formatter, ResultDetails details)`
`@@ -56,15 +57,20 @@ public void Error<TResult>(IWampFormatter<TResult> formatter, TResult details, s`
`56`	`57`	`mCaller.CallError(mRequestId, details, error, arguments.Cast<object>().ToArray(), argumentsKeywords);`
`57`	`58`	`}`
`58`	`59`
`59`		`- public event EventHandler Disconnected`
	`60`	`+ public event EventHandler Disconnected;`
	`61`	`+`
	`62`	`+ private void OnConnectionClosed(object sender, EventArgs e)`
`60`	`63`	`{`
`61`		`- add`
`62`		`- {`
`63`		`- mMonitor.ConnectionClosed += value;`
`64`		`- }`
`65`		`- remove`
	`64`	`+ RaiseDisconnected();`
	`65`	`+ }`
	`66`	`+`
	`67`	`+ private void RaiseDisconnected()`
	`68`	`+ {`
	`69`	`+ EventHandler handler = Disconnected;`
	`70`	`+`
	`71`	`+ if (handler != null)`
`66`	`72`	`{`
`67`		`- mMonitor.ConnectionClosed -= value;`
	`73`	`+ handler(this, EventArgs.Empty);`
`68`	`74`	`}`
`69`	`75`	`}`
`70`	`76`