@@ -138,7 +138,7 @@ namespace MQTTnet.Adapter | |||
} | |||
var body = header.BodyLength <= ReadBufferSize ? new MemoryStream(header.BodyLength) : new MemoryStream(); | |||
var buffer = new byte[ReadBufferSize]; | |||
while (body.Length < header.BodyLength) | |||
{ | |||
@@ -149,7 +149,7 @@ namespace MQTTnet.Adapter | |||
} | |||
var readBytesCount = await stream.ReadAsync(buffer, 0, bytesLeft, cancellationToken).ConfigureAwait(false); | |||
// Check if the client closed the connection before sending the full body. | |||
if (readBytesCount == 0) | |||
{ | |||
@@ -162,7 +162,7 @@ namespace MQTTnet.Adapter | |||
} | |||
body.Seek(0L, SeekOrigin.Begin); | |||
return new ReceivedMqttPacket(header, body); | |||
} | |||
@@ -190,7 +190,7 @@ namespace MQTTnet.Adapter | |||
} | |||
catch (COMException comException) | |||
{ | |||
if ((uint) comException.HResult == ErrorOperationAborted) | |||
if ((uint)comException.HResult == ErrorOperationAborted) | |||
{ | |||
throw new OperationCanceledException(); | |||
} | |||
@@ -1,17 +0,0 @@ | |||
using System; | |||
using System.Threading; | |||
using System.Threading.Tasks; | |||
using MQTTnet.Packets; | |||
namespace MQTTnet.Adapter | |||
{ | |||
public static class MqttChannelAdapterExtensions | |||
{ | |||
public static Task SendPacketsAsync(this IMqttChannelAdapter adapter, TimeSpan timeout, CancellationToken cancellationToken, params MqttBasePacket[] packets) | |||
{ | |||
if (adapter == null) throw new ArgumentNullException(nameof(adapter)); | |||
return adapter.SendPacketsAsync(timeout, cancellationToken, packets); | |||
} | |||
} | |||
} |
@@ -2,6 +2,6 @@ | |||
{ | |||
public interface IMqttPacketWithIdentifier | |||
{ | |||
ushort PacketIdentifier { get; set; } | |||
ushort? PacketIdentifier { get; set; } | |||
} | |||
} |
@@ -2,6 +2,6 @@ | |||
{ | |||
public class MqttBasePublishPacket : MqttBasePacket, IMqttPacketWithIdentifier | |||
{ | |||
public ushort PacketIdentifier { get; set; } | |||
public ushort? PacketIdentifier { get; set; } | |||
} | |||
} |
@@ -1,26 +0,0 @@ | |||
using System; | |||
namespace MQTTnet.Packets | |||
{ | |||
public static class MqttPacketExtensions | |||
{ | |||
public static TResponsePacket CreateResponse<TResponsePacket>(this MqttBasePacket packet) | |||
{ | |||
if (packet == null) throw new ArgumentNullException(nameof(packet)); | |||
var responsePacket = Activator.CreateInstance<TResponsePacket>(); | |||
if (responsePacket is IMqttPacketWithIdentifier responsePacketWithIdentifier) | |||
{ | |||
if (!(packet is IMqttPacketWithIdentifier requestPacketWithIdentifier)) | |||
{ | |||
throw new InvalidOperationException("Response packet has PacketIdentifier but request packet does not."); | |||
} | |||
responsePacketWithIdentifier.PacketIdentifier = requestPacketWithIdentifier.PacketIdentifier; | |||
} | |||
return responsePacket; | |||
} | |||
} | |||
} |
@@ -6,7 +6,7 @@ namespace MQTTnet.Packets | |||
{ | |||
public sealed class MqttSubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier | |||
{ | |||
public ushort PacketIdentifier { get; set; } | |||
public ushort? PacketIdentifier { get; set; } | |||
public IList<MqttSubscribeReturnCode> SubscribeReturnCodes { get; } = new List<MqttSubscribeReturnCode>(); | |||
@@ -5,7 +5,7 @@ namespace MQTTnet.Packets | |||
{ | |||
public sealed class MqttSubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier | |||
{ | |||
public ushort PacketIdentifier { get; set; } | |||
public ushort? PacketIdentifier { get; set; } | |||
public IList<TopicFilter> TopicFilters { get; set; } = new List<TopicFilter>(); | |||
@@ -2,7 +2,7 @@ | |||
{ | |||
public sealed class MqttUnsubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier | |||
{ | |||
public ushort PacketIdentifier { get; set; } | |||
public ushort? PacketIdentifier { get; set; } | |||
public override string ToString() | |||
{ | |||
@@ -4,7 +4,7 @@ namespace MQTTnet.Packets | |||
{ | |||
public sealed class MqttUnsubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier | |||
{ | |||
public ushort PacketIdentifier { get; set; } | |||
public ushort? PacketIdentifier { get; set; } | |||
public IList<string> TopicFilters { get; set; } = new List<string>(); | |||
@@ -397,7 +397,12 @@ namespace MQTTnet.Serializer | |||
private static byte Serialize(MqttPubRelPacket packet, MqttPacketWriter writer) | |||
{ | |||
writer.Write(packet.PacketIdentifier); | |||
if (!packet.PacketIdentifier.HasValue) | |||
{ | |||
throw new MqttProtocolViolationException("PubRel packet has no packet identifier."); | |||
} | |||
writer.Write(packet.PacketIdentifier.Value); | |||
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02); | |||
} | |||
@@ -410,7 +415,12 @@ namespace MQTTnet.Serializer | |||
if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) | |||
{ | |||
writer.Write(packet.PacketIdentifier); | |||
if (!packet.PacketIdentifier.HasValue) | |||
{ | |||
throw new MqttProtocolViolationException("Publish packet has no packet identifier."); | |||
} | |||
writer.Write(packet.PacketIdentifier.Value); | |||
} | |||
else | |||
{ | |||
@@ -444,21 +454,36 @@ namespace MQTTnet.Serializer | |||
private static byte Serialize(MqttPubAckPacket packet, MqttPacketWriter writer) | |||
{ | |||
writer.Write(packet.PacketIdentifier); | |||
if (!packet.PacketIdentifier.HasValue) | |||
{ | |||
throw new MqttProtocolViolationException("PubAck packet has no packet identifier."); | |||
} | |||
writer.Write(packet.PacketIdentifier.Value); | |||
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck); | |||
} | |||
private static byte Serialize(MqttPubRecPacket packet, MqttPacketWriter writer) | |||
{ | |||
writer.Write(packet.PacketIdentifier); | |||
if (!packet.PacketIdentifier.HasValue) | |||
{ | |||
throw new MqttProtocolViolationException("PubRec packet has no packet identifier."); | |||
} | |||
writer.Write(packet.PacketIdentifier.Value); | |||
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec); | |||
} | |||
private static byte Serialize(MqttPubCompPacket packet, MqttPacketWriter writer) | |||
{ | |||
writer.Write(packet.PacketIdentifier); | |||
if (!packet.PacketIdentifier.HasValue) | |||
{ | |||
throw new MqttProtocolViolationException("PubComp packet has no packet identifier."); | |||
} | |||
writer.Write(packet.PacketIdentifier.Value); | |||
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp); | |||
} | |||
@@ -467,7 +492,12 @@ namespace MQTTnet.Serializer | |||
{ | |||
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3]."); | |||
writer.Write(packet.PacketIdentifier); | |||
if (!packet.PacketIdentifier.HasValue) | |||
{ | |||
throw new MqttProtocolViolationException("Subscribe packet has no packet identifier."); | |||
} | |||
writer.Write(packet.PacketIdentifier.Value); | |||
if (packet.TopicFilters?.Count > 0) | |||
{ | |||
@@ -483,7 +513,12 @@ namespace MQTTnet.Serializer | |||
private static byte Serialize(MqttSubAckPacket packet, MqttPacketWriter writer) | |||
{ | |||
writer.Write(packet.PacketIdentifier); | |||
if (!packet.PacketIdentifier.HasValue) | |||
{ | |||
throw new MqttProtocolViolationException("SubAck packet has no packet identifier."); | |||
} | |||
writer.Write(packet.PacketIdentifier.Value); | |||
if (packet.SubscribeReturnCodes?.Any() == true) | |||
{ | |||
@@ -500,7 +535,12 @@ namespace MQTTnet.Serializer | |||
{ | |||
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2]."); | |||
writer.Write(packet.PacketIdentifier); | |||
if (!packet.PacketIdentifier.HasValue) | |||
{ | |||
throw new MqttProtocolViolationException("Unsubscribe packet has no packet identifier."); | |||
} | |||
writer.Write(packet.PacketIdentifier.Value); | |||
if (packet.TopicFilters?.Any() == true) | |||
{ | |||
@@ -513,9 +553,14 @@ namespace MQTTnet.Serializer | |||
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02); | |||
} | |||
private static byte Serialize(IMqttPacketWithIdentifier packet, BinaryWriter writer) | |||
private static byte Serialize(MqttUnsubAckPacket packet, BinaryWriter writer) | |||
{ | |||
writer.Write(packet.PacketIdentifier); | |||
if (!packet.PacketIdentifier.HasValue) | |||
{ | |||
throw new MqttProtocolViolationException("UnsubAck packet has no packet identifier."); | |||
} | |||
writer.Write(packet.PacketIdentifier.Value); | |||
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck); | |||
} | |||
@@ -78,7 +78,7 @@ namespace MQTTnet.Server | |||
throw new InvalidOperationException(); // should not happen | |||
} | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false); | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { packet }).ConfigureAwait(false); | |||
_logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId); | |||
} | |||
@@ -3,6 +3,7 @@ using System.Collections.Generic; | |||
using System.Threading; | |||
using System.Threading.Tasks; | |||
using MQTTnet.Adapter; | |||
using MQTTnet.Client; | |||
using MQTTnet.Diagnostics; | |||
using MQTTnet.Exceptions; | |||
using MQTTnet.Internal; | |||
@@ -14,6 +15,7 @@ namespace MQTTnet.Server | |||
{ | |||
public sealed class MqttClientSession : IDisposable | |||
{ | |||
private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); | |||
private readonly IMqttServerOptions _options; | |||
private readonly IMqttNetLogger _logger; | |||
private readonly MqttRetainedMessagesManager _retainedMessagesManager; | |||
@@ -129,6 +131,11 @@ namespace MQTTnet.Server | |||
var publishPacket = applicationMessage.ToPublishPacket(); | |||
publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel; | |||
if (publishPacket.QualityOfServiceLevel > 0) | |||
{ | |||
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); | |||
} | |||
PendingMessagesQueue.Enqueue(publishPacket); | |||
} | |||
@@ -205,7 +212,7 @@ namespace MQTTnet.Server | |||
if (packet is MqttPingReqPacket) | |||
{ | |||
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPingRespPacket()); | |||
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttPingRespPacket() }); | |||
} | |||
if (packet is MqttPubRelPacket pubRelPacket) | |||
@@ -215,7 +222,12 @@ namespace MQTTnet.Server | |||
if (packet is MqttPubRecPacket pubRecPacket) | |||
{ | |||
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, pubRecPacket.CreateResponse<MqttPubRelPacket>()); | |||
var responsePacket = new MqttPubRelPacket | |||
{ | |||
PacketIdentifier = pubRecPacket.PacketIdentifier | |||
}; | |||
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { responsePacket }); | |||
} | |||
if (packet is MqttPubAckPacket || packet is MqttPubCompPacket) | |||
@@ -246,11 +258,11 @@ namespace MQTTnet.Server | |||
private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) | |||
{ | |||
var subscribeResult = await SubscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false); | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, subscribeResult.ResponsePacket).ConfigureAwait(false); | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { subscribeResult.ResponsePacket }).ConfigureAwait(false); | |||
if (subscribeResult.CloseConnection) | |||
{ | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttDisconnectPacket()).ConfigureAwait(false); | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttDisconnectPacket() }).ConfigureAwait(false); | |||
await StopAsync().ConfigureAwait(false); | |||
} | |||
@@ -260,7 +272,7 @@ namespace MQTTnet.Server | |||
private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) | |||
{ | |||
var unsubscribeResult = await SubscriptionsManager.UnsubscribeAsync(unsubscribePacket).ConfigureAwait(false); | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, unsubscribeResult); | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { unsubscribeResult }); | |||
} | |||
private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters) | |||
@@ -302,7 +314,7 @@ namespace MQTTnet.Server | |||
await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false); | |||
var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }; | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false); | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false); | |||
} | |||
private async Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken) | |||
@@ -311,13 +323,13 @@ namespace MQTTnet.Server | |||
await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false); | |||
var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }; | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false); | |||
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false); | |||
} | |||
private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) | |||
{ | |||
var response = new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }; | |||
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response); | |||
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }); | |||
} | |||
} | |||
} |
@@ -53,10 +53,10 @@ namespace MQTTnet.Server | |||
var connectReturnCode = ValidateConnection(connectPacket); | |||
if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted) | |||
{ | |||
await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket | |||
await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttConnAckPacket | |||
{ | |||
ConnectReturnCode = connectReturnCode | |||
}).ConfigureAwait(false); | |||
}}).ConfigureAwait(false); | |||
return; | |||
} | |||
@@ -64,11 +64,11 @@ namespace MQTTnet.Server | |||
var result = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false); | |||
clientSession = result.Session; | |||
await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket | |||
await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttConnAckPacket | |||
{ | |||
ConnectReturnCode = connectReturnCode, | |||
IsSessionPresent = result.IsExistingSession | |||
}).ConfigureAwait(false); | |||
}}).ConfigureAwait(false); | |||
ClientConnectedCallback?.Invoke(new ConnectedMqttClient | |||
{ | |||
@@ -14,7 +14,7 @@ namespace MQTTnet.Server | |||
private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>(); | |||
private readonly IMqttServerOptions _options; | |||
private readonly string _clientId; | |||
public MqttClientSubscriptionsManager(IMqttServerOptions options, string clientId) | |||
{ | |||
_options = options ?? throw new ArgumentNullException(nameof(options)); | |||
@@ -30,7 +30,11 @@ namespace MQTTnet.Server | |||
var result = new MqttClientSubscribeResult | |||
{ | |||
ResponsePacket = subscribePacket.CreateResponse<MqttSubAckPacket>(), | |||
ResponsePacket = new MqttSubAckPacket | |||
{ | |||
PacketIdentifier = subscribePacket.PacketIdentifier | |||
}, | |||
CloseConnection = false | |||
}; | |||
@@ -87,7 +91,10 @@ namespace MQTTnet.Server | |||
_semaphore.Release(); | |||
} | |||
return unsubscribePacket.CreateResponse<MqttUnsubAckPacket>(); | |||
return new MqttUnsubAckPacket | |||
{ | |||
PacketIdentifier = unsubscribePacket.PacketIdentifier | |||
}; | |||
} | |||
public async Task<CheckSubscriptionsResult> CheckSubscriptionsAsync(MqttApplicationMessage applicationMessage) | |||