YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-19 02:53:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed parallel writes

Browse Source
This commit is contained in:
JanEggers
2018-11-04 11:39:12 +01:00
parent 2bff218806
commit a5caab62f8
2 changed files with 37 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.AspnetCore/MqttConnectionContext.cs
+15 -3
View File
@@ -27,6 +27,8 @@ namespace MQTTnet.AspNetCore
public event EventHandler ReadingPacketStarted;
public event EventHandler ReadingPacketCompleted;
private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1);
public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
{
if (Connection is TcpConnection tcp && !tcp.IsConnected)
@@ -105,10 +107,20 @@ namespace MQTTnet.AspNetCore
return null;
}
public Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
public async Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
{
var buffer = PacketSerializer.Serialize(packet);
return Connection.Transport.Output.WriteAsync(buffer.AsMemory(), cancellationToken).AsTask();
var buffer = PacketSerializer.Serialize(packet).AsMemory();
var output = Connection.Transport.Output;
await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await output.WriteAsync(buffer, cancellationToken);
}
finally
{
_writerSemaphore.Release();
}
}
public void Dispose()