From d09455f78e8c0d308b212eb6509b94d3de5ce05a Mon Sep 17 00:00:00 2001
From: Christian Kratky <christian.kratky@googlemail.com>
Date: Thu, 28 Sep 2017 09:29:47 +0200
Subject: [PATCH] Add semaphore for accessing send stream

---
 .../MqttChannelCommunicationAdapter.cs        | 42 ++++++++++---------
 .../PerformanceTest.cs                        | 20 ++++-----
 2 files changed, 32 insertions(+), 30 deletions(-)

diff --git a/MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs b/MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs
index c0c71eb6..8eab8cbc 100644
--- a/MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs
+++ b/MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs
@@ -16,8 +16,7 @@ namespace MQTTnet.Core.Adapter
     public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter
     {
         private readonly IMqttCommunicationChannel _channel;
-
-        private readonly Task _waitTask = Task.FromResult(0); // this task is used to prevent overlapping write
+        private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
 
         public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer)
         {
@@ -77,32 +76,31 @@ namespace MQTTnet.Core.Adapter
 
         public async Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, IEnumerable<MqttBasePacket> packets)
         {
+            await _semaphore.WaitAsync(cancellationToken);
+
             try
             {
-                var waitTask = _waitTask;
-
                 foreach (var packet in packets)
                 {
-                    if (packet == null) { continue; }
-
-                    MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), $"TX >>> {packet} [Timeout={timeout}]");
-
-                    var writeBuffer = PacketSerializer.Serialize(packet);
-
-                    waitTask = waitTask.ContinueWith(t => _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length, cancellationToken).ConfigureAwait(false), cancellationToken);
-                }
-
-                waitTask = waitTask.ContinueWith(t =>
-                {
-                    if (timeout > TimeSpan.Zero)
+                    if (packet == null)
                     {
-                        return _channel.SendStream.FlushAsync(cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
+                        continue;
                     }
 
-                    return _channel.SendStream.FlushAsync(cancellationToken).ConfigureAwait(false);
-                }, cancellationToken);
+                    MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), $"TX >>> {0} [Timeout={1}]", packet, timeout);
 
-                await waitTask; // configure await false generates stackoverflow
+                    var writeBuffer = PacketSerializer.Serialize(packet);
+                    await _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length, cancellationToken).ConfigureAwait(false);
+                }
+
+                if (timeout > TimeSpan.Zero)
+                {
+                    await _channel.SendStream.FlushAsync(cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
+                }
+                else
+                {
+                    await _channel.SendStream.FlushAsync(cancellationToken).ConfigureAwait(false);
+                }
             }
             catch (TaskCanceledException)
             {
@@ -120,6 +118,10 @@ namespace MQTTnet.Core.Adapter
             {
                 throw new MqttCommunicationException(exception);
             }
+            finally
+            {
+                _semaphore.Release();
+            }
         }
 
         public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)
diff --git a/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs b/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs
index 56d62a0a..41f24b2d 100644
--- a/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs
+++ b/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs
@@ -113,16 +113,16 @@ namespace MQTTnet.TestApp.NetFramework
                         .Select(i => CreateMessage())
                         .ToList();
 
-                    ////if (false)
-                    ////{
-                    ////    //send concurrent (test for raceconditions)
-                    ////    var sendTasks = msgs
-                    ////        .Select(msg => PublishSingleMessage(client, msg, ref msgCount))
-                    ////        .ToList();
+                    if (true)
+                    {
+                        //send concurrent (test for raceconditions)
+                        var sendTasks = msgs
+                            .Select(msg => PublishSingleMessage(client, msg, ref msgCount))
+                            .ToList();
 
-                    ////    await Task.WhenAll(sendTasks);
-                    ////}
-                    ////else
+                        await Task.WhenAll(sendTasks);
+                    }
+                    else
                     {
                         await client.PublishAsync(msgs);
                         msgCount += msgs.Count;
@@ -134,7 +134,7 @@ namespace MQTTnet.TestApp.NetFramework
                     var now = DateTime.Now;
                     if (last < now - TimeSpan.FromSeconds(1))
                     {
-                        Console.WriteLine($"sending {msgCount} inteded {msgChunkSize / interval.TotalSeconds}");
+                        Console.WriteLine($"sending {msgCount} intended {msgChunkSize / interval.TotalSeconds}");
                         msgCount = 0;
                         last = now;
                     }