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

Merge branch 'develop' of https://github.com/chkr1011/MQTTnet into develop

Browse Source 
# Conflicts:
#	MQTTnet.Core/Serializer/MqttPacketReader.cs
#	MQTTnet.Core/Serializer/MqttPacketSerializer.cs
#	Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs
This commit is contained in:
Christian Kratky
2017-09-13 21:57:39 +02:00
parent 0e48076d25 9a25fb63ad
commit e2d15c8764
51 changed files with 804 additions and 753 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MQTTnet.Core/Adapter/IMqttCommunicationAdapter.cs
+2 -1
View File
@@ -1,4 +1,5 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using MQTTnet.Core.Client;
using MQTTnet.Core.Packets;
@@ -12,7 +13,7 @@ namespace MQTTnet.Core.Adapter
Task DisconnectAsync();
Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout);
Task SendPacketsAsync(TimeSpan timeout, IEnumerable<MqttBasePacket> packets);
Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout);
MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs
+43 -31
View File
@@ -1,9 +1,12 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Threading.Tasks;
using MQTTnet.Core.Channel;
using MQTTnet.Core.Client;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Internal;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Serializer;
@@ -12,6 +15,9 @@ namespace MQTTnet.Core.Adapter
public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter
{
private readonly IMqttCommunicationChannel _channel;
private readonly byte[] _readBuffer = new byte[BufferConstants.Size];
private Task _sendTask = Task.FromResult(0); // this task is used to prevent overlapping write
public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer)
{
@@ -23,7 +29,7 @@ namespace MQTTnet.Core.Adapter
public Task ConnectAsync(MqttClientOptions options, TimeSpan timeout)
{
return ExecuteWithTimeoutAsync(_channel.ConnectAsync(options), timeout);
return _channel.ConnectAsync(options).TimeoutAfter(timeout);
}
public Task DisconnectAsync()
@@ -31,25 +37,38 @@ namespace MQTTnet.Core.Adapter
return _channel.DisconnectAsync();
}
public Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout)
public async Task SendPacketsAsync(TimeSpan timeout, IEnumerable<MqttBasePacket> packets)
{
MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), "TX >>> {0} [Timeout={1}]", packet, timeout);
lock (_channel)
{
foreach (var packet in packets)
{
MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), "TX >>> {0} [Timeout={1}]", packet, timeout);
return ExecuteWithTimeoutAsync(PacketSerializer.SerializeAsync(packet, _channel), timeout);
var writeBuffer = PacketSerializer.Serialize(packet);
_sendTask = _sendTask.ContinueWith(p => _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length));
}
}
await _sendTask; // configure await false geneates stackoverflow
await _channel.SendStream.FlushAsync().TimeoutAfter(timeout).ConfigureAwait(false);
}
public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout)
{
MqttBasePacket packet;
Tuple<MqttPacketHeader, MemoryStream> tuple;
if (timeout > TimeSpan.Zero)
{
packet = await ExecuteWithTimeoutAsync(PacketSerializer.DeserializeAsync(_channel), timeout).ConfigureAwait(false);
tuple = await ReceiveAsync(_channel.RawStream).TimeoutAfter(timeout).ConfigureAwait(false);
}
else
{
packet = await PacketSerializer.DeserializeAsync(_channel).ConfigureAwait(false);
tuple = await ReceiveAsync(_channel.ReceiveStream).ConfigureAwait(false);
}
var packet = PacketSerializer.Deserialize(tuple.Item1, tuple.Item2);
if (packet == null)
{
throw new MqttProtocolViolationException("Received malformed packet.");
@@ -59,34 +78,27 @@ namespace MQTTnet.Core.Adapter
return packet;
}
private static async Task<TResult> ExecuteWithTimeoutAsync<TResult>(Task<TResult> task, TimeSpan timeout)
private async Task<Tuple<MqttPacketHeader, MemoryStream>> ReceiveAsync(Stream stream)
{
var timeoutTask = Task.Delay(timeout);
if (await Task.WhenAny(timeoutTask, task).ConfigureAwait(false) == timeoutTask)
var header = MqttPacketReader.ReadHeaderFromSource(stream);
MemoryStream body;
if (header.BodyLength > 0)
{
throw new MqttCommunicationTimedOutException();
var totalRead = 0;
do
{
var read = await stream.ReadAsync(_readBuffer, totalRead, header.BodyLength - totalRead).ConfigureAwait(false);
totalRead += read;
} while (totalRead < header.BodyLength);
body = new MemoryStream(_readBuffer, 0, header.BodyLength);
}
else
{
body = new MemoryStream();
}
if (task.IsFaulted)
{
throw new MqttCommunicationException(task.Exception);
}
return task.Result;
}
private static async Task ExecuteWithTimeoutAsync(Task task, TimeSpan timeout)
{
var timeoutTask = Task.Delay(timeout);
if (await Task.WhenAny(timeoutTask, task).ConfigureAwait(false) == timeoutTask)
{
throw new MqttCommunicationTimedOutException();
}
if (task.IsFaulted)
{
throw new MqttCommunicationException(task.Exception);
}
return Tuple.Create(header, body);
}
}
}
MQTTnet.Core/Adapter/MqttCommunicationAdapterExtensions.cs
+14
View File
@@ -0,0 +1,14 @@
using System;
using System.Threading.Tasks;
using MQTTnet.Core.Packets;
namespace MQTTnet.Core.Adapter
{
public static class MqttCommunicationAdapterExtensions
{
public static Task SendPacketsAsync(this IMqttCommunicationAdapter adapter, TimeSpan timeout, params MqttBasePacket[] packets)
{
return adapter.SendPacketsAsync(timeout, packets);
}
}
}
MQTTnet.Core/Adapter/MqttConnectingFailedException.cs
+1 -1
View File
@@ -5,7 +5,7 @@ namespace MQTTnet.Core.Adapter
{
public class MqttConnectingFailedException : MqttCommunicationException
{
public MqttConnectingFailedException(MqttConnectReturnCode returnCode)
public MqttConnectingFailedException(MqttConnectReturnCode returnCode)
: base($"Connecting with MQTT server failed ({returnCode}).")
{
ReturnCode = returnCode;
MQTTnet.Core/Channel/BufferedCommunicationChannel.cs
-77
View File
@@ -1,77 +0,0 @@
using System.Threading.Tasks;
using MQTTnet.Core.Client;
using System;
namespace MQTTnet.Core.Channel
{
public class BufferedCommunicationChannel : IMqttCommunicationChannel
{
private readonly IMqttCommunicationChannel _inner;
private int _bufferSize;
private int _bufferOffset;
public BufferedCommunicationChannel(IMqttCommunicationChannel inner)
{
_inner = inner;
}
public Task ConnectAsync(MqttClientOptions options)
{
return _inner.ConnectAsync(options);
}
public Task DisconnectAsync()
{
return _inner.DisconnectAsync();
}
public int Peek()
{
return _inner.Peek();
}
public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
{
//read from buffer
if (_bufferSize > 0)
{
return ReadFomBuffer(length, buffer);
}
var available = _inner.Peek();
// if there are less or equal bytes available then requested then just read em
if (available <= length)
{
return await _inner.ReadAsync(length, buffer);
}
//if more bytes are available than requested do buffer them to reduce calls to network buffers
await WriteToBuffer(available, buffer).ConfigureAwait(false);
return ReadFomBuffer(length, buffer);
}
private async Task WriteToBuffer(int available, byte[] buffer)
{
await _inner.ReadAsync(available, buffer).ConfigureAwait(false);
_bufferSize = available;
_bufferOffset = 0;
}
private ArraySegment<byte> ReadFomBuffer(int length, byte[] buffer)
{
var result = new ArraySegment<byte>(buffer, _bufferOffset, length);
_bufferSize -= length;
_bufferOffset += length;
if (_bufferSize < 0)
{
}
return result;
}
public Task WriteAsync(byte[] buffer)
{
return _inner.WriteAsync(buffer);
}
}
}
MQTTnet.Core/Channel/IMqttCommunicationChannel.cs
+5 -8
View File
@@ -1,6 +1,6 @@
using System.Threading.Tasks;
using MQTTnet.Core.Client;
using System;
using System.IO;
namespace MQTTnet.Core.Channel
{
@@ -9,14 +9,11 @@ namespace MQTTnet.Core.Channel
Task ConnectAsync(MqttClientOptions options);
Task DisconnectAsync();
Stream SendStream { get; }
Task WriteAsync(byte[] buffer);
Stream ReceiveStream { get; }
/// <summary>
/// get the currently available number of bytes without reading them
/// </summary>
int Peek();
Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer);
Stream RawStream { get; }
}
}
MQTTnet.Core/Client/IMqttClient.cs
+1 -1
View File
@@ -15,7 +15,7 @@ namespace MQTTnet.Core.Client
Task ConnectAsync(MqttApplicationMessage willApplicationMessage = null);
Task DisconnectAsync();
Task PublishAsync(MqttApplicationMessage applicationMessage);
Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages);
Task<IList<MqttSubscribeResult>> SubscribeAsync(IList<TopicFilter> topicFilters);
Task<IList<MqttSubscribeResult>> SubscribeAsync(params TopicFilter[] topicFilters);
Task Unsubscribe(IList<string> topicFilters);
MQTTnet.Core/Client/MqttClient.cs
+36 -23
View File
@@ -161,32 +161,43 @@ namespace MQTTnet.Core.Client
return SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket);
}
public Task PublishAsync(MqttApplicationMessage applicationMessage)
public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
ThrowIfNotConnected();
var publishPacket = applicationMessage.ToPublishPacket();
var publishPackets = applicationMessages.Select(m => m.ToPublishPacket());
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
foreach (var qosGroup in publishPackets.GroupBy(p => p.QualityOfServiceLevel))
{
// No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
return SendAsync(publishPacket);
var qosPackets = qosGroup.ToArray();
switch (qosGroup.Key)
{
case MqttQualityOfServiceLevel.AtMostOnce:
// No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, qosPackets);
break;
case MqttQualityOfServiceLevel.AtLeastOnce:
{
foreach (var publishPacket in qosPackets)
{
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket);
}
break;
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
foreach (var publishPacket in qosPackets)
{
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
await PublishExactlyOncePacketAsync(publishPacket);
}
break;
}
default:
throw new InvalidOperationException();
}
}
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
return SendAndReceiveAsync<MqttPubAckPacket>(publishPacket);
}
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
{
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
return PublishExactlyOncePacketAsync(publishPacket);
}
throw new InvalidOperationException();
}
private async Task PublishExactlyOncePacketAsync(MqttBasePacket publishPacket)
@@ -277,12 +288,14 @@ namespace MQTTnet.Core.Client
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
{
FireApplicationMessageReceivedEvent(publishPacket);
return;
}
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{
FireApplicationMessageReceivedEvent(publishPacket);
await SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
return;
}
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
@@ -295,6 +308,7 @@ namespace MQTTnet.Core.Client
FireApplicationMessageReceivedEvent(publishPacket);
await SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
return;
}
throw new MqttCommunicationException("Received a not supported QoS level.");
@@ -312,13 +326,12 @@ namespace MQTTnet.Core.Client
private Task SendAsync(MqttBasePacket packet)
{
return _adapter.SendPacketAsync(packet, _options.DefaultCommunicationTimeout);
return _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, packet);
}
private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket
{
await _adapter.SendPacketAsync(requestPacket, _options.DefaultCommunicationTimeout).ConfigureAwait(false);
await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, requestPacket).ConfigureAwait(false);
return (TResponsePacket)await _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.DefaultCommunicationTimeout).ConfigureAwait(false);
}
MQTTnet.Core/Client/MqttClientExtensions.cs
+12
View File
@@ -0,0 +1,12 @@
using System.Threading.Tasks;
namespace MQTTnet.Core.Client
{
public static class MqttClientExtensions
{
public static Task PublishAsync(this IMqttClient client, params MqttApplicationMessage[] applicationMessages)
{
return client.PublishAsync(applicationMessages);
}
}
}
MQTTnet.Core/Client/MqttPacketDispatcher.cs
+16 -12
View File
@@ -3,6 +3,7 @@ using System.Collections.Generic;
using System.Threading.Tasks;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Internal;
using MQTTnet.Core.Packets;
using System.Collections.Concurrent;
@@ -13,7 +14,7 @@ namespace MQTTnet.Core.Client
private readonly object _syncRoot = new object();
private readonly HashSet<MqttBasePacket> _receivedPackets = new HashSet<MqttBasePacket>();
private readonly ConcurrentDictionary<Type, TaskCompletionSource<MqttBasePacket>> _packetByResponseType = new ConcurrentDictionary<Type, TaskCompletionSource<MqttBasePacket>>();
private readonly ConcurrentDictionary<ushort,TaskCompletionSource<MqttBasePacket>> _packetByIdentifier = new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>();
private readonly ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>> _packetByIdentifier = new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>();
public async Task<MqttBasePacket> WaitForPacketAsync(MqttBasePacket request, Type responseType, TimeSpan timeout)
{
@@ -22,16 +23,19 @@ namespace MQTTnet.Core.Client
var packetAwaiter = AddPacketAwaiter(request, responseType);
DispatchPendingPackets();
var hasTimeout = await Task.WhenAny(Task.Delay(timeout), packetAwaiter.Task).ConfigureAwait(false) != packetAwaiter.Task;
RemovePacketAwaiter(request, responseType);
if (hasTimeout)
try
{
return await packetAwaiter.Task.TimeoutAfter(timeout);
}
catch (MqttCommunicationTimedOutException)
{
MqttTrace.Warning(nameof(MqttPacketDispatcher), "Timeout while waiting for packet.");
throw new MqttCommunicationTimedOutException();
throw;
}
finally
{
RemovePacketAwaiter(request, responseType);
}
return packetAwaiter.Task.Result;
}
public void Dispatch(MqttBasePacket packet)
@@ -48,9 +52,9 @@ namespace MQTTnet.Core.Client
packetDispatched = true;
}
}
else if (_packetByResponseType.TryRemove(packet.GetType(), out var tcs) )
else if (_packetByResponseType.TryRemove(packet.GetType(), out var tcs))
{
tcs.TrySetResult( packet);
tcs.TrySetResult(packet);
packetDispatched = true;
}
@@ -96,11 +100,11 @@ namespace MQTTnet.Core.Client
{
if (request is IMqttPacketWithIdentifier withIdent)
{
_packetByIdentifier.TryRemove(withIdent.PacketIdentifier, out var tcs);
_packetByIdentifier.TryRemove(withIdent.PacketIdentifier, out var _);
}
else
{
_packetByResponseType.TryRemove(responseType, out var tcs);
_packetByResponseType.TryRemove(responseType, out var _);
}
}
MQTTnet.Core/Diagnostics/MqttTrace.cs
-1
View File
@@ -1,5 +1,4 @@
using System;
using System.Linq;
namespace MQTTnet.Core.Diagnostics
{
MQTTnet.Core/Diagnostics/MqttTraceLevel.cs
+1 -1
View File
@@ -1,6 +1,6 @@
namespace MQTTnet.Core.Diagnostics
{
public enum MqttTraceLevel
public enum MqttTraceLevel
{
Verbose,
Information,
MQTTnet.Core/Exceptions/MqttProtocolViolationException.cs
+1 -1
View File
@@ -4,7 +4,7 @@ namespace MQTTnet.Core.Exceptions
{
public sealed class MqttProtocolViolationException : Exception
{
public MqttProtocolViolationException(string message)
public MqttProtocolViolationException(string message)
: base(message)
{
}
MQTTnet.Core/Internal/AsyncAutoResetEvent.cs
-32
View File
@@ -1,32 +0,0 @@
using System.Collections.Generic;
using System.Threading.Tasks;
namespace MQTTnet.Core.Internal
{
public sealed class AsyncGate
{
private readonly Queue<TaskCompletionSource<bool>> _waitingTasks = new Queue<TaskCompletionSource<bool>>();
public Task WaitOneAsync()
{
var tcs = new TaskCompletionSource<bool>();
lock (_waitingTasks)
{
_waitingTasks.Enqueue(tcs);
}
return tcs.Task;
}
public void Set()
{
lock (_waitingTasks)
{
if (_waitingTasks.Count > 0)
{
_waitingTasks.Dequeue().SetResult(true);
}
}
}
}
}
MQTTnet.Core/Internal/TaskExtensions.cs
+55
View File
@@ -0,0 +1,55 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Exceptions;
namespace MQTTnet.Core.Internal
{
public static class TaskExtensions
{
public static Task TimeoutAfter(this Task task, TimeSpan timeout)
{
return TimeoutAfter(task.ContinueWith(t => 0), timeout);
}
public static async Task<TResult> TimeoutAfter<TResult>(this Task<TResult> task, TimeSpan timeout)
{
using (var cancellationTokenSource = new CancellationTokenSource())
{
var tcs = new TaskCompletionSource<TResult>();
cancellationTokenSource.Token.Register(() =>
{
tcs.TrySetCanceled();
});
try
{
cancellationTokenSource.CancelAfter(timeout);
task.ContinueWith(t =>
{
if (t.IsFaulted)
{
tcs.TrySetException(t.Exception);
}
if (t.IsCompleted)
{
tcs.TrySetResult(t.Result);
}
}, cancellationTokenSource.Token);
return await tcs.Task;
}
catch (TaskCanceledException)
{
throw new MqttCommunicationTimedOutException();
}
catch (Exception e)
{
throw new MqttCommunicationException(e);
}
}
}
}
}
MQTTnet.Core/MQTTnet.Core.csproj
+1
View File
@@ -5,6 +5,7 @@
<AssemblyName>MQTTnet.Core</AssemblyName>
<RootNamespace>MQTTnet.Core</RootNamespace>
<GeneratePackageOnBuild>False</GeneratePackageOnBuild>
<DebugType>Full</DebugType>
<Description></Description>
<Product></Product>
<Company></Company>
MQTTnet.Core/Packets/MqttPublishPacket.cs
+1 -1
View File
@@ -17,7 +17,7 @@ namespace MQTTnet.Core.Packets
public override string ToString()
{
return nameof(MqttPublishPacket) +
return nameof(MqttPublishPacket) +
": [Topic=" + Topic + "]" +
" [Payload=" + Convert.ToBase64String(Payload) + "]" +
" [QoSLevel=" + QualityOfServiceLevel + "]" +
MQTTnet.Core/Packets/MqttSubscribePacket.cs
+1 -1
View File
@@ -6,7 +6,7 @@ namespace MQTTnet.Core.Packets
public sealed class MqttSubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier
{
public ushort PacketIdentifier { get; set; }
public IList<TopicFilter> TopicFilters { get; set; } = new List<TopicFilter>();
public override string ToString()
MQTTnet.Core/Packets/MqttUnsubscribe.cs
+1 -1
View File
@@ -5,7 +5,7 @@ namespace MQTTnet.Core.Packets
public sealed class MqttUnsubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier
{
public ushort PacketIdentifier { get; set; }
public IList<string> TopicFilters { get; set; } = new List<string>();
}
}
MQTTnet.Core/Serializer/IMqttPacketSerializer.cs
+3 -4
View File
@@ -1,5 +1,4 @@
using System.Threading.Tasks;
using MQTTnet.Core.Channel;
using System.IO;
using MQTTnet.Core.Packets;
namespace MQTTnet.Core.Serializer
@@ -8,8 +7,8 @@ namespace MQTTnet.Core.Serializer
{
MqttProtocolVersion ProtocolVersion { get; set; }
Task SerializeAsync(MqttBasePacket mqttPacket, IMqttCommunicationChannel destination);
byte[] Serialize(MqttBasePacket mqttPacket);
Task<MqttBasePacket> DeserializeAsync(IMqttCommunicationChannel source);
MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream stream);
}
}
MQTTnet.Core/Serializer/MqttPacketReader.cs
+10 -33
View File
@@ -1,10 +1,8 @@
using System;
using System.IO;
using System.Text;
using System.Threading.Tasks;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Protocol;
using MQTTnet.Core.Channel;
using MQTTnet.Core.Packets;
namespace MQTTnet.Core.Serializer
@@ -13,14 +11,14 @@ namespace MQTTnet.Core.Serializer
{
private readonly MqttPacketHeader _header;
public MqttPacketReader(MqttPacketHeader header, Stream body)
: base(body)
public MqttPacketReader(Stream stream, MqttPacketHeader header)
: base(stream, Encoding.UTF8, true)
{
_header = header;
}
public bool EndOfRemainingData => BaseStream.Position == _header.BodyLength;
public override ushort ReadUInt16()
{
var buffer = ReadBytes(2);
@@ -49,14 +47,11 @@ namespace MQTTnet.Core.Serializer
return ReadBytes(_header.BodyLength - (int)BaseStream.Position);
}
public static async Task<MqttPacketHeader> ReadHeaderFromSourceAsync(IMqttCommunicationChannel source, byte[] buffer)
public static MqttPacketHeader ReadHeaderFromSource(Stream stream)
{
var fixedHeader = await ReadStreamByteAsync(source, buffer).ConfigureAwait(false);
var byteReader = new ByteReader(fixedHeader);
byteReader.Read(4);
var controlPacketType = (MqttControlPacketType)byteReader.Read(4);
var bodyLength = await ReadBodyLengthFromSourceAsync(source, buffer).ConfigureAwait(false);
var fixedHeader = (byte)stream.ReadByte();
var controlPacketType = (MqttControlPacketType)(fixedHeader >> 4);
var bodyLength = ReadBodyLengthFromSource(stream);
return new MqttPacketHeader
{
@@ -66,25 +61,7 @@ namespace MQTTnet.Core.Serializer
};
}
private static async Task<byte> ReadStreamByteAsync(IMqttCommunicationChannel source, byte[] readBuffer)
{
var result = await ReadFromSourceAsync(source, 1, readBuffer).ConfigureAwait(false);
return result.Array[result.Offset];
}
public static async Task<ArraySegment<byte>> ReadFromSourceAsync(IMqttCommunicationChannel source, int length, byte[] buffer)
{
try
{
return await source.ReadAsync(length, buffer);
}
catch (Exception exception)
{
throw new MqttCommunicationException(exception);
}
}
private static async Task<int> ReadBodyLengthFromSourceAsync(IMqttCommunicationChannel source, byte[] buffer)
private static int ReadBodyLengthFromSource(Stream stream)
{
// Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html.
var multiplier = 1;
@@ -92,7 +69,7 @@ namespace MQTTnet.Core.Serializer
byte encodedByte;
do
{
encodedByte = await ReadStreamByteAsync(source, buffer).ConfigureAwait(false);
encodedByte = (byte)stream.ReadByte();
value += (encodedByte & 127) * multiplier;
multiplier *= 128;
if (multiplier > 128 * 128 * 128)
MQTTnet.Core/Serializer/MqttPacketSerializer.cs
+118 -117
View File
@@ -3,8 +3,6 @@ using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using MQTTnet.Core.Channel;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
@@ -17,12 +15,10 @@ namespace MQTTnet.Core.Serializer
private static byte[] ProtocolVersionV310Name { get; } = Encoding.UTF8.GetBytes("MQIs");
public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
private byte[] _readBuffer = new byte[BufferConstants.Size]; // TODO: What happens if the message is bigger?
public async Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination)
public byte[] Serialize(MqttBasePacket packet)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
if (destination == null) throw new ArgumentNullException(nameof(destination));
using (var stream = new MemoryStream())
using (var writer = new MqttPacketWriter(stream))
@@ -30,9 +26,12 @@ namespace MQTTnet.Core.Serializer
var header = new List<byte> { SerializePacket(packet, writer) };
var body = stream.ToArray();
MqttPacketWriter.BuildLengthHeader(body.Length, header);
await destination.WriteAsync(header.ToArray()).ConfigureAwait(false);
await destination.WriteAsync(body).ConfigureAwait(false);
var headerArray = header.ToArray();
var writeBuffer = new byte[header.Count + body.Length];
Buffer.BlockCopy(headerArray, 0, writeBuffer, 0, headerArray.Length);
Buffer.BlockCopy(body, 0, writeBuffer, headerArray.Length, body.Length);
return writeBuffer;
}
}
@@ -111,119 +110,112 @@ namespace MQTTnet.Core.Serializer
throw new MqttProtocolViolationException("Packet type invalid.");
}
public async Task<MqttBasePacket> DeserializeAsync(IMqttCommunicationChannel source)
public MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream body)
{
if (source == null) throw new ArgumentNullException(nameof(source));
if (header == null) throw new ArgumentNullException(nameof(header));
if (body == null) throw new ArgumentNullException(nameof(body));
var header = await MqttPacketReader.ReadHeaderFromSourceAsync(source, _readBuffer).ConfigureAwait(false);
var body = await GetBody(source, header).ConfigureAwait(false);
using (var mqttPacketReader = new MqttPacketReader(header, body))
using (var reader = new MqttPacketReader(body, header))
{
switch (header.ControlPacketType)
{
case MqttControlPacketType.Connect:
{
return DeserializeConnect(mqttPacketReader);
}
case MqttControlPacketType.ConnAck:
{
return DeserializeConnAck(mqttPacketReader);
}
case MqttControlPacketType.Disconnect:
{
return new MqttDisconnectPacket();
}
case MqttControlPacketType.Publish:
{
return DeserializePublish(mqttPacketReader, header);
}
case MqttControlPacketType.PubAck:
{
return new MqttPubAckPacket
{
PacketIdentifier = mqttPacketReader.ReadUInt16()
};
}
case MqttControlPacketType.PubRec:
{
return new MqttPubRecPacket
{
PacketIdentifier = mqttPacketReader.ReadUInt16()
};
}
case MqttControlPacketType.PubRel:
{
return new MqttPubRelPacket
{
PacketIdentifier = mqttPacketReader.ReadUInt16()
};
}
case MqttControlPacketType.PubComp:
{
return new MqttPubCompPacket
{
PacketIdentifier = mqttPacketReader.ReadUInt16()
};
}
case MqttControlPacketType.PingReq:
{
return new MqttPingReqPacket();
}
case MqttControlPacketType.PingResp:
{
return new MqttPingRespPacket();
}
case MqttControlPacketType.Subscribe:
{
return DeserializeSubscribe(mqttPacketReader);
}
case MqttControlPacketType.SubAck:
{
return DeserializeSubAck(mqttPacketReader);
}
case MqttControlPacketType.Unsubscibe:
{
return DeserializeUnsubscribe(mqttPacketReader);
}
case MqttControlPacketType.UnsubAck:
{
return new MqttUnsubAckPacket
{
PacketIdentifier = mqttPacketReader.ReadUInt16()
};
}
default:
{
throw new MqttProtocolViolationException($"Packet type ({(int)header.ControlPacketType}) not supported.");
}
}
return Deserialize(header, reader);
}
}
private async Task<MemoryStream> GetBody(IMqttCommunicationChannel source, MqttPacketHeader header)
private static MqttBasePacket Deserialize(MqttPacketHeader header, MqttPacketReader reader)
{
if (header.BodyLength > 0)
switch (header.ControlPacketType)
{
var segment = await MqttPacketReader.ReadFromSourceAsync(source, header.BodyLength, _readBuffer).ConfigureAwait(false);
return new MemoryStream(segment.Array, segment.Offset, segment.Count);
case MqttControlPacketType.Connect:
{
return DeserializeConnect(reader);
}
case MqttControlPacketType.ConnAck:
{
return DeserializeConnAck(reader);
}
case MqttControlPacketType.Disconnect:
{
return new MqttDisconnectPacket();
}
case MqttControlPacketType.Publish:
{
return DeserializePublish(reader, header);
}
case MqttControlPacketType.PubAck:
{
return new MqttPubAckPacket
{
PacketIdentifier = reader.ReadUInt16()
};
}
case MqttControlPacketType.PubRec:
{
return new MqttPubRecPacket
{
PacketIdentifier = reader.ReadUInt16()
};
}
case MqttControlPacketType.PubRel:
{
return new MqttPubRelPacket
{
PacketIdentifier = reader.ReadUInt16()
};
}
case MqttControlPacketType.PubComp:
{
return new MqttPubCompPacket
{
PacketIdentifier = reader.ReadUInt16()
};
}
case MqttControlPacketType.PingReq:
{
return new MqttPingReqPacket();
}
case MqttControlPacketType.PingResp:
{
return new MqttPingRespPacket();
}
case MqttControlPacketType.Subscribe:
{
return DeserializeSubscribe(reader);
}
case MqttControlPacketType.SubAck:
{
return DeserializeSubAck(reader);
}
case MqttControlPacketType.Unsubscibe:
{
return DeserializeUnsubscribe(reader);
}
case MqttControlPacketType.UnsubAck:
{
return new MqttUnsubAckPacket
{
PacketIdentifier = reader.ReadUInt16()
};
}
default:
{
throw new MqttProtocolViolationException(
$"Packet type ({(int)header.ControlPacketType}) not supported.");
}
}
return new MemoryStream();
}
private static MqttBasePacket DeserializeUnsubscribe(MqttPacketReader reader)
@@ -514,12 +506,21 @@ namespace MQTTnet.Core.Serializer
writer.Write(packet.Payload);
}
var fixedHeader = new ByteWriter();
fixedHeader.Write(packet.Retain);
fixedHeader.Write((byte)packet.QualityOfServiceLevel, 2);
fixedHeader.Write(packet.Dup);
byte fixedHeader = 0;
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader.Value);
if (packet.Retain)
{
fixedHeader |= 0x01;
}
fixedHeader |= (byte)((byte)packet.QualityOfServiceLevel << 1);
if (packet.Dup)
{
fixedHeader |= 0x08;
}
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader);
}
private static byte Serialize(MqttPubAckPacket packet, MqttPacketWriter writer)
MQTTnet.Core/Serializer/MqttPacketWriter.cs
+3 -4
View File
@@ -8,10 +8,9 @@ namespace MQTTnet.Core.Serializer
{
public sealed class MqttPacketWriter : BinaryWriter
{
public MqttPacketWriter( Stream stream )
public MqttPacketWriter(Stream stream)
: base(stream)
{
}
public static byte BuildFixedHeader(MqttControlPacketType packetType, byte flags = 0)
@@ -20,7 +19,7 @@ namespace MQTTnet.Core.Serializer
fixedHeader |= flags;
return (byte)fixedHeader;
}
public override void Write(ushort value)
{
var buffer = BitConverter.GetBytes(value);
@@ -43,7 +42,7 @@ namespace MQTTnet.Core.Serializer
Write(value.Value);
}
public void WriteWithLengthPrefix(string value)
{
WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value ?? string.Empty));
MQTTnet.Core/Server/MqttClientMessageQueue.cs
+9 -35
View File
@@ -1,20 +1,17 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Internal;
using MQTTnet.Core.Packets;
namespace MQTTnet.Core.Server
{
public sealed class MqttClientMessageQueue
{
private readonly List<MqttClientPublishPacketContext> _pendingPublishPackets = new List<MqttClientPublishPacketContext>();
private readonly AsyncGate _gate = new AsyncGate();
private readonly BlockingCollection<MqttClientPublishPacketContext> _pendingPublishPackets = new BlockingCollection<MqttClientPublishPacketContext>();
private readonly MqttServerOptions _options;
private CancellationTokenSource _cancellationTokenSource;
@@ -43,26 +40,22 @@ namespace MQTTnet.Core.Server
_adapter = null;
_cancellationTokenSource?.Cancel();
_cancellationTokenSource = null;
_pendingPublishPackets?.Dispose();
}
public void Enqueue(MqttPublishPacket publishPacket)
{
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
lock (_pendingPublishPackets)
{
_pendingPublishPackets.Add(new MqttClientPublishPacketContext(publishPacket));
_gate.Set();
}
_pendingPublishPackets.Add(new MqttClientPublishPacketContext(publishPacket));
}
private async Task SendPendingPublishPacketsAsync(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
foreach (var publishPacket in _pendingPublishPackets.GetConsumingEnumerable(cancellationToken))
{
try
{
await _gate.WaitOneAsync().ConfigureAwait(false);
if (cancellationToken.IsCancellationRequested)
{
return;
@@ -73,25 +66,12 @@ namespace MQTTnet.Core.Server
continue;
}
List<MqttClientPublishPacketContext> pendingPublishPackets;
lock (_pendingPublishPackets)
{
pendingPublishPackets = _pendingPublishPackets.ToList();
}
foreach (var publishPacket in pendingPublishPackets)
{
await TrySendPendingPublishPacketAsync(publishPacket).ConfigureAwait(false);
}
await TrySendPendingPublishPacketAsync(publishPacket).ConfigureAwait(false);
}
catch (Exception e)
{
MqttTrace.Error(nameof(MqttClientMessageQueue), e, "Error while sending pending publish packets.");
}
finally
{
Cleanup();
}
}
}
@@ -105,30 +85,24 @@ namespace MQTTnet.Core.Server
}
publishPacketContext.PublishPacket.Dup = publishPacketContext.SendTries > 0;
await _adapter.SendPacketAsync(publishPacketContext.PublishPacket, _options.DefaultCommunicationTimeout).ConfigureAwait(false);
await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, publishPacketContext.PublishPacket).ConfigureAwait(false);
publishPacketContext.IsSent = true;
}
catch (MqttCommunicationException exception)
{
MqttTrace.Warning(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
_pendingPublishPackets.Add(publishPacketContext);
}
catch (Exception exception)
{
MqttTrace.Error(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
_pendingPublishPackets.Add(publishPacketContext);
}
finally
{
publishPacketContext.SendTries++;
}
}
private void Cleanup()
{
lock (_pendingPublishPackets)
{
_pendingPublishPackets.RemoveAll(p => p.IsSent);
}
}
}
}
MQTTnet.Core/Server/MqttClientSession.cs
+7 -7
View File
@@ -103,12 +103,12 @@ namespace MQTTnet.Core.Server
{
if (packet is MqttSubscribePacket subscribePacket)
{
return Adapter.SendPacketAsync(_subscriptionsManager.Subscribe(subscribePacket), _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _subscriptionsManager.Subscribe(subscribePacket));
}
if (packet is MqttUnsubscribePacket unsubscribePacket)
{
return Adapter.SendPacketAsync(_subscriptionsManager.Unsubscribe(unsubscribePacket), _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _subscriptionsManager.Unsubscribe(unsubscribePacket));
}
if (packet is MqttPublishPacket publishPacket)
@@ -123,7 +123,7 @@ namespace MQTTnet.Core.Server
if (packet is MqttPubRecPacket pubRecPacket)
{
return Adapter.SendPacketAsync(pubRecPacket.CreateResponse<MqttPubRelPacket>(), _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, pubRecPacket.CreateResponse<MqttPubRelPacket>());
}
if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
@@ -134,7 +134,7 @@ namespace MQTTnet.Core.Server
if (packet is MqttPingReqPacket)
{
return Adapter.SendPacketAsync(new MqttPingRespPacket(), _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttPingRespPacket());
}
if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
@@ -160,7 +160,7 @@ namespace MQTTnet.Core.Server
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{
_publishPacketReceivedCallback(this, publishPacket);
return Adapter.SendPacketAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
}
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
@@ -173,7 +173,7 @@ namespace MQTTnet.Core.Server
_publishPacketReceivedCallback(this, publishPacket);
return Adapter.SendPacketAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
}
throw new MqttCommunicationException("Received a not supported QoS level.");
@@ -186,7 +186,7 @@ namespace MQTTnet.Core.Server
_unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier);
}
return Adapter.SendPacketAsync(new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier });
}
}
}
MQTTnet.Core/Server/MqttClientSessionsManager.cs
+4 -4
View File
@@ -40,21 +40,21 @@ namespace MQTTnet.Core.Server
var connectReturnCode = ValidateConnection(connectPacket);
if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
{
await eventArgs.ClientAdapter.SendPacketAsync(new MqttConnAckPacket
await eventArgs.ClientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttConnAckPacket
{
ConnectReturnCode = connectReturnCode
}, _options.DefaultCommunicationTimeout).ConfigureAwait(false);
}).ConfigureAwait(false);
return;
}
var clientSession = GetOrCreateClientSession(connectPacket);
await eventArgs.ClientAdapter.SendPacketAsync(new MqttConnAckPacket
await eventArgs.ClientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttConnAckPacket
{
ConnectReturnCode = connectReturnCode,
IsSessionPresent = clientSession.IsExistingSession
}, _options.DefaultCommunicationTimeout).ConfigureAwait(false);
}).ConfigureAwait(false);
await clientSession.Session.RunAsync(eventArgs.Identifier, connectPacket.WillMessage, eventArgs.ClientAdapter).ConfigureAwait(false);
}
MQTTnet.Core/Server/MqttServer.cs
+2 -2
View File
@@ -20,7 +20,7 @@ namespace MQTTnet.Core.Server
{
_options = options ?? throw new ArgumentNullException(nameof(options));
_adapters = adapters ?? throw new ArgumentNullException(nameof(adapters));
_clientSessionsManager = new MqttClientSessionsManager(options);
_clientSessionsManager.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
}
@@ -61,7 +61,7 @@ namespace MQTTnet.Core.Server
adapter.ClientConnected += OnClientConnected;
adapter.Start(_options);
}
MqttTrace.Information(nameof(MqttServer), "Started.");
}
MQTTnet.Core/Server/MqttServerOptions.cs
+1 -1
View File
@@ -9,7 +9,7 @@ namespace MQTTnet.Core.Server
public MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; } = new MqttServerDefaultEndpointOptions();
public MqttServerTlsEndpointOptions TlsEndpointOptions { get; } = new MqttServerTlsEndpointOptions();
public int ConnectionBacklog { get; set; } = 10;
public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(15);