RawSocket pub/sub optimization

Serializing the message once for all the subscribers, even for RawSocket transport

Author: darkl

src/net45/Default/WampSharp.NewtonsoftJson/Newtonsoft/JTokenMessageParser.cs

@@ -1,5 +1,6 @@
 using System;
 using System.IO;
+using System.Text;
 using WampSharp.Logging;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
@@ -49,6 +50,11 @@ public string Format(WampMessage<object> message)
             return result;
         }
 
+        public byte[] GetBytes(string raw)
+        {
+            return Encoding.UTF8.GetBytes(raw);
+        }
+
         public WampMessage<JToken> Parse(Stream stream)
         {
             try

src/net45/Default/WampSharp.NewtonsoftMsgpack/MsgPack/MessagePackParser.cs

@@ -1,5 +1,6 @@
 using System;
 using System.IO;
+using System.Text;
 using WampSharp.Logging;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
@@ -90,6 +91,11 @@ public byte[] Format(WampMessage<object> message)
             }
         }
 
+        public byte[] GetBytes(byte[] raw)
+        {
+            return raw;
+        }
+
         public WampMessage<JToken> Parse(Stream stream)
         {
             try

src/net45/Extensions/WampSharp.AspNetCore.RawSocket/AspNetCoreRawSocketTransport.cs

@@ -16,10 +16,7 @@ public class AspNetCoreRawSocketTransport : TextBinaryTransport<SocketData>
         private Func<ConnectionContext, Func<Task>, Task> mHandler;
         private readonly Handshaker mHandshaker = new Handshaker();
 
-        public byte MaxSize
-        {
-            get;
-        }
+        public byte MaxSize { get; }
 
         public AspNetCoreRawSocketTransport(IConnectionBuilder app,
                                             byte maxSize = 15,
@@ -61,22 +58,27 @@ protected override string GetSubProtocol(SocketData connection)
             return connection.HandshakeResponse.SerializerType.GetSubProtocol();
         }
 
-        protected override IWampConnection<TMessage> CreateBinaryConnection<TMessage>
-        (SocketData connection,
-            IWampBinaryBinding<TMessage> binding)
+        protected override IWampConnection<TMessage> CreateBinaryConnection<TMessage>(SocketData connection,
+                                                                                      IWampBinaryBinding<TMessage>
+                                                                                          binding)
         {
             return CreateConnection(connection, binding);
         }
 
-        protected override IWampConnection<TMessage> CreateTextConnection<TMessage>
-        (SocketData connection,
-            IWampTextBinding<TMessage> binding)
+        protected override IWampConnection<TMessage> CreateTextConnection<TMessage>(SocketData connection,
+                                                                                    IWampTextBinding<TMessage> binding)
         {
             return CreateConnection(connection, binding);
         }
 
-        private IWampConnection<TMessage> CreateConnection<TMessage>(SocketData connection, IWampStreamingMessageParser<TMessage> binding)
+        private IWampConnection<TMessage> CreateConnection<TMessage, TRaw>(SocketData connection,
+                                                                           IWampTransportBinding<TMessage, TRaw> binding)
         {
+            if (binding.ComputeBytes == null)
+            {
+                binding.ComputeBytes = true;
+            }
+
             return new RawSocketConnection<TMessage>(connection, binding, mAutoPingInterval);
         }
 

src/net45/Extensions/WampSharp.RawSocket/RawSocket/RawSocketTransport.cs

@@ -144,8 +144,13 @@ protected override IWampConnection<TMessage> CreateTextConnection<TMessage>
             return CreateConnection(connection, binding);
         }
 
-        private TcpClientConnection<TMessage> CreateConnection<TMessage>(RawSocketTcpClient connection, IWampStreamingMessageParser<TMessage> binding)
+        private TcpClientConnection<TMessage> CreateConnection<TMessage, TRaw>(RawSocketTcpClient connection, IWampTransportBinding<TMessage, TRaw> binding)
         {
+            if (binding.ComputeBytes == null)
+            {
+                binding.ComputeBytes = true;
+            }
+
             return new TcpClientConnection<TMessage>
             (connection.Client,
                 connection.Stream,

src/net45/WampSharp/WAMP2/V2/Authentication/Host/WampAuthenticationBinaryBinding.cs

@@ -28,6 +28,11 @@ public byte[] Format(WampMessage<object> message)
             return mBinding.Format(message);
         }
 
+        public byte[] GetBytes(byte[] raw)
+        {
+            return mBinding.GetBytes(raw);
+        }
+
         public WampMessage<TMessage> Parse(Stream stream)
         {
             return mBinding.Parse(stream);
@@ -37,5 +42,11 @@ public void Format(WampMessage<object> message, Stream stream)
         {
             mBinding.Format(message, stream);
         }
+
+        public bool? ComputeBytes
+        {
+            get => mBinding.ComputeBytes;
+            set => mBinding.ComputeBytes = value;
+        }
     }
 }

src/net45/WampSharp/WAMP2/V2/Authentication/Host/WampAuthenticationTextBinding.cs

@@ -28,6 +28,11 @@ public string Format(WampMessage<object> message)
             return mBinding.Format(message);
         }
 
+        public byte[] GetBytes(string raw)
+        {
+            return mBinding.GetBytes(raw);
+        }
+
         public WampMessage<TMessage> Parse(Stream stream)
         {
             return mBinding.Parse(stream);
@@ -37,5 +42,11 @@ public void Format(WampMessage<object> message, Stream stream)
         {
             mBinding.Format(message, stream);
         }
+
+        public bool? ComputeBytes
+        {
+            get => mBinding.ComputeBytes;
+            set => mBinding.ComputeBytes = value;
+        }
     }
 }

src/net45/WampSharp/WAMP2/V2/Binding/IWampTransportBinding.cs

@@ -11,5 +11,6 @@ public interface IWampTransportBinding<TMessage, TRaw> :
         IWampBinding<TMessage>,
         IWampMessageParser<TMessage, TRaw>
     {
+        bool? ComputeBytes { get; set; }
     }
 }

src/net45/WampSharp/WAMP2/V2/Binding/Messages/RawMessage.cs

@@ -9,5 +9,7 @@ public RawMessage(WampMessage<object> other) : base(other)
         }
 
         public TRaw Raw { get; set; }
+
+        public byte[] Bytes { get; set; }
     }
 }

src/net45/WampSharp/WAMP2/V2/Binding/Parsers/IWampMessageParser.cs

@@ -21,5 +21,12 @@ public interface IWampMessageParser<TMessage, TRaw> : IWampStreamingMessageParse
         /// <param name="message">The given <see cref="WampMessage{TMessage}"/>.</param>
         /// <returns>A raw format representing the given <see cref="WampMessage{TMessage}"/>.</returns>
         TRaw Format(WampMessage<object> message);
+
+        /// <summary>
+        /// Serializes a raw message to bytes.
+        /// </summary>
+        /// <param name="raw">The given raw message.</param>
+        /// <returns>Bytes representing the binary format of the given raw message.</returns>
+        byte[] GetBytes(TRaw raw);
     }
 }

src/net45/WampSharp/WAMP2/V2/Binding/WampTransportBinding.cs

@@ -33,19 +33,28 @@ public TRaw Format(WampMessage<object> message)
             return textMessage.Raw;
         }
 
+        public byte[] GetBytes(TRaw raw)
+        {
+            return mParser.GetBytes(raw);
+        }
+
         public override WampMessage<object> GetRawMessage(WampMessage<object> message)
         {
             return GetFormattedMessage(message);
         }
 
         private RawMessage<TRaw> GetFormattedMessage(WampMessage<object> message)
         {
-            RawMessage<TRaw> result = message as RawMessage<TRaw>;
 
-            if (result == null)
+            if (!(message is RawMessage<TRaw> result))
             {
                 result = new RawMessage<TRaw>(message);
                 result.Raw = mParser.Format(message);
+
+                if (ComputeBytes == true)
+                {
+                    result.Bytes = mParser.GetBytes(result.Raw);
+                }
             }
 
             return result;
@@ -58,8 +67,18 @@ public WampMessage<TMessage> Parse(Stream stream)
 
         public void Format(WampMessage<object> message, Stream stream)
         {
-            // TODO: You know! reuse the RawMessage if possible :)
-            mParser.Format(message, stream);
+            if (ComputeBytes == true && 
+                message is RawMessage<TRaw> casted)
+            {
+                byte[] bytes = casted.Bytes;
+                stream.Write(bytes, 0, bytes.Length);
+            }
+            else
+            {
+                mParser.Format(message, stream);
+            }
         }
+
+        public bool? ComputeBytes { get; set; }
     }
 }