fixed parallel writes #932

This commit is contained in:
JanEggers
2020-05-27 20:38:34 +02:00
parent 60941c04e5
commit 16180e83ba
2 changed files with 4 additions and 6 deletions
@@ -163,18 +163,16 @@ namespace MQTTnet.AspNetCore
public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, CancellationToken cancellationToken) public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, CancellationToken cancellationToken)
{ {
var formatter = PacketFormatterAdapter; var formatter = PacketFormatterAdapter;
var buffer = formatter.Encode(packet);
var msg = buffer.AsMemory();
var output = _output;
using (await _writerLock.WaitAsync(cancellationToken).ConfigureAwait(false)) using (await _writerLock.WaitAsync(cancellationToken).ConfigureAwait(false))
{ {
var buffer = formatter.Encode(packet);
var msg = buffer.AsMemory();
var output = _output;
var result = await output.WriteAsync(msg, cancellationToken).ConfigureAwait(false); var result = await output.WriteAsync(msg, cancellationToken).ConfigureAwait(false);
if (result.IsCompleted) if (result.IsCompleted)
{ {
BytesSent += msg.Length; BytesSent += msg.Length;
} }
PacketFormatterAdapter.FreeBuffer(); PacketFormatterAdapter.FreeBuffer();
} }
} }
@@ -45,7 +45,7 @@ namespace MQTTnet.AspNetCore.Tests
connection.Transport = pipe; connection.Transport = pipe;
var ctx = new MqttConnectionContext(serializer, connection); var ctx = new MqttConnectionContext(serializer, connection);
var tasks = Enumerable.Range(1, 10).Select(_ => Task.Run(async () => var tasks = Enumerable.Range(1, 100).Select(_ => Task.Run(async () =>
{ {
for (int i = 0; i < 100; i++) for (int i = 0; i < 100; i++)
{ {