WebSocket Binary Message Support (Chunked Stream) Added

Gencebay Demir · Gencebay Demir · commit 49237619aab2 · 2016-12-08T01:08:27.000+03:00
diff --git a/src/NetCoreStack.WebSockets.ProxyClient/ApplicationBuilderExtensions.cs b/src/NetCoreStack.WebSockets.ProxyClient/ApplicationBuilderExtensions.cs
@@ -19,7 +19,7 @@ public static IApplicationBuilder UseProxyWebSockets(this IApplicationBuilder ap
             if (webSocketConnector != null && appLifeTime != null)
             {
                 appLifeTime.ApplicationStopping.Register(OnShutdown, webSocketConnector);
-                Task.Run(async () => await webSocketConnector.InitializeAsync());
+                Task.Run(async () => await webSocketConnector.ConnectAsync());
             }
 
             return app;
diff --git a/src/NetCoreStack.WebSockets.ProxyClient/ClientWebSocketConnector.cs b/src/NetCoreStack.WebSockets.ProxyClient/ClientWebSocketConnector.cs
@@ -1,5 +1,7 @@
 ﻿using Microsoft.Extensions.Options;
 using System;
+using System.IO;
+using System.Linq;
 using System.Net.WebSockets;
 using System.Threading;
 using System.Threading.Tasks;
@@ -13,38 +15,82 @@ internal class ClientWebSocketConnector : IWebSocketConnector
         private readonly ConnectorOptions _options;
         private readonly InvocatorRegistry _invocatorRegistry;
 
-        public ClientWebSocketConnector(IOptions<ConnectorOptions> options, InvocatorRegistry invocatorRegistry)
+        public ClientWebSocketConnector(IOptions<ConnectorOptions> options, 
+            InvocatorRegistry invocatorRegistry)
         {
             _options = options.Value;
             _invocatorRegistry = invocatorRegistry;
         }
 
-        public async Task InitializeAsync()
+        private async Task Receive()
         {
-            var uri = new Uri($"ws://{_options.WebSocketHostAddress}");
-            _webSocket = new ClientWebSocket();
-            await _webSocket.ConnectAsync(uri, CancellationToken.None);
-            var buffer = new byte[1024 * 4];
+            var buffer = new byte[SocketsConstants.ChunkSize];
             var result = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
             while (!result.CloseStatus.HasValue)
             {
-                var context = result.ToContext(buffer);
-                if (context.Command == WebSocketCommands.Handshake)
-                    _connectionId = context.Value.ToString();
+                if (result.MessageType == WebSocketMessageType.Text)
+                {
+                    var context = result.ToContext(buffer);
+                    if (context.Command == WebSocketCommands.Handshake)
+                        _connectionId = context.Value.ToString();
+
+                    var _invocators = _invocatorRegistry.GetInvocators(context);
+                    _invocators.ForEach(async x => await x.InvokeAsync(context));
+                    result = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
+                }
 
-                var _invocators = _invocatorRegistry.GetInvocators(context);
-                _invocators.ForEach(async x => await x.InvokeAsync(context));
-                result = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
+                if (result.MessageType == WebSocketMessageType.Binary)
+                {
+                    byte[] binaryResult = null;
+                    using (var ms = new MemoryStream())
+                    {
+                        while (!result.EndOfMessage)
+                        {
+                            if (!result.CloseStatus.HasValue)
+                            {
+                                await ms.WriteAsync(buffer, 0, result.Count);
+                            }
+                            result = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
+                        }
+                        if (result.EndOfMessage)
+                        {
+                            if (!result.CloseStatus.HasValue)
+                            {
+                                await ms.WriteAsync(buffer, 0, result.Count);
+                            }
+                        }
+                        binaryResult = ms.ToArray();
+                    }
+                    var context = result.ToBinaryContext(binaryResult);
+                    var _invocators = _invocatorRegistry.GetInvocators(context);
+                    _invocators.ForEach(async x => await x.InvokeAsync(context));
+                    result = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
+                }
             }
             await _webSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None);
         }
 
+        public async Task ConnectAsync()
+        {
+            var uri = new Uri($"ws://{_options.WebSocketHostAddress}");
+            _webSocket = new ClientWebSocket();
+            await _webSocket.ConnectAsync(uri, CancellationToken.None);
+            await Task.WhenAll(Receive());
+        }
+
         public async Task SendAsync(WebSocketMessageContext context)
         {
             var segments = context.ToSegment();
             await _webSocket.SendAsync(segments, WebSocketMessageType.Text, true, CancellationToken.None);
         }
 
+        public async Task SendBinaryAsync(byte[] bytes)
+        {
+            // TODO Chunked
+            var segments = new ArraySegment<byte>(bytes, 0, bytes.Count());
+            await _webSocket.SendAsync(segments, WebSocketMessageType.Binary, true, CancellationToken.None);
+        }
+
         internal void Close(string statusDescription)
         {
             _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, statusDescription, CancellationToken.None);
diff --git a/src/NetCoreStack.WebSockets.ProxyClient/IWebSocketConnector.cs b/src/NetCoreStack.WebSockets.ProxyClient/IWebSocketConnector.cs
@@ -4,7 +4,8 @@ namespace NetCoreStack.WebSockets.ProxyClient
 {
     public interface IWebSocketConnector
     {
-        Task InitializeAsync();
+        Task ConnectAsync();
         Task SendAsync(WebSocketMessageContext context);
+        Task SendBinaryAsync(byte[] bytes);
     }
 }
diff --git a/src/NetCoreStack.WebSockets/Internal/ConnectionManager.cs b/src/NetCoreStack.WebSockets/Internal/ConnectionManager.cs
@@ -1,7 +1,11 @@
 ﻿using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
 using System;
 using System.Collections.Concurrent;
+using System.IO;
+using System.Linq;
 using System.Net.WebSockets;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -81,6 +85,48 @@ public async Task SendAsync(string connectionId, WebSocketMessageContext context
             await SendAsync(transport, descriptor);
         }
 
+        public async Task SendBinaryAsync(string connectionId, byte[] bytes, JsonObject properties)
+        {
+            WebSocketTransport transport = null;
+            if (!Connections.TryGetValue(connectionId, out transport))
+            {
+                throw new ArgumentOutOfRangeException(nameof(transport));
+            }
+
+            var props = JsonConvert.SerializeObject(properties);
+            var propsBytes = Encoding.UTF8.GetBytes($"{SocketsConstants.Splitter}{props}");
+
+            var bytesCount = bytes.Length;
+            bytes = bytes.Concat(propsBytes).ToArray();
+
+            var buffer = new byte[SocketsConstants.ChunkSize];
+            using (var ms = new MemoryStream(bytes))
+            {
+                using (BinaryReader br = new BinaryReader(ms))
+                {
+                    byte[] chunkBytes = null;
+                    do
+                    {
+                        chunkBytes = br.ReadBytes(SocketsConstants.ChunkSize);
+                        var segments = new ArraySegment<byte>(chunkBytes);
+                        var endOfMessage = false;
+
+                        if (chunkBytes.Length < SocketsConstants.ChunkSize)
+                            endOfMessage = true;
+
+                        await transport.WebSocket.SendAsync(segments, 
+                            WebSocketMessageType.Binary, 
+                            endOfMessage, 
+                            CancellationToken.None);
+
+                        if (endOfMessage)
+                            break;
+
+                    } while (chunkBytes.Length <= SocketsConstants.ChunkSize);
+                }
+            }
+        }
+
         public async Task SendAsync(string connectionId, WebSocketMessageContext context, WebSocket webSocket)
         {
             if (string.IsNullOrEmpty(connectionId))
diff --git a/src/NetCoreStack.WebSockets/Internal/IConnectionManager.cs b/src/NetCoreStack.WebSockets/Internal/IConnectionManager.cs
@@ -1,5 +1,4 @@
-﻿using NetCoreStack.WebSockets;
-using System.Net.WebSockets;
+﻿using System.Net.WebSockets;
 using System.Threading.Tasks;
 
 namespace NetCoreStack.WebSockets.Internal
@@ -10,6 +9,8 @@ public interface IConnectionManager
 
         Task SendAsync(string connectionId, WebSocketMessageContext context);
 
+        Task SendBinaryAsync(string connectionId, byte[] bytes, JsonObject properties);
+
         Task SendAsync(string connectionId, WebSocketMessageContext context, WebSocket webSocket);
         void CloseConnection(string connectionId);
     }
diff --git a/src/NetCoreStack.WebSockets/Internal/SocketsConstants.cs b/src/NetCoreStack.WebSockets/Internal/SocketsConstants.cs
@@ -0,0 +1,8 @@
+﻿namespace NetCoreStack.WebSockets
+{
+    public class SocketsConstants
+    {
+        public const string Splitter = "<<";
+        public const int ChunkSize = 1024 * 4;
+    }
+}
diff --git a/src/NetCoreStack.WebSockets/NetCoreStack.WebSockets.xproj b/src/NetCoreStack.WebSockets/NetCoreStack.WebSockets.xproj
@@ -4,18 +4,16 @@
     <VisualStudioVersion Condition="'$(VisualStudioVersion)' == ''">14.0</VisualStudioVersion>
     <VSToolsPath Condition="'$(VSToolsPath)' == ''">$(MSBuildExtensionsPath32)\Microsoft\VisualStudio\v$(VisualStudioVersion)</VSToolsPath>
   </PropertyGroup>
-
   <Import Project="$(VSToolsPath)\DotNet\Microsoft.DotNet.Props" Condition="'$(VSToolsPath)' != ''" />
   <PropertyGroup Label="Globals">
     <ProjectGuid>c6bd4266-256a-4ff6-b991-d44efc825fc6</ProjectGuid>
-    <RootNamespace>NetCoreStack.WebSockets.Common</RootNamespace>
+    <RootNamespace>NetCoreStack.WebSockets</RootNamespace>
     <BaseIntermediateOutputPath Condition="'$(BaseIntermediateOutputPath)'=='' ">.\obj</BaseIntermediateOutputPath>
     <OutputPath Condition="'$(OutputPath)'=='' ">.\bin\</OutputPath>
     <TargetFrameworkVersion>v4.6</TargetFrameworkVersion>
   </PropertyGroup>
-
   <PropertyGroup>
     <SchemaVersion>2.0</SchemaVersion>
   </PropertyGroup>
   <Import Project="$(VSToolsPath)\DotNet\Microsoft.DotNet.targets" Condition="'$(VSToolsPath)' != ''" />
-</Project>
+</Project>
diff --git a/src/NetCoreStack.WebSockets/Types/JsonObject.cs b/src/NetCoreStack.WebSockets/Types/JsonObject.cs
@@ -0,0 +1,20 @@
+﻿using System.Collections.Generic;
+
+namespace NetCoreStack.WebSockets
+{
+    public abstract class JsonObject
+    {
+        public JsonObject()
+        {
+        }
+
+        public IDictionary<string, object> ToJson()
+        {
+            Dictionary<string, object> dictionary = new Dictionary<string, object>();
+            Serialize(dictionary);
+            return dictionary;
+        }
+
+        protected abstract void Serialize(IDictionary<string, object> value);
+    }
+}
diff --git a/src/NetCoreStack.WebSockets/Types/SocketObject.cs b/src/NetCoreStack.WebSockets/Types/SocketObject.cs
@@ -0,0 +1,18 @@
+﻿using System;
+using System.Collections.Generic;
+
+namespace NetCoreStack.WebSockets
+{
+    public class SocketObject : JsonObject
+    {
+        public string Key { get; set; }
+
+        public object Value { get; set; }
+
+        protected override void Serialize(IDictionary<string, object> value)
+        {
+            value.Add(nameof(Key), Key);
+            value.Add(nameof(Value), Value);
+        }
+    }
+}
diff --git a/src/NetCoreStack.WebSockets/WebSocketExtensions.cs b/src/NetCoreStack.WebSockets/WebSocketExtensions.cs
@@ -1,24 +1,51 @@
 ﻿using Newtonsoft.Json;
 using System;
+using System.Linq;
 using System.Net.WebSockets;
 using System.Text;
 
 namespace NetCoreStack.WebSockets
 {
     public static class WebSocketExtensions
     {
-        public static WebSocketMessageContext ToContext(this WebSocketReceiveResult result, byte[] buffer)
+        public static WebSocketMessageContext ToContext(this WebSocketReceiveResult result, byte[] values)
         {
             if (result == null)
             {
                 throw new ArgumentNullException(nameof(result));
             }
             
-            var content = Encoding.UTF8.GetString(buffer, 0, result.Count);
+            var content = Encoding.UTF8.GetString(values, 0, result.Count);
             var webSocketContext = JsonConvert.DeserializeObject<WebSocketMessageContext>(content);
             return webSocketContext;
         }
 
+        public static WebSocketMessageContext ToBinaryContext(this WebSocketReceiveResult result, byte[] values)
+        {
+            if (result == null)
+            {
+                throw new ArgumentNullException(nameof(result));
+            }
+            
+            var webSocketContext = new WebSocketMessageContext();
+            var content = Encoding.UTF8.GetString(values);
+            if (content != null)
+            {
+                string[] parts = content.Split(new string[] { SocketsConstants.Splitter }, StringSplitOptions.None);
+                if (parts.Length != 2)
+                {
+                    throw new InvalidOperationException(nameof(parts));
+                }
+
+                webSocketContext.Value = parts.First();
+                webSocketContext.State = JsonConvert.DeserializeObject<SocketObject>(parts.Last());
+            }
+
+            webSocketContext.MessageType = WebSocketMessageType.Binary;
+            webSocketContext.Command = WebSocketCommands.DataSend;
+            return webSocketContext;
+        }
+
         public static ArraySegment<byte> ToSegment(this WebSocketMessageContext webSocketContext)
         {
             if (webSocketContext == null)
diff --git a/src/NetCoreStack.WebSockets/WebSocketMessageContext.cs b/src/NetCoreStack.WebSockets/WebSocketMessageContext.cs
@@ -1,7 +1,10 @@
-﻿namespace NetCoreStack.WebSockets
+﻿using System.Net.WebSockets;
+
+namespace NetCoreStack.WebSockets
 {
     public class WebSocketMessageContext
     {
+        public WebSocketMessageType MessageType { get; set; }
         public WebSocketCommands? Command { get; set; }
         public object Value { get; set; }
         public object State { get; set; }
diff --git a/test/ServerTestApp/Controllers/DiscoveryController.cs b/test/ServerTestApp/Controllers/DiscoveryController.cs
@@ -1,4 +1,6 @@
 ﻿using Microsoft.AspNetCore.Mvc;
+using Microsoft.Extensions.Caching.Distributed;
+using Microsoft.Extensions.Caching.Memory;
 using Microsoft.Extensions.Logging;
 using NetCoreStack.WebSockets;
 using NetCoreStack.WebSockets.Internal;
@@ -13,11 +15,17 @@ public class DiscoveryController : Controller
     {
         private readonly ILoggerFactory _loggerFactory;
         private readonly IConnectionManager _connectionManager;
+        private readonly IDistributedCache _distrubutedCache;
+        private readonly IMemoryCache _memoryCache;
 
         public DiscoveryController(IConnectionManager connectionManager, 
+            IDistributedCache distrubutedCache,
+            IMemoryCache memoryCache,
             ILoggerFactory loggerFactory)
         {
             _connectionManager = connectionManager;
+            _distrubutedCache = distrubutedCache;
+            _memoryCache = memoryCache;
             _loggerFactory = loggerFactory;
         }
 
@@ -36,5 +44,21 @@ public async Task<IActionResult> SendAsync([FromBody]SimpleModel model)
             await _connectionManager.BroadcastAsync(webSocketContext);
             return Ok();
         }
+
+        [HttpPost(nameof(SendBinaryAsync))]
+        public async Task<IActionResult> SendBinaryAsync([FromBody]SimpleModel model)
+        {
+            var bytes = _distrubutedCache.Get(model.Message);
+            await _connectionManager.SendBinaryAsync(model.ConnectionId, bytes, new SocketObject { Key = model.Message });
+            return Ok();
+        }
+
+        [HttpPost(nameof(SendBinaryFromMemoryAsync))]
+        public async Task<IActionResult> SendBinaryFromMemoryAsync([FromBody]SimpleModel model)
+        {
+            var bytes = (byte[])_memoryCache.Get(model.Message);
+            await _connectionManager.SendBinaryAsync(model.ConnectionId, bytes, new SocketObject { Key = model.Message });
+            return Ok();
+        }
     }
 }
diff --git a/test/ServerTestApp/Startup.cs b/test/ServerTestApp/Startup.cs
@@ -39,6 +39,14 @@ public void ConfigureServices(IServiceCollection services)
                 c.DescribeAllEnumsAsStrings();
             });
 
+            services.AddMemoryCache();
+
+            services.AddDistributedRedisCache(options =>
+            {
+                options.Configuration = "localhost";
+                options.InstanceName = "SocketsInstance";
+            });
+
             services.AddTransient<IHandshakeStateTransport, MyHandshakeStateTransport>();
 
             // Add Net Core Stack socket services.
diff --git a/test/ServerTestApp/project.json b/test/ServerTestApp/project.json
diff --git a/test/WebClientTestApp/CustomWebSocketCommandInvocator.cs b/test/WebClientTestApp/CustomWebSocketCommandInvocator.cs

Original file line number	Diff line number	Diff line change
`@@ -19,7 +19,7 @@ public static IApplicationBuilder UseProxyWebSockets(this IApplicationBuilder ap`
`19`	`19`	`if (webSocketConnector != null && appLifeTime != null)`
`20`	`20`	`{`
`21`	`21`	`appLifeTime.ApplicationStopping.Register(OnShutdown, webSocketConnector);`
`22`		`- Task.Run(async () => await webSocketConnector.InitializeAsync());`
	`22`	`+ Task.Run(async () => await webSocketConnector.ConnectAsync());`
`23`	`23`	`}`
`24`	`24`
`25`	`25`	`return app;`
Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,8 @@ namespace NetCoreStack.WebSockets.ProxyClient`
`4`	`4`	`{`
`5`	`5`	`public interface IWebSocketConnector`
`6`	`6`	`{`
`7`		`- Task InitializeAsync();`
	`7`	`+ Task ConnectAsync();`
`8`	`8`	`Task SendAsync(WebSocketMessageContext context);`
	`9`	`+ Task SendBinaryAsync(byte[] bytes);`
`9`	`10`	`}`
`10`	`11`	`}`