Merge pull request #61 from Code-Sharp/reconnect

Reconnect mechanism

Author: darkl

.travis.yml

@@ -14,4 +14,4 @@ script:
 
 branches:
  only:
-  - develop
+  - develop

README.md

@@ -1,6 +1,6 @@
 WampSharp
 =========
-
+[![NuGet Version][NuGetImgMaster]][NuGetLinkMaster]
 
 A C# implementation of [WAMP (The Web Application Messaging Protocol)][WampLink]
 
@@ -15,6 +15,11 @@ Master | Provider
 [![Build Status][WinImgMaster]][WinLinkMaster] | Windows CI Provided By [CodeBetter][] and [JetBrains][] 
 [![Build Status][MonoImgMaster]][MonoLinkMaster] | Mono CI Provided by [travis-ci][] 
 
+
+## WampSharp v1.2.1.6-beta
+
+WampSharp v1.2.1.6-beta released, see version [release notes](https://github.com/Code-Sharp/WampSharp/wiki/WampSharp-v1.2.1.6-beta-release-notes).
+
 ## Get Started
 
 See [Get started tutorial](https://github.com/Code-Sharp/WampSharp/wiki/Getting-started-with-WAMPv2) and
@@ -32,6 +37,12 @@ WAMPv1 support is still available. You can read about it at the [Wiki](https://g
 
 If you're updating from a previous WampSharp version and you're not interested yet in updating your application to WAMPv2, please read the following [notes](https://github.com/Code-Sharp/WampSharp/wiki/Notes-for-WAMPv1-users).
 
+## Donations
+
+If you found WampSharp helpful and want to donate, you are welcome to do so via [PayPal](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=UHRAS9KZPNPX4).
+
+Your donations help keep this project development alive.
+
 [WampLink]:http://wamp.ws
 
 [WinImgMaster]:http://teamcity.codebetter.com/app/rest/builds/buildType:\(id:WampSharp_Dev_Build\)/statusIcon

releasenotes.md

@@ -1,4 +1,4 @@
-WampSharp v1.2.1.0-beta release notes
+WampSharp v1.2.1.6-beta release notes
 =================================
 
 **Contents**

src/mono/WampSharp/WampSharp.csproj

@@ -597,6 +597,9 @@
     <Compile Include="..\..\net45\WampSharp\Core\Utilities\AsyncDisposable\IAsyncDisposable.cs">
       <Link>Core\Utilities\AsyncDisposable\IAsyncDisposable.cs</Link>
     </Compile>
+    <Compile Include="..\..\net45\WampSharp\WAMP2\V2\Api\WampChannelReconnector.cs">
+      <Link>WAMP2\V2\Api\WampChannelReconnector.cs</Link>
+    </Compile>
     <Compile Include="..\..\net45\WampSharp\WAMP2\V2\Client\Session\WampAuthenticationNotImplementedException.cs">
       <Link>WAMP2\V2\Client\Session\WampAuthenticationNotImplementedException.cs</Link>
     </Compile>

src/net40/WampSharp/WampSharp.csproj

@@ -601,6 +601,9 @@
     <Compile Include="..\..\net45\WampSharp\Core\Utilities\AsyncDisposable\IAsyncDisposable.cs">
       <Link>Core\Utilities\AsyncDisposable\IAsyncDisposable.cs</Link>
     </Compile>
+    <Compile Include="..\..\net45\WampSharp\WAMP2\V2\Api\WampChannelReconnector.cs">
+      <Link>WAMP2\V2\Api\WampChannelReconnector.cs</Link>
+    </Compile>
     <Compile Include="..\..\net45\WampSharp\WAMP2\V2\Client\Session\WampAuthenticationNotImplementedException.cs">
       <Link>WAMP2\V2\Client\Session\WampAuthenticationNotImplementedException.cs</Link>
     </Compile>

src/net45/WampSharp.Default/WebSocket4Net/WebSocket4NetConnection.cs

@@ -63,7 +63,6 @@ private void WebSocketOnClosed(object sender, EventArgs eventArgs)
         private void WebSocketOnError(object sender, ErrorEventArgs e)
         {
             RaiseConnectionError(e.Exception);
-            RaiseConnectionClosed();
         }
 
         public void Connect()
@@ -76,6 +75,17 @@ public virtual void Dispose()
             mWebSocket.Close();
         }
 
+        void IWampConnection<TMessage>.Send(WampMessage<TMessage> message)
+        {
+            try
+            {
+                Send(message);
+            }
+            catch (Exception ex)
+            {
+            }
+        }
+
         public abstract void Send(WampMessage<TMessage> message);
 
         public event EventHandler ConnectionOpen;

src/net45/WampSharp/WAMP2/V2/Api/WampChannelReconnector.cs

@@ -0,0 +1,84 @@
+using System;
+using System.Reactive;
+using System.Reactive.Disposables;
+using System.Reactive.Linq;
+using System.Threading.Tasks;
+using WampSharp.V2.Client;
+using WampSharp.V2.Realm;
+
+namespace WampSharp.V2
+{
+    /// <summary>
+    /// An object that reconnects to realm on connection loss.
+    /// </summary>
+    public class WampChannelReconnector : IDisposable
+    {
+        private IObservable<Unit> mMerged;
+        private IDisposable mDisposable = Disposable.Empty;
+        private bool mStarted = false;
+        private readonly object mLock = new object();
+
+        /// <summary>
+        /// Initializes a new instance of <see cref="WampChannelReconnector"/>.
+        /// </summary>
+        /// <param name="channel">The channel used to connect.</param>
+        /// <param name="connector">The Task to use in order to connect.</param>
+        public WampChannelReconnector(IWampChannel channel, Func<Task> connector)
+        {
+            IWampClientConnectionMonitor monitor = channel.RealmProxy.Monitor;
+
+            var connectionBrokenObservable =
+                Observable.FromEventPattern<WampSessionCloseEventArgs>
+                    (x => monitor.ConnectionBroken += x,
+                        x => monitor.ConnectionBroken -= x)
+                    .Select(x => Unit.Default);
+
+            var onceAndConnectionBroken =
+                Observable.Return(Unit.Default).Concat
+                    (connectionBrokenObservable);
+
+            IObservable<IObservable<Unit>> reconnect =
+                from connectionBroke in onceAndConnectionBroken
+                let tryReconnect = Observable.FromAsync(connector)
+                    .Catch<Unit, Exception>(x => Observable.Empty<Unit>())
+                select tryReconnect;
+
+            mMerged = reconnect.Concat();
+        }
+
+        /// <summary>
+        /// Start trying connection establishment to router.
+        /// </summary>
+        public void Start()
+        {
+            lock (mLock)
+            {
+                if (mStarted)
+                {
+                    throw new Exception("Already started");
+                }
+                else
+                {
+                    if (mMerged != null)
+                    {
+                        mDisposable = mMerged.Subscribe(x => { });
+                        mStarted = true;
+                    }
+                    else
+                    {
+                        throw new ObjectDisposedException(typeof (WampChannelReconnector).Name);
+                    }
+                }
+            }
+        }
+
+        public void Dispose()
+        {
+            lock (mLock)
+            {
+                mMerged = null;
+                mDisposable.Dispose();
+            }
+        }
+    }
+}

src/net45/WampSharp/WAMP2/V2/Client/Session/WampSessionClient.cs

@@ -110,8 +110,7 @@ public void Challenge(string authMethod, ChallengeDetails extra)
         public void Welcome(long session, TMessage details)
         {
             mSession = session;
-            mOpenTask.SetResult(true);
-            mOpenTask = null;
+            mOpenTask.TrySetResult(true);
 
             OnConnectionEstablished(new WampSessionEventArgs
                 (session, new SerializedValue<TMessage>(mFormatter, details)));
@@ -173,11 +172,11 @@ public IWampRealmProxy Realm
         }
 
         public Task OpenTask
-        {
-            get
-            {
-                return mOpenTask.Task;
-            }
+        {
+            get
+            {
+                return mOpenTask.Task;
+            }
         }
 
         public void Close(string reason, object details)
@@ -242,9 +241,9 @@ private void SetOpenTaskErrorIfNeeded(Exception exception)
                 mOpenTask = new TaskCompletionSource<bool>();
             }
 
-            if (openTask != null)
-            {
-                openTask.SetException(exception);
+            if (openTask != null)
+            {
+                openTask.TrySetException(exception);
             }
         }
 

src/net45/WampSharp/WampSharp.csproj

@@ -243,6 +243,7 @@
     <Compile Include="WAMP2\V2\Api\DelegatePubSub\WampSubscriberRegistrar.cs" />
     <Compile Include="WAMP2\V2\Api\WampEventContext.cs" />
     <Compile Include="Core\Utilities\AsyncDisposable\IAsyncDisposable.cs" />
+    <Compile Include="WAMP2\V2\Api\WampChannelReconnector.cs" />
     <Compile Include="WAMP2\V2\Client\Session\WampAuthenticationNotImplementedException.cs" />
     <Compile Include="WAMP2\V2\PubSub\Subscriber\LocalSubscriber.cs" />
     <Compile Include="WAMP2\V2\PubSub\Subscriber\MethodInfoSubscriber.cs" />