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

Covnert the body buffer to a memory stream which will grow automatically.

Browse Source
This commit is contained in:
Christian
2018-01-24 20:49:47 +01:00
parent e56203faac
commit 1415e4b878
7 changed files with 57 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs
+38 -29
View File
@@ -17,8 +17,7 @@ namespace MQTTnet.Adapter
public class MqttChannelAdapter : IMqttChannelAdapter
{
private const uint ErrorOperationAborted = 0x800703E3;
private static readonly byte[] EmptyBody = new byte[0];
private const int ReadBufferSize = 4096; // TODO: Move buffer size to config
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly IMqttNetLogger _logger;
@@ -91,28 +90,35 @@ namespace MQTTnet.Adapter
MqttBasePacket packet = null;
await ExecuteAndWrapExceptionAsync(async () =>
{
ReceivedMqttPacket receivedMqttPacket;
if (timeout > TimeSpan.Zero)
ReceivedMqttPacket receivedMqttPacket = null;
try
{
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
}
else
{
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).ConfigureAwait(false);
}
if (timeout > TimeSpan.Zero)
{
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
}
else
{
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).ConfigureAwait(false);
}
if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
{
throw new TaskCanceledException();
}
if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
{
throw new TaskCanceledException();
}
packet = PacketSerializer.Deserialize(receivedMqttPacket.Header, receivedMqttPacket.Body);
if (packet == null)
{
throw new MqttProtocolViolationException("Received malformed packet.");
}
packet = PacketSerializer.Deserialize(receivedMqttPacket.Header, receivedMqttPacket.Body);
if (packet == null)
{
throw new MqttProtocolViolationException("Received malformed packet.");
}
_logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
_logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
}
finally
{
receivedMqttPacket?.Dispose();
}
}).ConfigureAwait(false);
return packet;
@@ -120,7 +126,7 @@ namespace MQTTnet.Adapter
private static async Task<ReceivedMqttPacket> ReceiveAsync(Stream stream, CancellationToken cancellationToken)
{
var header = await MqttPacketReader.ReadHeaderFromSourceAsync(stream, cancellationToken).ConfigureAwait(false);
var header = await MqttPacketReader.ReadHeaderAsync(stream, cancellationToken).ConfigureAwait(false);
if (header == null)
{
return null;
@@ -128,23 +134,26 @@ namespace MQTTnet.Adapter
if (header.BodyLength == 0)
{
return new ReceivedMqttPacket(header, EmptyBody);
return new ReceivedMqttPacket(header, new MemoryStream(new byte[0], false));
}
var body = new byte[header.BodyLength];
var offset = 0;
do
var body = header.BodyLength <= ReadBufferSize ? new MemoryStream(new byte[header.BodyLength]) : new MemoryStream();
var buffer = new byte[ReadBufferSize];
while (body.Length < header.BodyLength)
{
var readBytesCount = await stream.ReadAsync(body, offset, body.Length - offset, cancellationToken).ConfigureAwait(false);
var readBytesCount = await stream.ReadAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false);
// Check if the client closed the connection before sending the full body.
if (readBytesCount == 0)
{
throw new MqttCommunicationException("Connection closed while reading remaining packet body.");
}
offset += readBytesCount;
} while (offset < header.BodyLength);
body.Write(buffer, 0, readBytesCount);
}
body.Seek(0L, SeekOrigin.Begin);
return new ReceivedMqttPacket(header, body);
}