Merge pull request #55 from Code-Sharp/vtortola-work

Sending only one message at a time

Author: darkl

src/net45/Extensions/WampSharp.Vtortola/VtortolaWampBinaryConnection.cs

@@ -1,4 +1,5 @@
 using System.IO;
+using System.Threading.Tasks;
 using vtortola.WebSockets;
 using WampSharp.Core.Message;
 using WampSharp.V2.Binding;
@@ -26,13 +27,13 @@ protected override WampMessage<TMessage> ParseMessage(WebSocketMessageReadStream
             }
         }
 
-        public override void Send(WampMessage<TMessage> message)
+        protected async override Task SendAsync(WampMessage<TMessage> message)
         {
             using (WebSocketMessageWriteStream stream = 
                 mWebsocket.CreateMessageWriter(WebSocketMessageType.Binary))
             {
                 byte[] raw = mBinding.Format(message);
-                stream.Write(raw, 0, raw.Length);
+                await stream.WriteAsync(raw, 0, raw.Length).ConfigureAwait(false);
             }
         }
     }

src/net45/Extensions/WampSharp.Vtortola/VtortolaWampConnection.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Threading.Tasks.Dataflow;
 using vtortola.WebSockets;
 using WampSharp.Core.Listener;
 using WampSharp.Core.Message;
@@ -10,10 +11,14 @@ namespace WampSharp.Vtortola
     internal abstract class VtortolaWampConnection<TMessage> : IWampConnection<TMessage>
     {
         protected readonly WebSocket mWebsocket;
+        private readonly ActionBlock<WampMessage<TMessage>> mSendBlock;
 
-        public VtortolaWampConnection(WebSocket websocket)
+        protected VtortolaWampConnection(WebSocket websocket)
         {
             mWebsocket = websocket;
+
+            mSendBlock = new ActionBlock<WampMessage<TMessage>>(x => InnerSend(x),
+                new ExecutionDataflowBlockOptions() {MaxDegreeOfParallelism = 1});
         }
 
         public async Task HandleWebSocketAsync()
@@ -48,7 +53,7 @@ await mWebsocket.ReadMessageAsync(CancellationToken.None)
 
                 if (ConnectionClosed != null)
                 {
-                    ConnectionClosed(this, EventArgs.Empty);                    
+                    ConnectionClosed(this, EventArgs.Empty);
                 }
             }
             catch (Exception ex)
@@ -64,14 +69,35 @@ await mWebsocket.ReadMessageAsync(CancellationToken.None)
 
         public void Dispose()
         {
+            mSendBlock.Complete();
+            mSendBlock.Completion.Wait();
             mWebsocket.Dispose();
         }
 
-        public abstract void Send(WampMessage<TMessage> message);
+        public void Send(WampMessage<TMessage> message)
+        {
+            mSendBlock.Post(message);
+        }
+
+        private async Task InnerSend(WampMessage<TMessage> message)
+        {
+            if (mWebsocket.IsConnected)
+            {
+                try
+                {
+                    await SendAsync(message)
+                        .ConfigureAwait(false);
+                }
+                catch
+                {
+                }                
+            }
+        }
 
         public event EventHandler ConnectionOpen;
         public event EventHandler<WampMessageArrivedEventArgs<TMessage>> MessageArrived;
         public event EventHandler ConnectionClosed;
         public event EventHandler<WampConnectionErrorEventArgs> ConnectionError;
+        protected abstract Task SendAsync(WampMessage<TMessage> message);
     }
 }

src/net45/Extensions/WampSharp.Vtortola/VtortolaWampTextConnection.cs

@@ -1,4 +1,5 @@
 using System.IO;
+using System.Threading.Tasks;
 using vtortola.WebSockets;
 using WampSharp.Core.Message;
 using WampSharp.V2.Binding;
@@ -25,7 +26,7 @@ protected override WampMessage<TMessage> ParseMessage(WebSocketMessageReadStream
             }
         }
 
-        public override void Send(WampMessage<TMessage> message)
+        protected override async Task SendAsync(WampMessage<TMessage> message)
         {
             string raw = mBinding.Format(message);
 
@@ -34,7 +35,8 @@ public override void Send(WampMessage<TMessage> message)
             {
                 using (StreamWriter streamWriter = new StreamWriter(stream))
                 {
-                    streamWriter.Write(raw);
+                    await streamWriter.WriteAsync(raw)
+                        .ConfigureAwait(false);
                 }
             }
         }

src/net45/WampSharp/WAMP2/V2/Session/WampSessionServer.cs

@@ -42,7 +42,7 @@ public void OnNewClient(IWampClient<TMessage> client)
 
         public void OnClientDisconnect(IWampClient<TMessage> client)
         {
-            if (!client.GoodbyeSent)
+            if ((client.Realm != null) && !client.GoodbyeSent)
             {
                 client.Realm.SessionLost(client.Session);
             }