Disconnection detect issues

Elad Zelingher · Elad Zelingher · commit dc51977e898a · 2015-07-04T12:57:51.000+03:00
diff --git a/src/net45/Default/WampSharp.Fleck/Fleck/FleckWampConnection.cs b/src/net45/Default/WampSharp.Fleck/Fleck/FleckWampConnection.cs
@@ -1,4 +1,7 @@
 ﻿using System;
+using System.Reactive;
+using System.Reactive.Linq;
+using System.Threading.Tasks;
 using Fleck;
 using WampSharp.Core.Listener;
 
@@ -7,10 +10,14 @@ 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 +26,63 @@ public FleckWampConnection(IWebSocketConnection webSocketConnection)
         private void OnConnectionOpen()
         {
             RaiseConnectionOpen();
+
+            StartPing();
+        }
+
+#if NET40
+        private void StartPing()
+        {
+            Observable.Generate(0, x => true, x => x, x => x)
+                      .Select(x =>
+                                  Observable.FromAsync(() =>
+                                  {
+                                      byte[] ticks = GetCurrentTicks();
+                                      return mWebSocketConnection.SendPing(ticks);
+                                  }).Concat(Observable.Timer(mAutoSendPingInterval)
+                                                      .Select(y => Unit.Default))
+                )
+                      .Merge(1);
+        }
+
+#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)
+                {
+                }
+            }
+        }
+#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
@@ -1,5 +1,6 @@
 ﻿using System;
 using System.Reactive.Disposables;
+using SystemEx;
 using WampSharp.Core.Dispatch;
 using WampSharp.Core.Message;
 using WampSharp.Logging;
@@ -75,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)
@@ -136,6 +138,7 @@ protected virtual void OnNewConnection(IWampConnection<TMessage> connection)
 
             connection.MessageArrived += OnNewMessage;
             connection.ConnectionOpen += OnConnectionOpen;
+            connection.ConnectionError += OnConnectionError;
             connection.ConnectionClosed += OnConnectionClose;
         }
 
@@ -156,11 +159,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
`@@ -1,5 +1,6 @@`
`1`	`1`	`using System;`
`2`	`2`	`using System.Reactive.Disposables;`
	`3`	`+using SystemEx;`
`3`	`4`	`using WampSharp.Core.Dispatch;`
`4`	`5`	`using WampSharp.Core.Message;`
`5`	`6`	`using WampSharp.Logging;`
`@@ -75,6 +76,7 @@ public virtual void Stop()`
`75`	`76`
`76`	`77`	`protected virtual void OnConnectionException(IWampConnection<TMessage> connection, Exception exception)`
`77`	`78`	`{`
	`79`	`+ OnCloseConnection(connection);`
`78`	`80`	`}`
`79`	`81`
`80`	`82`	`protected virtual void OnCloseConnection(IWampConnection<TMessage> connection)`
`@@ -136,6 +138,7 @@ protected virtual void OnNewConnection(IWampConnection<TMessage> connection)`
`136`	`138`
`137`	`139`	`connection.MessageArrived += OnNewMessage;`
`138`	`140`	`connection.ConnectionOpen += OnConnectionOpen;`
	`141`	`+ connection.ConnectionError += OnConnectionError;`
`139`	`142`	`connection.ConnectionClosed += OnConnectionClose;`
`140`	`143`	`}`
`141`	`144`
`@@ -156,11 +159,22 @@ private void OnConnectionOpen(object sender, EventArgs e)`
`156`	`159`	`OnConnectionOpen(connection);`
`157`	`160`	`}`
`158`	`161`
	`162`	`+ private void OnConnectionError(object sender, WampConnectionErrorEventArgs e)`
	`163`	`+ {`
	`164`	`+ OnConnectionClosed(sender);`
	`165`	`+ }`
	`166`	`+`
`159`	`167`	`private void OnConnectionClose(object sender, EventArgs e)`
	`168`	`+ {`
	`169`	`+ OnConnectionClosed(sender);`
	`170`	`+ }`
	`171`	`+`
	`172`	`+ private void OnConnectionClosed(object sender)`
`160`	`173`	`{`
`161`	`174`	`IWampConnection<TMessage> connection = sender as IWampConnection<TMessage>;`
`162`	`175`	`connection.ConnectionClosed -= OnConnectionClose;`
`163`	`176`	`connection.MessageArrived -= OnNewMessage;`
	`177`	`+ connection.ConnectionError -= OnConnectionError;`
`164`	`178`	`OnCloseConnection(connection);`
`165`	`179`	`}`
`166`	`180`	`}`
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`