Compression fix, GZip check default

Author: gedem

src/NetCoreStack.WebSockets/ConnectionManager.cs

@@ -39,7 +39,7 @@ public ConnectionManager(IStreamCompressor compressor,
             Connections = new ConcurrentDictionary<string, WebSocketTransport>();
         }
 
-        private async Task<byte[]> PrepareBytesAsync(byte[] input, IDictionary<string, object> properties, bool compression = true)
+        private async Task<byte[]> PrepareBytesAsync(byte[] input, IDictionary<string, object> properties)
         {
             if (input == null)
             {
@@ -49,19 +49,23 @@ private async Task<byte[]> PrepareBytesAsync(byte[] input, IDictionary<string, o
             if (properties == null)
                 properties = new Dictionary<string, object>();
 
-            object compressedKey = null;
-            if (properties.TryGetValue(CompressedKey, out compressedKey))
-                properties[CompressedKey] = compression;
+            bool compressed = GZipHelper.IsGZipHeader(input);
+
+            object key = null;
+            if (properties.TryGetValue(CompressedKey, out key))
+                properties[CompressedKey] = compressed;
             else
-                properties.Add(CompressedKey, compression);
+                properties.Add(CompressedKey, compressed);
+
+            string props = JsonConvert.SerializeObject(properties);
+            byte[] header = Encoding.UTF8.GetBytes($"{props}{Splitter}");
 
-            var props = JsonConvert.SerializeObject(properties);
-            var propsBytes = Encoding.UTF8.GetBytes($"{props}{Splitter}");
+            if (compressed)
+                header = await _compressor.CompressAsync(header);
 
-            var bytesCount = input.Length;
-            input = propsBytes.Concat(input).ToArray();
+            input = header.Concat(input).ToArray();
 
-            if (compression)
+            if (!compressed)
                 return await _compressor.CompressAsync(input);
 
             return input;
@@ -156,14 +160,14 @@ public async Task BroadcastAsync(WebSocketMessageContext context)
             }
         }
 
-        public async Task BroadcastBinaryAsync(byte[] inputs, IDictionary<string, object> properties, bool compression = true)
+        public async Task BroadcastBinaryAsync(byte[] inputs, IDictionary<string, object> properties)
         {
             if (!Connections.Any())
             {
                 return;
             }
 
-            var bytes = await PrepareBytesAsync(inputs, properties, compression);
+            var bytes = await PrepareBytesAsync(inputs, properties);
 
             using (var ms = new MemoryStream(bytes))
             {
@@ -215,7 +219,7 @@ public async Task SendAsync(string connectionId, WebSocketMessageContext context
             await SendAsync(transport, descriptor);
         }
 
-        public async Task SendBinaryAsync(string connectionId, byte[] input, IDictionary<string, object> properties, bool compression = true)
+        public async Task SendBinaryAsync(string connectionId, byte[] input, IDictionary<string, object> properties)
         {
             if (!Connections.Any())
             {
@@ -228,7 +232,7 @@ public async Task SendBinaryAsync(string connectionId, byte[] input, IDictionary
                 throw new ArgumentOutOfRangeException(nameof(transport));
             }
 
-            byte[] bytes = await PrepareBytesAsync(input, properties, compression);
+            byte[] bytes = await PrepareBytesAsync(input, properties);
 
             using (var ms = new MemoryStream(bytes))
             {

src/NetCoreStack.WebSockets/Interfaces/IConnectionManager.cs

@@ -24,9 +24,8 @@ public interface IConnectionManager
         /// </summary>
         /// <param name="inputs"></param>
         /// <param name="properties"></param>
-        /// <param name="compression"></param>
         /// <returns></returns>
-        Task BroadcastBinaryAsync(byte[] input, IDictionary<string, object> properties, bool compression = true);
+        Task BroadcastBinaryAsync(byte[] input, IDictionary<string, object> properties);
 
         /// <summary>
         /// Send text message to specified connection
@@ -44,7 +43,7 @@ public interface IConnectionManager
         /// <param name="properties">Additional transport data</param>
         /// <param name="compression">Compression status of the data, default value is true</param>
         /// <returns></returns>
-        Task SendBinaryAsync(string connectionId, byte[] input, IDictionary<string, object> properties, bool compression = true);
+        Task SendBinaryAsync(string connectionId, byte[] input, IDictionary<string, object> properties);
 
         /// <summary>
         /// Close the specified connection

test/ServerTestApp/Controllers/DiscoveryController.cs

@@ -107,10 +107,10 @@ public async Task<IActionResult> SendCompressedBinaryAsync([FromBody]Context mod
                 try
                 {
                     var routeValueDictionary = new RouteValueDictionary(new { Key = key });
-                    var bytes = _distrubutedCache.Get(key);
-                    var compressor = HttpContext.RequestServices.GetService<ICompressor>();
-                    var compressedBytes = await compressor.CompressAsync(bytes);
-                    await _connectionManager.BroadcastBinaryAsync(compressedBytes, routeValueDictionary, false);
+
+                    // Get compressed bytes from redis
+                    var compressedBytes = _distrubutedCache.Get(key);
+                    await _connectionManager.BroadcastBinaryAsync(compressedBytes, routeValueDictionary);
                 }
                 catch (Exception ex)
                 {