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

optimized communication adapter

Browse Source
This commit is contained in:
Eggers Jan
2017-09-13 09:14:08 +02:00
parent 473c8e0a15
commit 97338b016d
1 changed files with 13 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs
+13 -14
View File
@@ -16,6 +16,7 @@ namespace MQTTnet.Core.Adapter
public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter
{ {
private readonly IMqttCommunicationChannel _channel; private readonly IMqttCommunicationChannel _channel;
private readonly byte[] _readBuffer = new byte[BufferConstants.Size];
public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer) public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer)
{ {
@@ -37,12 +38,16 @@ namespace MQTTnet.Core.Adapter
public async Task SendPacketsAsync( TimeSpan timeout, IEnumerable<MqttBasePacket> packets ) public async Task SendPacketsAsync( TimeSpan timeout, IEnumerable<MqttBasePacket> packets )
{ {
foreach (var packet in packets ) lock ( _channel )
{ {
MqttTrace.Information( nameof( MqttChannelCommunicationAdapter ), "TX >>> {0} [Timeout={1}]", packet, timeout ); foreach (var packet in packets )
{
MqttTrace.Information( nameof( MqttChannelCommunicationAdapter ), "TX >>> {0} [Timeout={1}]", packet, timeout );
var writeBuffer = PacketSerializer.Serialize(packet); var writeBuffer = PacketSerializer.Serialize(packet);
_sendTask = SendAsync( writeBuffer );
_sendTask = _sendTask.ContinueWith( p => _channel.SendStream.WriteAsync( writeBuffer, 0, writeBuffer.Length ) );
}
} }
await _sendTask.ConfigureAwait( false ); await _sendTask.ConfigureAwait( false );
@@ -51,12 +56,6 @@ namespace MQTTnet.Core.Adapter
private Task _sendTask = Task.FromResult(0); // this task is used to prevent overlapping write private Task _sendTask = Task.FromResult(0); // this task is used to prevent overlapping write
private async Task SendAsync(byte[] buffer)
{
await _sendTask.ConfigureAwait(false);
await _channel.SendStream.WriteAsync(buffer, 0, buffer.Length).ConfigureAwait( false );
}
public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout) public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout)
{ {
Tuple<MqttPacketHeader, MemoryStream> tuple; Tuple<MqttPacketHeader, MemoryStream> tuple;
@@ -82,20 +81,20 @@ namespace MQTTnet.Core.Adapter
private async Task<Tuple<MqttPacketHeader, MemoryStream>> ReceiveAsync() private async Task<Tuple<MqttPacketHeader, MemoryStream>> ReceiveAsync()
{ {
var header = MqttPacketReader.ReadHeaderFromSource(_channel.ReceiveStream); var stream = _channel.ReceiveStream;
var header = MqttPacketReader.ReadHeaderFromSource(stream);
MemoryStream body = null; MemoryStream body = null;
if (header.BodyLength > 0) if (header.BodyLength > 0)
{ {
var totalRead = 0; var totalRead = 0;
var readBuffer = new byte[header.BodyLength];
do do
{ {
var read = await _channel.ReceiveStream.ReadAsync(readBuffer, totalRead, header.BodyLength - totalRead) var read = await stream.ReadAsync(_readBuffer, totalRead, header.BodyLength - totalRead)
.ConfigureAwait( false ); .ConfigureAwait( false );
totalRead += read; totalRead += read;
} while (totalRead < header.BodyLength); } while (totalRead < header.BodyLength);
body = new MemoryStream(readBuffer, 0, header.BodyLength); body = new MemoryStream(_readBuffer, 0, header.BodyLength);
} }
else else
{ {