Support ReadOnlySequence type (#146)

Author: Odonno

src/Websocket.Client/RequestMessage.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Buffers;
 
 namespace Websocket.Client
 {
@@ -33,4 +34,14 @@ public RequestBinarySegmentMessage(ArraySegment<byte> data)
             Data = data;
         }
     }
+
+    internal class RequestBinarySequenceMessage : RequestMessage
+    {
+        public ReadOnlySequence<byte> Data { get; }
+
+        public RequestBinarySequenceMessage(ReadOnlySequence<byte> data)
+        {
+            Data = data;
+        }
+    }
 }

src/Websocket.Client/WebsocketClient.Sending.cs

@@ -1,5 +1,6 @@
 using Microsoft.Extensions.Logging;
 using System;
+using System.Buffers;
 using System.Net.WebSockets;
 using System.Runtime.InteropServices;
 using System.Threading;
@@ -15,12 +16,13 @@ public partial class WebsocketClient
             SingleReader = true,
             SingleWriter = false
         });
-        private readonly Channel<ArraySegment<byte>> _messagesBinaryToSendQueue = Channel.CreateUnbounded<ArraySegment<byte>>(new UnboundedChannelOptions()
+        private readonly Channel<ReadOnlySequence<byte>> _messagesBinaryToSendQueue = Channel.CreateUnbounded<ReadOnlySequence<byte>>(new UnboundedChannelOptions()
         {
             SingleReader = true,
             SingleWriter = false
         });
 
+        private static readonly byte[] EMPTY_ARRAY = { };
 
         /// <summary>
         /// Send text message to the websocket channel. 
@@ -45,7 +47,7 @@ public bool Send(byte[] message)
         {
             Validations.Validations.ValidateInput(message, nameof(message));
 
-            return _messagesBinaryToSendQueue.Writer.TryWrite(new ArraySegment<byte>(message));
+            return _messagesBinaryToSendQueue.Writer.TryWrite(new ReadOnlySequence<byte>(message));
         }
 
         /// <summary>
@@ -58,6 +60,19 @@ public bool Send(ArraySegment<byte> message)
         {
             Validations.Validations.ValidateInput(message, nameof(message));
 
+            return _messagesBinaryToSendQueue.Writer.TryWrite(new ReadOnlySequence<byte>(message));
+        }
+
+        /// <summary>
+        /// Send binary message to the websocket channel. 
+        /// It inserts the message to the queue and actual sending is done on another thread
+        /// </summary>
+        /// <param name="message">Binary message to be sent</param>
+        /// <returns>true if the message was written to the queue</returns>
+        public bool Send(ReadOnlySequence<byte> message)
+        {
+            Validations.Validations.ValidateInput(message, nameof(message));
+
             return _messagesBinaryToSendQueue.Writer.TryWrite(message);
         }
 
@@ -84,7 +99,7 @@ public Task SendInstant(string message)
         /// <param name="message">Message to be sent</param>
         public Task SendInstant(byte[] message)
         {
-            return SendInternalSynchronized(new ArraySegment<byte>(message));
+            return SendInternalSynchronized(message);
         }
 
         /// <summary>
@@ -115,6 +130,20 @@ public bool SendAsText(ArraySegment<byte> message)
             return _messagesTextToSendQueue.Writer.TryWrite(new RequestBinarySegmentMessage(message));
         }
 
+        /// <summary>
+        /// Send already converted text message to the websocket channel. 
+        /// Use this method to avoid double serialization of the text message.
+        /// It inserts the message to the queue and actual sending is done on another thread
+        /// </summary>
+        /// <param name="message">Message to be sent</param>
+        /// <returns>true if the message was written to the queue</returns>
+        public bool SendAsText(ReadOnlySequence<byte> message)
+        {
+            Validations.Validations.ValidateInput(message, nameof(message));
+
+            return _messagesTextToSendQueue.Writer.TryWrite(new RequestBinarySequenceMessage(message));
+        }
+
         /// <summary>
         /// Stream/publish fake message (via 'MessageReceived' observable).
         /// Use for testing purposes to simulate a server message. 
@@ -230,57 +259,109 @@ private async Task SendInternalSynchronized(RequestMessage message)
 
         private async Task SendInternal(RequestMessage message)
         {
-            if (!IsClientConnected())
-            {
-                _logger.LogDebug(L("Client is not connected to server, cannot send: {message}"), Name, message);
-                return;
-            }
-
-            _logger.LogTrace(L("Sending: {message}"), Name, message);
-
-            ReadOnlyMemory<byte> payload;
-
             switch (message)
             {
                 case RequestTextMessage textMessage:
-                    payload = MemoryMarshal.AsMemory<byte>(GetEncoding().GetBytes(textMessage.Text));
+                    await SendTextMessage(textMessage).ConfigureAwait(false);
                     break;
                 case RequestBinaryMessage binaryMessage:
-                    payload = MemoryMarshal.AsMemory<byte>(binaryMessage.Data);
+                    await SendBinaryMessage(binaryMessage).ConfigureAwait(false);
                     break;
                 case RequestBinarySegmentMessage segmentMessage:
-                    payload = segmentMessage.Data.AsMemory();
+                    await SendBinarySegmentMessage(segmentMessage).ConfigureAwait(false);
+                    break;
+                case RequestBinarySequenceMessage sequenceMessage:
+                    await SendBinarySequenceMessage(sequenceMessage).ConfigureAwait(false);
                     break;
                 default:
                     throw new ArgumentException($"Unknown message type: {message.GetType()}");
             }
+        }
+        
+        private async Task SendTextMessage(RequestTextMessage textMessage)
+        {
+            var payload = MemoryMarshal.AsMemory<byte>(GetEncoding().GetBytes(textMessage.Text));
+            await SendInternal(payload, WebSocketMessageType.Text).ConfigureAwait(false);
+        }
 
-            await _client!
-                .SendAsync(payload, WebSocketMessageType.Text, true, _cancellation?.Token ?? CancellationToken.None)
-                .ConfigureAwait(false);
+        private async Task SendBinaryMessage(RequestBinaryMessage binaryMessage)
+        {
+            var payload = MemoryMarshal.AsMemory<byte>(binaryMessage.Data);
+            await SendInternal(payload, WebSocketMessageType.Text).ConfigureAwait(false);
         }
 
-        private async Task SendInternalSynchronized(ArraySegment<byte> message)
+        private async Task SendBinarySegmentMessage(RequestBinarySegmentMessage segmentMessage)
+        {
+            await SendInternal(segmentMessage.Data, WebSocketMessageType.Text).ConfigureAwait(false);
+        }
+
+        private async Task SendBinarySequenceMessage(RequestBinarySequenceMessage sequenceMessage)
+        {
+            await SendInternal(sequenceMessage.Data, WebSocketMessageType.Text).ConfigureAwait(false);
+        }
+
+        private async Task SendInternalSynchronized(ReadOnlySequence<byte> message)
         {
             using (await _locker.LockAsync())
             {
-                await SendInternal(message);
+                await SendInternal(message, WebSocketMessageType.Binary);
+            }
+        }
+        private async Task SendInternalSynchronized(ReadOnlyMemory<byte> message)
+        {
+            using (await _locker.LockAsync())
+            {
+                await SendInternal(message, WebSocketMessageType.Binary);
             }
         }
 
-        private async Task SendInternal(ArraySegment<byte> payload)
+        private async Task SendInternal(ReadOnlySequence<byte> payload, WebSocketMessageType messageType)
         {
-            if (!IsClientConnected())
+            if (payload.IsSingleSegment)
+            {
+                await SendInternal(payload.First, messageType).ConfigureAwait(false);
+                return;
+            }
+
+            if (!BeforeSendInternal(payload.Length))
             {
-                _logger.LogDebug(L("Client is not connected to server, cannot send binary, length: {length}"), Name, payload.Count);
                 return;
             }
 
-            _logger.LogTrace(L("Sending binary, length: {length}"), Name, payload.Count);
+            foreach (var memory in payload)
+            {
+                await _client!
+                    .SendAsync(memory, messageType, false, _cancellation?.Token ?? CancellationToken.None)
+                    .ConfigureAwait(false);
+            }
+
+            await _client!
+                .SendAsync(EMPTY_ARRAY, messageType, true, _cancellation?.Token ?? CancellationToken.None)
+                .ConfigureAwait(false);
+        }
+
+        private async Task SendInternal(ReadOnlyMemory<byte> payload, WebSocketMessageType messageType)
+        {
+            if (!BeforeSendInternal(payload.Length))
+            {
+                return;
+            }
 
             await _client!
-                .SendAsync(payload, WebSocketMessageType.Binary, true, _cancellation?.Token ?? CancellationToken.None)
+                .SendAsync(payload, messageType, true, _cancellation?.Token ?? CancellationToken.None)
                 .ConfigureAwait(false);
         }
+
+        private bool BeforeSendInternal(long length)
+        {
+            if (!IsClientConnected())
+            {
+                _logger.LogDebug(L("Client is not connected to server, cannot send binary, length: {length}"), Name, length);
+                return false;
+            }
+
+            _logger.LogTrace(L("Sending binary, length: {length}"), Name, length);
+            return true;
+        }
     }
 }

src/Websocket.Client/WebsocketClient.cs

@@ -458,7 +458,7 @@ private async Task Listen(WebSocket client, CancellationToken token)
                 do
                 {
                     ValueWebSocketReceiveResult result;
-                    var ms = (RecyclableMemoryStream)_memoryStreamManager.GetStream();
+                    var ms = _memoryStreamManager.GetStream();
 
                     while (true)
                     {