WingMan – Diff between revs 14 and 34

Subversion Repositories:

?pathlinks?

Rev 14 – Rev 35 – Go to most recent revision – Details – Blame – Last modification – View Log

 ï»¿using System;
 using System.IO;
+using System.IO.Compression;
 using System.Net;
 using System.Threading;
 using System.Threading.Tasks;
+using LZ4;
 using MQTTnet;
 using MQTTnet.Client;
 using MQTTnet.Extensions.ManagedClient;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
         private async void ClientOnApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
+        {
             try
             {
+                using (var inputStream = new MemoryStream(e.ApplicationMessage.Payload))
+                {
+                    using (var decryptedStream = await AES.Decrypt(inputStream, Password))
+                    {
+                        using (var lz4Decompress = new LZ4Stream(decryptedStream, CompressionMode.Decompress))
+                        {
+                            using (var outpuStream = new MemoryStream())
+                            {
+                                await lz4Decompress.CopyToAsync(outpuStream);
+                                outpuStream.Position = 0L;
+                                e.ApplicationMessage.Payload = outpuStream.ToArray();
+                            }
+                        }
+                    }
                 e.ApplicationMessage.Payload = await AES.Decrypt(e.ApplicationMessage.Payload, Password);
                 await Task.Delay(0, CancellationToken).ContinueWith(_ => OnMessageReceived?.Invoke(sender, e),
                     CancellationToken, TaskContinuationOptions.None, TaskScheduler);
         private async void ServerOnApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
+        {
             try
             {
+                using (var inputStream = new MemoryStream(e.ApplicationMessage.Payload))
+                {
+                    using (var decryptedStream = await AES.Decrypt(inputStream, Password))
+                    {
+                        using (var lz4Decompress = new LZ4Stream(decryptedStream, CompressionMode.Decompress))
+                        {
+                            using (var outpuStream = new MemoryStream())
+                            {
+                                await lz4Decompress.CopyToAsync(outpuStream);
+                                outpuStream.Position = 0L;
+                                e.ApplicationMessage.Payload = outpuStream.ToArray();
+                            }
+                        }
+                    }
                 e.ApplicationMessage.Payload = await AES.Decrypt(e.ApplicationMessage.Payload, Password);
                 await Task.Delay(0, CancellationToken).ContinueWith(_ => OnMessageReceived?.Invoke(sender, e),
                     CancellationToken, TaskContinuationOptions.None, TaskScheduler);
             Running = false;
+        }
         public async Task Broadcast(string topic, byte[] payload)
         {
-            using (var payloadStream = new MemoryStream(await AES.Encrypt(payload, Password)))
+            using (var compressStream = new MemoryStream())
             {
-                switch (Type)
+                using (var lz4Stream = new LZ4Stream(compressStream, CompressionMode.Compress))
                 {
-                    case MqttCommunicationType.Client:
+                    await lz4Stream.WriteAsync(payload, 0, payload.Length);
+                    await lz4Stream.FlushAsync();
-                        await Client.PublishAsync(new[]
+                    compressStream.Position = 0L;
-                        {
-                            new MqttApplicationMessage
+                    using (var outputStream = await AES.Encrypt(compressStream, Password))
-                            {
-                                Topic = topic,
+                    {
-                                Payload = payloadStream.ToArray(),
-                                QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
+                        var data = outputStream.ToArray();
-                            }
-                        });
-                        break;
-                    case MqttCommunicationType.Server:
-                        await Server.PublishAsync(new[]
+                        switch (Type)
+                        {
+                            case MqttCommunicationType.Client:
+                                await Client.PublishAsync(new[]
-                        {
+                                {
+                                    new MqttApplicationMessage
+                                    {
+                                        Topic = topic,
+                                        Payload = data,
+                                        QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
+                                    }
+                                });
+                                break;
+                            case MqttCommunicationType.Server:
-                            new MqttApplicationMessage
+                                await Server.PublishAsync(new[]
+                                {
+                                    new MqttApplicationMessage
-                            {
+                                    {
-                                Topic = topic,
+                                        Topic = topic,
-                                Payload = payloadStream.ToArray(),
+                                        Payload = data,
+                                        QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
-                                QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
+                                    }
+                                });
-                            }
+                                break;
-                        });
+                        }
                         break;
+                }
+            }
+        }
+    }