Disposing the subscription only after the connection closed event was raised

Author: darkl

src/net45/Tests/WampSharp.Tests.Wampv2/Integration/GoodbyeTests.cs

@@ -0,0 +1,63 @@
+using System.Reactive.Concurrency;
+using System.Threading.Tasks;
+using NUnit.Framework;
+using WampSharp.Binding;
+using WampSharp.V2;
+using WampSharp.V2.Client;
+using WampSharp.V2.Core.Contracts;
+using WampSharp.V2.Rpc;
+using WampSharp.V2.Transports;
+
+namespace WampSharp.Tests.Wampv2.Integration
+{
+    public class GoodbyeTests
+    {
+        [Test]
+        public async Task InMemoryTest()
+        {
+            var transport = new InMemoryTransport(Scheduler.Default);
+            var binding = new JTokenJsonBinding();
+            var realm = "some.realm";
+
+            var router = new WampHost();
+            router.RegisterTransport(transport, new[] { binding });
+            router.Open();
+
+            var calleeConnection = transport.CreateClientConnection(binding, Scheduler.Default);
+            WampChannelFactory factory = new WampChannelFactory();
+
+            var callee = factory.CreateChannel(realm, calleeConnection, binding);
+            await callee.Open();
+            await callee.RealmProxy.Services.RegisterCallee(new WampTest());
+
+
+            var callerConnection = transport.CreateClientConnection(binding, Scheduler.Default);
+            var caller = factory.CreateChannel(realm, callerConnection, binding);
+            await caller.Open();
+
+            var proxy = caller.RealmProxy.Services.GetCalleeProxy<IWampTest>();
+            var result = await proxy.Echo("1");
+            Assert.That(result, Is.EqualTo("1"));
+
+            await caller.Close(WampErrors.CloseNormal, new GoodbyeDetails());
+            await callee.Close(WampErrors.CloseNormal, new GoodbyeDetails());
+
+            router.Dispose();
+        }
+
+        public interface IWampTest
+        {
+            [WampProcedure("com.test.echo")]
+            Task<string> Echo(string message);
+        }
+
+        public class WampTest : IWampTest
+        {
+            public Task<string> Echo(string message)
+            {
+                return Task.FromResult(message);
+            }
+        }
+
+    }
+}

src/net45/WampSharp/WAMP2/V2/Transports/InMemory/InMemoryConnectionListener.cs

@@ -99,6 +99,8 @@ private void OnError(Exception exception)
             private void OnCompleted()
             {
                 mConnectionClosed.OnNext(Unit.Default);
+                mSubscription.Dispose();
+                mSubscription = null;
             }
 
             private void OnNewMessage(WampMessage<TMessage> wampMessage)
@@ -113,9 +115,7 @@ public void Connect()
 
             public void Dispose()
             {
-                mSubscription.Dispose();
                 mOutgoing.OnCompleted();
-                mSubscription = null;
             }
 
             public void Send(WampMessage<object> message)