diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index e8a62e4e..95ea4711 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -84,7 +84,7 @@ namespace MQTTnet.AspNetCore _output = Connection.Transport.Output; } - public Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken) + public Task DisconnectAsync(CancellationToken cancellationToken) { _input?.Complete(); _output?.Complete(); diff --git a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs index a7ad56b0..e46f4cb2 100644 --- a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs +++ b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs @@ -205,25 +205,36 @@ namespace MQTTnet.Extensions.WebSocket4Net _webSocket.Open(); #pragma warning restore AsyncFixer02 // Long-running or blocking operations inside an async method - var exception = await MqttTaskTimeout.WaitAsync(c => + using (var timeoutCts = new CancellationTokenSource(_clientOptions.CommunicationTimeout)) + using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(timeoutCts.Token, cancellationToken)) { - c.Register(() => taskCompletionSource.TrySetCanceled()); - return taskCompletionSource.Task; - }, _clientOptions.CommunicationTimeout, cancellationToken).ConfigureAwait(false); - - if (exception != null) - { - if (exception is AuthenticationException authenticationException) + using (linkedCts.Token.Register(() => taskCompletionSource.TrySetCanceled())) { - throw new MqttCommunicationException(authenticationException.InnerException); - } + try + { + await taskCompletionSource.Task.ConfigureAwait(false); + } + catch (Exception exception) + { + var timeoutReached = timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested; + if (timeoutReached) + { + throw new MqttCommunicationTimedOutException(exception); + } + + if (exception is AuthenticationException authenticationException) + { + throw new MqttCommunicationException(authenticationException.InnerException); + } - if (exception is OperationCanceledException) - { - throw new MqttCommunicationTimedOutException(); - } + if (exception is OperationCanceledException) + { + throw new MqttCommunicationTimedOutException(); + } - throw new MqttCommunicationException(exception); + throw new MqttCommunicationException(exception); + } + } } } catch (OperationCanceledException) diff --git a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs index c6026a56..f6d9b51b 100644 --- a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs @@ -25,7 +25,7 @@ namespace MQTTnet.Adapter Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken); - Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken); + Task DisconnectAsync(CancellationToken cancellationToken); Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken); diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 8ca09964..9b22973b 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -76,40 +76,28 @@ namespace MQTTnet.Adapter } catch (Exception exception) { - if (IsWrappedException(exception)) + if (!WrapAndThrowException(exception)) { throw; } - - WrapAndThrowException(exception); } } - public async Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken) + public async Task DisconnectAsync(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); ThrowIfDisposed(); try { - if (timeout == TimeSpan.Zero) - { - await _channel.DisconnectAsync(cancellationToken).ConfigureAwait(false); - } - else - { - await MqttTaskTimeout.WaitAsync( - t => _channel.DisconnectAsync(t), timeout, cancellationToken).ConfigureAwait(false); - } + await _channel.DisconnectAsync(cancellationToken).ConfigureAwait(false); } catch (Exception exception) { - if (IsWrappedException(exception)) + if (!WrapAndThrowException(exception)) { throw; } - - WrapAndThrowException(exception); } } @@ -136,12 +124,10 @@ namespace MQTTnet.Adapter } catch (Exception exception) { - if (IsWrappedException(exception)) + if (!WrapAndThrowException(exception)) { throw; } - - WrapAndThrowException(exception); } finally { @@ -192,12 +178,10 @@ namespace MQTTnet.Adapter } catch (Exception exception) { - if (IsWrappedException(exception)) + if (!WrapAndThrowException(exception)) { throw; } - - WrapAndThrowException(exception); } return null; @@ -386,17 +370,17 @@ namespace MQTTnet.Adapter return value; } - - static bool IsWrappedException(Exception exception) - { - return exception is OperationCanceledException || - exception is MqttCommunicationTimedOutException || - exception is MqttCommunicationException || - exception is MqttProtocolViolationException; - } - - static void WrapAndThrowException(Exception exception) + + static bool WrapAndThrowException(Exception exception) { + if (exception is OperationCanceledException || + exception is MqttCommunicationTimedOutException || + exception is MqttCommunicationException || + exception is MqttProtocolViolationException) + { + return false; + } + if (exception is IOException && exception.InnerException is SocketException innerException) { exception = innerException; diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 8a71eab0..2e926fe9 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -23,6 +23,7 @@ namespace MQTTnet.Client { public class MqttClient : Disposable, IMqttClient { + readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); readonly object _disconnectLock = new object(); @@ -156,8 +157,7 @@ namespace MQTTnet.Client if (clientWasConnected) { - var disconnectPacketFactory = new MqttDisconnectPacketFactory(); - var disconnectPacket = disconnectPacketFactory.Create(options); + var disconnectPacket = _packetFactories.Disconnect.Create(options); await SendAsync(disconnectPacket, cancellationToken).ConfigureAwait(false); } } @@ -207,8 +207,7 @@ namespace MQTTnet.Client ThrowIfDisposed(); ThrowIfNotConnected(); - var subscribePacketFactory = new MqttSubscribePacketFactory(); - var subscribePacket = subscribePacketFactory.Create(options); + var subscribePacket = _packetFactories.Subscribe.Create(options); subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); var subAckPacket = await SendAndReceiveAsync(subscribePacket, cancellationToken).ConfigureAwait(false); @@ -223,9 +222,8 @@ namespace MQTTnet.Client ThrowIfDisposed(); ThrowIfNotConnected(); - - var unsubscribeFactory = new MqttUnsubscribePacketFactory(); - var unsubscribePacket = unsubscribeFactory.Create(options); + + var unsubscribePacket = _packetFactories.Unsubscribe.Create(options); unsubscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); var unsubAckPacket = await SendAndReceiveAsync(unsubscribePacket, cancellationToken).ConfigureAwait(false); @@ -243,10 +241,7 @@ namespace MQTTnet.Client ThrowIfDisposed(); ThrowIfNotConnected(); - var publishPacketFactory = new MqttPublishPacketFactory(); - var publishPacket = publishPacketFactory.Create(applicationMessage); - - //_adapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(applicationMessage); + var publishPacket = _packetFactories.Publish.Create(applicationMessage); switch (applicationMessage.QualityOfServiceLevel) { @@ -303,8 +298,7 @@ namespace MQTTnet.Client try { - var connectPacketFactory = new MqttConnectPacketFactory(); - var connectPacket = connectPacketFactory.Create(options); + var connectPacket = _packetFactories.Connect.Create(options); var connAckPacket = await SendAndReceiveAsync(connectPacket, cancellationToken).ConfigureAwait(false); @@ -336,7 +330,10 @@ namespace MQTTnet.Client void ThrowIfConnected(string message) { - if (IsConnected) throw new MqttProtocolViolationException(message); + if (IsConnected) + { + throw new MqttProtocolViolationException(message); + } } Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientConnectResult connectResult) @@ -360,7 +357,11 @@ namespace MQTTnet.Client if (_adapter != null) { _logger.Verbose("Disconnecting [Timeout={0}]", Options.CommunicationTimeout); - await _adapter.DisconnectAsync(Options.CommunicationTimeout, CancellationToken.None).ConfigureAwait(false); + + using (var timeout = new CancellationTokenSource(Options.CommunicationTimeout)) + { + await _adapter.DisconnectAsync(timeout.Token).ConfigureAwait(false); + } } _logger.Verbose("Disconnected from adapter."); @@ -434,7 +435,7 @@ namespace MQTTnet.Client packetIdentifier = packetWithIdentifier.PacketIdentifier; } - using (var packetAwaiter = _packetDispatcher.AddAwaitable(packetIdentifier)) + using (var packetAwaitable = _packetDispatcher.AddAwaitable(packetIdentifier)) { try { @@ -443,12 +444,12 @@ namespace MQTTnet.Client catch (Exception exception) { _logger.Warning(exception, "Error when sending request packet ({0}).", requestPacket.GetType().Name); - packetAwaiter.Fail(exception); + packetAwaitable.Fail(exception); } try { - return await packetAwaiter.WaitOneAsync(Options.CommunicationTimeout).ConfigureAwait(false); + return await packetAwaitable.WaitOneAsync(Options.CommunicationTimeout).ConfigureAwait(false); } catch (Exception exception) { @@ -686,8 +687,7 @@ namespace MQTTnet.Client { if (!eventArgs.ProcessingFailed) { - var pubAckPacketFactory = new MqttPubAckPacketFactory(); - var pubAckPacket = pubAckPacketFactory.Create(eventArgs); + var pubAckPacket = _packetFactories.PubAck.Create(eventArgs); return SendAsync(pubAckPacket, cancellationToken); } } @@ -695,8 +695,7 @@ namespace MQTTnet.Client { if (!eventArgs.ProcessingFailed) { - var pubRecPacketFactory = new MqttPubRecPacketFactory(); - var pubRecPacket = pubRecPacketFactory.Create(eventArgs); + var pubRecPacket = _packetFactories.PubRec.Create(eventArgs); return SendAsync(pubRecPacket, cancellationToken); } } @@ -714,8 +713,7 @@ namespace MQTTnet.Client { // The packet is unknown. Probably due to a restart of the client. // So wen send this to the server to trigger a full resend of the message. - var pubRelPacketFactory = new MqttPubRelPacketFactory(); - var pubRelPacket = pubRelPacketFactory.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound); + var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound); return SendAsync(pubRelPacket, cancellationToken); } @@ -724,8 +722,7 @@ namespace MQTTnet.Client Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) { - var pubCompPacketFactory = new MqttPubCompPacketFactory(); - var pubCompPacket =pubCompPacketFactory.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); return SendAsync(pubCompPacket, cancellationToken); } @@ -774,8 +771,7 @@ namespace MQTTnet.Client var pubRecPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); - var pubRelPacketFactory = new MqttPubRelPacketFactory(); - var pubRelPacket = pubRelPacketFactory.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); var pubCompPacket = await SendAndReceiveAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs index a5d9e6cd..d58ad581 100644 --- a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs +++ b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs @@ -1,12 +1,14 @@ - -using System.Collections.Generic; +using System.Collections.Generic; using MQTTnet.Packets; namespace MQTTnet.Client.Publishing { public sealed class MqttClientPublishResult { - public ushort? PacketIdentifier { get; set; } + /// + /// Gets the packet identifier which was used for this publish. + /// + public ushort PacketIdentifier { get; set; } /// /// Gets or sets the reason code. diff --git a/Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs b/Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs index defe2a96..9e6d2bb9 100644 --- a/Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs @@ -17,7 +17,6 @@ namespace MQTTnet.Formatter Password = clientOptions.Credentials?.Password, CleanSession = clientOptions.CleanSession, KeepAlivePeriod = (ushort) clientOptions.KeepAlivePeriod.TotalSeconds, - WillMessage = clientOptions.WillMessage, Properties = { AuthenticationMethod = clientOptions.AuthenticationMethod, @@ -32,6 +31,26 @@ namespace MQTTnet.Formatter } }; + if (clientOptions.WillMessage != null) + { + connectPacket.WillFlag = true; + connectPacket.WillTopic = clientOptions.WillMessage.Topic; + connectPacket.WillQoS = clientOptions.WillMessage.QualityOfServiceLevel; + connectPacket.WillMessage = clientOptions.WillMessage.Payload; + connectPacket.WillRetain = clientOptions.WillMessage.Retain; + connectPacket.WillProperties.ContentType = clientOptions.WillMessage.ContentType; + connectPacket.WillProperties.CorrelationData = clientOptions.WillMessage.CorrelationData; + connectPacket.WillProperties.ResponseTopic = clientOptions.WillMessage.ResponseTopic; + connectPacket.WillProperties.MessageExpiryInterval = clientOptions.WillMessage.MessageExpiryInterval; + connectPacket.WillProperties.PayloadFormatIndicator = clientOptions.WillMessage.PayloadFormatIndicator; + //connectPacket.WillProperties.WillDelayInterval = clientOptions.WillMessage.; + + if (clientOptions.WillMessage.UserProperties != null) + { + connectPacket.WillProperties.UserProperties.AddRange(clientOptions.WillMessage.UserProperties); + } + } + if (clientOptions.UserProperties != null) { connectPacket.Properties.UserProperties.AddRange(clientOptions.UserProperties); diff --git a/Source/MQTTnet/Formatter/MqttPacketFactories.cs b/Source/MQTTnet/Formatter/MqttPacketFactories.cs new file mode 100644 index 00000000..eb3be807 --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttPacketFactories.cs @@ -0,0 +1,29 @@ +namespace MQTTnet.Formatter +{ + public sealed class MqttPacketFactories + { + public MqttConnectPacketFactory Connect { get; } = new MqttConnectPacketFactory(); + + public MqttConnAckPacketFactory ConnAck { get; } = new MqttConnAckPacketFactory(); + + public MqttDisconnectPacketFactory Disconnect { get; } = new MqttDisconnectPacketFactory(); + + public MqttSubscribePacketFactory Subscribe { get; } = new MqttSubscribePacketFactory(); + + public MqttSubAckPacketFactory SubAck { get; } = new MqttSubAckPacketFactory(); + + public MqttUnsubscribePacketFactory Unsubscribe { get; } = new MqttUnsubscribePacketFactory(); + + public MqttUnsubAckPacketFactory UnsubAck { get; } = new MqttUnsubAckPacketFactory(); + + public MqttPublishPacketFactory Publish { get; } = new MqttPublishPacketFactory(); + + public MqttPubAckPacketFactory PubAck { get; } = new MqttPubAckPacketFactory(); + + public MqttPubRelPacketFactory PubRel { get; } = new MqttPubRelPacketFactory(); + + public MqttPubRecPacketFactory PubRec { get; } = new MqttPubRecPacketFactory(); + + public MqttPubCompPacketFactory PubComp { get; } = new MqttPubCompPacketFactory(); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs index fbcf1db3..d0d98ec1 100644 --- a/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs @@ -1,4 +1,5 @@ using System; +using MQTTnet.Exceptions; using MQTTnet.Packets; namespace MQTTnet.Formatter @@ -41,5 +42,36 @@ namespace MQTTnet.Formatter return packet; } + + public MqttPublishPacket Create(MqttConnectPacket connectPacket) + { + if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket)); + + if (!connectPacket.WillFlag) + { + throw new MqttProtocolViolationException("The CONNECT packet contains no will message (WillFlag)."); + } + + var packet = new MqttPublishPacket + { + Topic = connectPacket.WillTopic, + Payload = connectPacket.WillMessage, + QualityOfServiceLevel = connectPacket.WillQoS, + Retain = connectPacket.WillRetain, + Dup = false, + Properties = + { + ContentType = connectPacket.WillProperties.ContentType, + CorrelationData = connectPacket.WillProperties.CorrelationData, + MessageExpiryInterval = connectPacket.WillProperties.MessageExpiryInterval, + PayloadFormatIndicator = connectPacket.WillProperties.PayloadFormatIndicator, + ResponseTopic = connectPacket.WillProperties.ResponseTopic + } + }; + + packet.Properties.UserProperties.AddRange(connectPacket.WillProperties.UserProperties); + + return packet; + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs index 42aeee20..0b526e05 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs @@ -11,7 +11,7 @@ namespace MQTTnet.Formatter.V3 public class MqttV310PacketFormatter : IMqttPacketFormatter { const int FixedHeaderSize = 1; - + static readonly MqttDisconnectPacket DisconnectPacket = new MqttDisconnectPacket(); readonly IMqttPacketWriter _packetWriter; @@ -30,7 +30,7 @@ namespace MQTTnet.Formatter.V3 _packetWriter.Seek(5); var fixedHeader = EncodePacket(packet, _packetWriter); - var remainingLength = (uint)(_packetWriter.Length - 5); + var remainingLength = (uint) (_packetWriter.Length - 5); var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength); @@ -57,7 +57,7 @@ namespace MQTTnet.Formatter.V3 throw new MqttProtocolViolationException($"The packet type is invalid ({controlPacketType})."); } - switch ((MqttControlPacketType)controlPacketType) + switch ((MqttControlPacketType) controlPacketType) { case MqttControlPacketType.Connect: return DecodeConnectPacket(receivedMqttPacket.BodyReader); case MqttControlPacketType.ConnAck: return DecodeConnAckPacket(receivedMqttPacket.BodyReader); @@ -187,7 +187,7 @@ namespace MQTTnet.Formatter.V3 var topicFilter = new MqttTopicFilter { Topic = body.ReadStringWithLengthPrefix(), - QualityOfServiceLevel = (MqttQualityOfServiceLevel)body.ReadByte() + QualityOfServiceLevel = (MqttQualityOfServiceLevel) body.ReadByte() }; packet.TopicFilters.Add(topicFilter); @@ -201,7 +201,7 @@ namespace MQTTnet.Formatter.V3 ThrowIfBodyIsEmpty(receivedMqttPacket.BodyReader); var retain = (receivedMqttPacket.FixedHeader & 0x1) > 0; - var qualityOfServiceLevel = (MqttQualityOfServiceLevel)(receivedMqttPacket.FixedHeader >> 1 & 0x3); + var qualityOfServiceLevel = (MqttQualityOfServiceLevel) (receivedMqttPacket.FixedHeader >> 1 & 0x3); var dup = (receivedMqttPacket.FixedHeader & 0x8) > 0; var topic = receivedMqttPacket.BodyReader.ReadStringWithLengthPrefix(); @@ -267,13 +267,12 @@ namespace MQTTnet.Formatter.V3 if (willFlag) { - packet.WillMessage = new MqttApplicationMessage - { - Topic = body.ReadStringWithLengthPrefix(), - Payload = body.ReadWithLengthPrefix(), - QualityOfServiceLevel = (MqttQualityOfServiceLevel)willQoS, - Retain = willRetain - }; + packet.WillFlag = true; + packet.WillQoS = (MqttQualityOfServiceLevel) willQoS; + packet.WillRetain = willRetain; + + packet.WillTopic = body.ReadStringWithLengthPrefix(); + packet.WillMessage = body.ReadWithLengthPrefix(); } if (usernameFlag) @@ -301,7 +300,7 @@ namespace MQTTnet.Formatter.V3 while (!body.EndOfStream) { - packet.ReturnCodes.Add((MqttSubscribeReturnCode)body.ReadByte()); + packet.ReturnCodes.Add((MqttSubscribeReturnCode) body.ReadByte()); } return packet; @@ -314,7 +313,7 @@ namespace MQTTnet.Formatter.V3 var packet = new MqttConnAckPacket(); body.ReadByte(); // Reserved. - packet.ReturnCode = (MqttConnectReturnCode)body.ReadByte(); + packet.ReturnCode = (MqttConnectReturnCode) body.ReadByte(); return packet; } @@ -351,12 +350,12 @@ namespace MQTTnet.Formatter.V3 connectFlags |= 0x2; } - if (packet.WillMessage != null) + if (packet.WillFlag) { connectFlags |= 0x4; - connectFlags |= (byte)((byte)packet.WillMessage.QualityOfServiceLevel << 3); + connectFlags |= (byte) ((byte) packet.WillQoS << 3); - if (packet.WillMessage.Retain) + if (packet.WillRetain) { connectFlags |= 0x20; } @@ -381,10 +380,10 @@ namespace MQTTnet.Formatter.V3 packetWriter.Write(packet.KeepAlivePeriod); packetWriter.WriteWithLengthPrefix(packet.ClientId); - if (packet.WillMessage != null) + if (packet.WillFlag) { - packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic); - packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload); + packetWriter.WriteWithLengthPrefix(packet.WillTopic); + packetWriter.WriteWithLengthPrefix(packet.WillMessage); } if (packet.Username != null) @@ -403,7 +402,7 @@ namespace MQTTnet.Formatter.V3 protected virtual byte EncodeConnAckPacket(MqttConnAckPacket packet, IMqttPacketWriter packetWriter) { packetWriter.Write(0); // Reserved. - packetWriter.Write((byte)packet.ReturnCode); + packetWriter.Write((byte) packet.ReturnCode); return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck); } @@ -455,7 +454,7 @@ namespace MQTTnet.Formatter.V3 fixedHeader |= 0x01; } - fixedHeader |= (byte)((byte)packet.QualityOfServiceLevel << 1); + fixedHeader |= (byte) ((byte) packet.QualityOfServiceLevel << 1); if (packet.Dup) { @@ -525,14 +524,14 @@ namespace MQTTnet.Formatter.V3 { throw new MqttProtocolViolationException("RetainAsPublished is not supported in 3.1.1."); } - + if (topicFilter.RetainHandling != MqttRetainHandling.SendAtSubscribe) { throw new MqttProtocolViolationException("RetainHandling is not supported in 3.1.1."); } - + packetWriter.WriteWithLengthPrefix(topicFilter.Topic); - packetWriter.Write((byte)topicFilter.QualityOfServiceLevel); + packetWriter.Write((byte) topicFilter.QualityOfServiceLevel); } } @@ -552,7 +551,7 @@ namespace MQTTnet.Formatter.V3 { foreach (var packetSubscribeReturnCode in packet.ReturnCodes) { - packetWriter.Write((byte)packetSubscribeReturnCode); + packetWriter.Write((byte) packetSubscribeReturnCode); } } @@ -606,4 +605,4 @@ namespace MQTTnet.Formatter.V3 } } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs index 09b0d0ec..bb4039fa 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs @@ -24,12 +24,12 @@ namespace MQTTnet.Formatter.V3 connectFlags |= 0x2; } - if (packet.WillMessage != null) + if (packet.WillFlag) { connectFlags |= 0x4; - connectFlags |= (byte)((byte)packet.WillMessage.QualityOfServiceLevel << 3); + connectFlags |= (byte)((byte)packet.WillQoS << 3); - if (packet.WillMessage.Retain) + if (packet.WillRetain) { connectFlags |= 0x20; } @@ -54,10 +54,10 @@ namespace MQTTnet.Formatter.V3 packetWriter.Write(packet.KeepAlivePeriod); packetWriter.WriteWithLengthPrefix(packet.ClientId); - if (packet.WillMessage != null) + if (packet.WillFlag) { - packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic); - packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload); + packetWriter.WriteWithLengthPrefix(packet.WillTopic); + packetWriter.WriteWithLengthPrefix(packet.WillMessage); } if (packet.Username != null) diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs index fb240645..4a0e11cd 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs @@ -1,5 +1,4 @@ using System; -using System.Collections.Generic; using MQTTnet.Adapter; using MQTTnet.Exceptions; using MQTTnet.Packets; @@ -72,11 +71,9 @@ namespace MQTTnet.Formatter.V5 if (willMessageFlag) { - packet.WillMessage = new MqttApplicationMessage - { - QualityOfServiceLevel = (MqttQualityOfServiceLevel)willMessageQoS, - Retain = willMessageRetainFlag - }; + packet.WillFlag = true; + packet.WillQoS = (MqttQualityOfServiceLevel) willMessageQoS; + packet.WillRetain = willMessageRetainFlag; } packet.KeepAlivePeriod = body.ReadTwoByteInteger(); @@ -128,7 +125,7 @@ namespace MQTTnet.Formatter.V5 packet.ClientId = body.ReadStringWithLengthPrefix(); - if (packet.WillMessage != null) + if (packet.WillFlag) { var willPropertiesReader = new MqttV500PropertiesReader(body); @@ -136,50 +133,31 @@ namespace MQTTnet.Formatter.V5 { if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.PayloadFormatIndicator) { - packet.WillMessage.PayloadFormatIndicator = propertiesReader.ReadPayloadFormatIndicator(); + packet.WillProperties.PayloadFormatIndicator = propertiesReader.ReadPayloadFormatIndicator(); } else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.MessageExpiryInterval) { - packet.WillMessage.MessageExpiryInterval = propertiesReader.ReadMessageExpiryInterval(); - } - else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.TopicAlias) - { - packet.WillMessage.TopicAlias = propertiesReader.ReadTopicAlias(); + packet.WillProperties.MessageExpiryInterval = propertiesReader.ReadMessageExpiryInterval(); } else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.ResponseTopic) { - packet.WillMessage.ResponseTopic = propertiesReader.ReadResponseTopic(); + packet.WillProperties.ResponseTopic = propertiesReader.ReadResponseTopic(); } else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.CorrelationData) { - packet.WillMessage.CorrelationData = propertiesReader.ReadCorrelationData(); - } - else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier) - { - if (packet.WillMessage.SubscriptionIdentifiers == null) - { - packet.WillMessage.SubscriptionIdentifiers = new List(); - } - - packet.WillMessage.SubscriptionIdentifiers.Add(propertiesReader.ReadSubscriptionIdentifier()); + packet.WillProperties.CorrelationData = propertiesReader.ReadCorrelationData(); } else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.ContentType) { - packet.WillMessage.ContentType = propertiesReader.ReadContentType(); + packet.WillProperties.ContentType = propertiesReader.ReadContentType(); } else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.WillDelayInterval) { - // This is a special case! - packet.Properties.WillDelayInterval = propertiesReader.ReadWillDelayInterval(); + packet.WillProperties.WillDelayInterval = propertiesReader.ReadWillDelayInterval(); } else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - if (packet.WillMessage.UserProperties == null) - { - packet.WillMessage.UserProperties = new List(); - } - - propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.WillProperties.UserProperties); } else { @@ -187,8 +165,8 @@ namespace MQTTnet.Formatter.V5 } } - packet.WillMessage.Topic = body.ReadStringWithLengthPrefix(); - packet.WillMessage.Payload = body.ReadWithLengthPrefix(); + packet.WillTopic = body.ReadStringWithLengthPrefix(); + packet.WillMessage = body.ReadWithLengthPrefix(); } if (usernameFlag) diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs index 141346a8..2a640887 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs @@ -88,12 +88,12 @@ namespace MQTTnet.Formatter.V5 connectFlags |= 0x2; } - if (packet.WillMessage != null) + if (packet.WillFlag) { connectFlags |= 0x4; - connectFlags |= (byte)((byte)packet.WillMessage.QualityOfServiceLevel << 3); + connectFlags |= (byte)((byte)packet.WillQoS << 3); - if (packet.WillMessage.Retain) + if (packet.WillRetain) { connectFlags |= 0x20; } @@ -135,25 +135,23 @@ namespace MQTTnet.Formatter.V5 packetWriter.WriteWithLengthPrefix(packet.ClientId); - if (packet.WillMessage != null) + if (packet.WillFlag) { var willPropertiesWriter = new MqttV500PropertiesWriter(); - willPropertiesWriter.WritePayloadFormatIndicator(packet.WillMessage.PayloadFormatIndicator); - willPropertiesWriter.WriteMessageExpiryInterval(packet.WillMessage.MessageExpiryInterval); - willPropertiesWriter.WriteTopicAlias(packet.WillMessage.TopicAlias); - willPropertiesWriter.WriteResponseTopic(packet.WillMessage.ResponseTopic); - willPropertiesWriter.WriteCorrelationData(packet.WillMessage.CorrelationData); - willPropertiesWriter.WriteSubscriptionIdentifiers(packet.WillMessage.SubscriptionIdentifiers); - willPropertiesWriter.WriteContentType(packet.WillMessage.ContentType); - willPropertiesWriter.WriteUserProperties(packet.WillMessage.UserProperties); + willPropertiesWriter.WritePayloadFormatIndicator(packet.WillProperties.PayloadFormatIndicator); + willPropertiesWriter.WriteMessageExpiryInterval(packet.WillProperties.MessageExpiryInterval); + willPropertiesWriter.WriteResponseTopic(packet.WillProperties.ResponseTopic); + willPropertiesWriter.WriteCorrelationData(packet.WillProperties.CorrelationData); + willPropertiesWriter.WriteContentType(packet.WillProperties.ContentType); + willPropertiesWriter.WriteUserProperties(packet.WillProperties.UserProperties); // This is a special case! willPropertiesWriter.WriteWillDelayInterval(packet.Properties?.WillDelayInterval); willPropertiesWriter.WriteTo(packetWriter); - packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic); - packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload); + packetWriter.WriteWithLengthPrefix(packet.WillTopic); + packetWriter.WriteWithLengthPrefix(packet.WillMessage); } if (packet.Username != null) diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs index bb8beace..34f5a637 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs @@ -143,7 +143,7 @@ namespace MQTTnet.Formatter.V5 return _body.ReadVariableLengthInteger(); } - public MqttPayloadFormatIndicator? ReadPayloadFormatIndicator() + public MqttPayloadFormatIndicator ReadPayloadFormatIndicator() { return (MqttPayloadFormatIndicator)_body.ReadByte(); } diff --git a/Source/MQTTnet/Internal/MqttPacketBus.cs b/Source/MQTTnet/Internal/MqttPacketBus.cs index 683a9d9a..986beed0 100644 --- a/Source/MQTTnet/Internal/MqttPacketBus.cs +++ b/Source/MQTTnet/Internal/MqttPacketBus.cs @@ -11,11 +11,11 @@ namespace MQTTnet.Internal { readonly object _syncRoot = new object(); - readonly Queue[] _partitions = new Queue[] + readonly LinkedList[] _partitions = new LinkedList[] { - new Queue(4096), - new Queue(1028), - new Queue(128) + new LinkedList(), + new LinkedList(), + new LinkedList() }; readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0); @@ -33,11 +33,14 @@ namespace MQTTnet.Internal } } - public int DataPacketsCount() + public int DataPacketsCount { - lock (_syncRoot) + get { - return _partitions[(int)MqttPacketBusPartition.Data].Count; + lock (_syncRoot) + { + return _partitions[(int) MqttPacketBusPartition.Data].Count; + } } } @@ -55,7 +58,7 @@ namespace MQTTnet.Internal lock (_syncRoot) { - _partitions[(int) partition].Enqueue(item); + _partitions[(int) partition].AddLast(item); } _semaphore.Release(); @@ -85,7 +88,9 @@ namespace MQTTnet.Internal if (_partitions[_activePartition].Count > 0) { - return _partitions[_activePartition].Dequeue(); + var item = _partitions[_activePartition].First; + _partitions[_activePartition].RemoveFirst(); + return item.Value; } } } @@ -102,5 +107,16 @@ namespace MQTTnet.Internal { _semaphore?.Dispose(); } + + public void Clear() + { + lock (_syncRoot) + { + foreach (var partition in _partitions) + { + partition.Clear(); + } + } + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Internal/MqttTaskTimeout.cs b/Source/MQTTnet/Internal/MqttTaskTimeout.cs index ba4ec7fd..dc7acea2 100644 --- a/Source/MQTTnet/Internal/MqttTaskTimeout.cs +++ b/Source/MQTTnet/Internal/MqttTaskTimeout.cs @@ -30,29 +30,5 @@ namespace MQTTnet.Internal } } } - - public static async Task WaitAsync(Func> action, TimeSpan timeout, CancellationToken cancellationToken) - { - if (action == null) throw new ArgumentNullException(nameof(action)); - - using (var timeoutCts = new CancellationTokenSource(timeout)) - using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(timeoutCts.Token, cancellationToken)) - { - try - { - return await action(linkedCts.Token).ConfigureAwait(false); - } - catch (OperationCanceledException exception) - { - var timeoutReached = timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested; - if (timeoutReached) - { - throw new MqttCommunicationTimedOutException(exception); - } - - throw; - } - } - } } } diff --git a/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs b/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs index 2fa0e7ce..7df0ac3f 100644 --- a/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs +++ b/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs @@ -113,7 +113,7 @@ namespace MQTTnet.LowLevelClient { try { - await _adapter.DisconnectAsync(_options.CommunicationTimeout, cancellationToken).ConfigureAwait(false); + await _adapter.DisconnectAsync(cancellationToken).ConfigureAwait(false); } catch (Exception exception) { diff --git a/Source/MQTTnet/MqttApplicationMessage.cs b/Source/MQTTnet/MqttApplicationMessage.cs index d1fe2297..a05ba1e3 100644 --- a/Source/MQTTnet/MqttApplicationMessage.cs +++ b/Source/MQTTnet/MqttApplicationMessage.cs @@ -51,7 +51,7 @@ namespace MQTTnet /// The feature is very similar to the HTTP header concept. /// Hint: MQTT 5 feature only. /// - public List UserProperties { get; set; } + public List UserProperties { get; set; } = new List(); /// /// Gets or sets the content type. @@ -74,7 +74,7 @@ namespace MQTTnet /// If no payload format indicator is provided, the default value is 0. /// Hint: MQTT 5 feature only. /// - public MqttPayloadFormatIndicator? PayloadFormatIndicator { get; set; } + public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; } = MqttPayloadFormatIndicator.Unspecified; /// /// Gets or sets the message expiry interval. @@ -84,7 +84,7 @@ namespace MQTTnet /// When the retained=true option is set on the PUBLISH message, this interval also defines how long a message is retained on a topic. /// Hint: MQTT 5 feature only. /// - public uint? MessageExpiryInterval { get; set; } + public uint MessageExpiryInterval { get; set; } /// /// Gets or sets the topic alias. diff --git a/Source/MQTTnet/MqttApplicationMessageBuilder.cs b/Source/MQTTnet/MqttApplicationMessageBuilder.cs index 18be1a23..d494195f 100644 --- a/Source/MQTTnet/MqttApplicationMessageBuilder.cs +++ b/Source/MQTTnet/MqttApplicationMessageBuilder.cs @@ -87,7 +87,7 @@ namespace MQTTnet /// When the retained=true option is set on the PUBLISH message, this interval also defines how long a message is retained on a topic. /// Hint: MQTT 5 feature only. /// - uint? _messageExpiryInterval; + uint _messageExpiryInterval; /// /// The payload format indicator. @@ -97,7 +97,7 @@ namespace MQTTnet /// If no payload format indicator is provided, the default value is 0. /// Hint: MQTT 5 feature only. /// - MqttPayloadFormatIndicator? _payloadFormatIndicator; + MqttPayloadFormatIndicator _payloadFormatIndicator; /// /// The user properties. diff --git a/Source/MQTTnet/Packets/MqttConnectPacket.cs b/Source/MQTTnet/Packets/MqttConnectPacket.cs index dcc6d447..b15941ed 100644 --- a/Source/MQTTnet/Packets/MqttConnectPacket.cs +++ b/Source/MQTTnet/Packets/MqttConnectPacket.cs @@ -1,4 +1,6 @@ -namespace MQTTnet.Packets +using MQTTnet.Protocol; + +namespace MQTTnet.Packets { public sealed class MqttConnectPacket : MqttBasePacket { @@ -14,13 +16,23 @@ /// Also called "Clean Start" in MQTTv5. /// public bool CleanSession { get; set; } - - public MqttApplicationMessage WillMessage { get; set; } /// /// Added in MQTTv5. /// public MqttConnectPacketProperties Properties { get; } = new MqttConnectPacketProperties(); + + public bool WillFlag { get; set; } + + public string WillTopic { get; set; } + + public byte[] WillMessage { get; set; } + + public MqttQualityOfServiceLevel WillQoS { get; set; } = MqttQualityOfServiceLevel.AtMostOnce; + + public bool WillRetain { get; set; } + + public MqttConnectPacketWillProperties WillProperties { get; } = new MqttConnectPacketWillProperties(); public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttConnectPacketWillProperties.cs b/Source/MQTTnet/Packets/MqttConnectPacketWillProperties.cs new file mode 100644 index 00000000..72b20e07 --- /dev/null +++ b/Source/MQTTnet/Packets/MqttConnectPacketWillProperties.cs @@ -0,0 +1,22 @@ +using System.Collections.Generic; +using MQTTnet.Protocol; + +namespace MQTTnet.Packets +{ + public sealed class MqttConnectPacketWillProperties + { + public uint WillDelayInterval { get; set; } + + public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; } = MqttPayloadFormatIndicator.Unspecified; + + public uint MessageExpiryInterval { get; set; } + + public string ContentType { get; set; } + + public string ResponseTopic { get; set; } + + public byte[] CorrelationData { get; set; } + + public List UserProperties { get; } = new List(); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs b/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs index 0278d909..e028c751 100644 --- a/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs @@ -5,9 +5,9 @@ namespace MQTTnet.Packets { public sealed class MqttPublishPacketProperties { - public MqttPayloadFormatIndicator? PayloadFormatIndicator { get; set; } + public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; } = MqttPayloadFormatIndicator.Unspecified; - public uint? MessageExpiryInterval { get; set; } + public uint MessageExpiryInterval { get; set; } public ushort? TopicAlias { get; set; } diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs index 5ea2cf52..3355b99c 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs @@ -19,12 +19,13 @@ namespace MQTTnet.Server.Internal { readonly Dictionary _topicAlias = new Dictionary(); readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); + readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); readonly MqttClientSessionsManager _sessionsManager; - + readonly MqttNetSourceLogger _logger; readonly IMqttServerOptions _serverOptions; - + readonly MqttConnectPacket _connectPacket; CancellationTokenSource _cancellationToken; @@ -39,10 +40,10 @@ namespace MQTTnet.Server.Internal { _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager)); - + ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter)); Endpoint = channelAdapter.Endpoint; - + Session = session ?? throw new ArgumentNullException(nameof(session)); _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket)); @@ -50,10 +51,10 @@ namespace MQTTnet.Server.Internal _logger = logger.WithSource(nameof(MqttClientConnection)); } - public string ClientId => _connectPacket.ClientId; + public string Id => _connectPacket.ClientId; + + public string Endpoint { get; } - public string Endpoint {get;} - public IMqttChannelAdapter ChannelAdapter { get; } public MqttClientConnectionStatistics Statistics { get; } = new MqttClientConnectionStatistics(); @@ -61,13 +62,13 @@ namespace MQTTnet.Server.Internal public bool IsRunning { get; private set; } public ushort KeepAlivePeriod => _connectPacket.KeepAlivePeriod; - + public MqttClientSession Session { get; } public bool IsTakenOver { get; set; } public bool IsCleanDisconnect { get; private set; } - + public async Task StopAsync(MqttClientDisconnectReason reason) { IsRunning = false; @@ -88,7 +89,7 @@ namespace MQTTnet.Server.Internal { ChannelAdapter.ResetStatistics(); } - + public void Dispose() { _cancellationToken.Dispose(); @@ -96,10 +97,11 @@ namespace MQTTnet.Server.Internal public async Task RunAsync() { - _logger.Info("Client '{0}': Session started.", ClientId); + _logger.Info("Client '{0}': Session started.", Id); + + Session.LatestConnectPacket = _connectPacket; + Session.WillMessageSent = false; - Session.WillMessage = _connectPacket.WillMessage; - using (var cancellationToken = new CancellationTokenSource()) { _cancellationToken = cancellationToken; @@ -108,7 +110,8 @@ namespace MQTTnet.Server.Internal { Task.Run(() => SendPacketsLoop(cancellationToken.Token), cancellationToken.Token).RunInBackground(_logger); - Session.IsCleanSession = false; + // TODO: Check!. + //Session.IsCleanSession = false; IsRunning = true; @@ -117,21 +120,22 @@ namespace MQTTnet.Server.Internal finally { IsRunning = false; - + cancellationToken.Cancel(); _cancellationToken = null; } } - + _packetDispatcher.CancelAll(); - if (!IsTakenOver && !IsCleanDisconnect && Session.WillMessage != null) + if (!IsTakenOver && !IsCleanDisconnect && Session.LatestConnectPacket.WillFlag && !Session.WillMessageSent) { - _sessionsManager.DispatchApplicationMessage(Session.WillMessage, this); - Session.WillMessage = null; + var publishPacket = _packetFactories.Publish.Create(Session.LatestConnectPacket); + _sessionsManager.DispatchPublishPacket(publishPacket, this); + Session.WillMessageSent = true; } - _logger.Info("Client '{0}': Connection stopped.", ClientId); + _logger.Info("Client '{0}': Connection stopped.", Id); } Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) @@ -163,23 +167,19 @@ namespace MQTTnet.Server.Internal } else if (packet is MqttPubAckPacket pubAckPacket) { - // TODO: Do anything! + Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier); } else if (packet is MqttPubCompPacket pubCompPacket) { - // TODO: Do anything! + Session.AcknowledgePublishPacket(pubCompPacket.PacketIdentifier); } else if (packet is MqttPubRecPacket pubRecPacket) { - // TODO: Do anything! - // TODO: Send pub rel! - var pubRelPacketFactory = new MqttPubRelPacketFactory(); - var pubRelPacket = pubRelPacketFactory.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); - Session.EnqueuePacket(new MqttPacketBusItem(pubRelPacket)); + HandleIncomingPubRecPacket(pubRecPacket); } else if (packet is MqttPubRelPacket pubRelPacket) { - await HandleIncomingPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false); + HandleIncomingPubRelPacket(pubRelPacket); } else if (packet is MqttSubscribePacket subscribePacket) { @@ -220,30 +220,25 @@ namespace MQTTnet.Server.Internal { if (exception is MqttCommunicationException) { - _logger.Warning(exception, "Client '{0}': Communication exception while receiving client packets.", ClientId); + _logger.Warning(exception, "Client '{0}': Communication exception while receiving client packets.", Id); } else { - _logger.Error(exception, "Client '{0}': Error while receiving client packets.", ClientId); + _logger.Error(exception, "Client '{0}': Error while receiving client packets.", Id); } } } async Task SendPacketsLoop(CancellationToken cancellationToken) { - //MqttQueuedApplicationMessage queuedApplicationMessage = null; - MqttPacketBusItem packetBusItem = null; - //MqttPublishPacket publishPacket = null; try { while (!cancellationToken.IsCancellationRequested) { - //queuedApplicationMessage = await Session.DequeuePacketAsync(cancellationToken).ConfigureAwait(false); - packetBusItem = await Session.DequeuePacketAsync(cancellationToken).ConfigureAwait(false); - + // Also check the cancellation token here because the dequeue is blocking and may take some time. if (cancellationToken.IsCancellationRequested) { @@ -254,7 +249,7 @@ namespace MQTTnet.Server.Internal { await SendPacketAsync(packetBusItem.Packet, cancellationToken).ConfigureAwait(false); packetBusItem.MarkAsDelivered(); - + _logger.Verbose("Sent packet '{0}'.", packetBusItem.Packet); } catch (OperationCanceledException) @@ -265,7 +260,7 @@ namespace MQTTnet.Server.Internal { packetBusItem.MarkAsFailed(exception); } - + // TODO: Move QoS 1 + 2 to other layer! // if (queuedApplicationMessage == null) // { @@ -305,7 +300,7 @@ namespace MQTTnet.Server.Internal // } // } - _logger.Verbose("Client '{0}': Queued application message sent.", ClientId); + _logger.Verbose("Client '{0}': Queued application message sent.", Id); } } catch (OperationCanceledException) @@ -315,15 +310,15 @@ namespace MQTTnet.Server.Internal { if (exception is MqttCommunicationTimedOutException) { - _logger.Warning(exception, "Client '{0}': Sending publish packet failed: Timeout.", ClientId); + _logger.Warning(exception, "Client '{0}': Sending publish packet failed: Timeout.", Id); } else if (exception is MqttCommunicationException) { - _logger.Warning(exception, "Client '{0}': Sending publish packet failed: Communication exception.", ClientId); + _logger.Warning(exception, "Client '{0}': Sending publish packet failed: Communication exception.", Id); } else { - _logger.Error(exception, "Client '{0}': Sending publish packet failed.", ClientId); + _logger.Error(exception, "Client '{0}': Sending publish packet failed.", Id); } if (packetBusItem?.Packet is MqttPublishPacket publishPacket) @@ -333,23 +328,8 @@ namespace MQTTnet.Server.Internal publishPacket.Dup = true; Session.EnqueuePacket(new MqttPacketBusItem(publishPacket)); } - - // if (queuedApplicationMessage != null) - // { - // queuedApplicationMessage.IsDuplicate = true; - // Session.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage); - // } } - // if (publishPacket?.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) - // { - // if (queuedApplicationMessage != null) - // { - // queuedApplicationMessage.IsDuplicate = true; - // Session.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage); - // } - // } - StopInternal(); } } @@ -387,35 +367,35 @@ namespace MQTTnet.Server.Internal _cancellationToken?.Cancel(); } - Task HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) + void HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket) { - var pubCompPacketFactory = new MqttPubCompPacketFactory(); - var pubCompPacket =pubCompPacketFactory.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); + Session.EnqueuePacket(new MqttPacketBusItem(pubRelPacket)); + } + + void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket) + { + var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); Session.EnqueuePacket(new MqttPacketBusItem(pubCompPacket)); - - //return SendPacketAsync(pubCompPacket, cancellationToken); - return PlatformAbstractionLayer.CompletedTask; } async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) { var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false); - var subAckPacketFactory = new MqttSubAckPacketFactory(); - var subAckPacket = subAckPacketFactory.Create(subscribePacket, subscribeResult); + var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult); Session.EnqueuePacket(new MqttPacketBusItem(subAckPacket)); - + if (subscribeResult.CloseConnection) { StopInternal(); return; } - var publishPacketFactory = new MqttPublishPacketFactory(); foreach (var retainedApplicationMessage in subscribeResult.RetainedApplicationMessages) { - var publishPacket = publishPacketFactory.Create(retainedApplicationMessage.ApplicationMessage); + var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage); Session.EnqueuePacket(new MqttPacketBusItem(publishPacket)); } } @@ -424,9 +404,8 @@ namespace MQTTnet.Server.Internal { var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false); - var unsubAckPacketFactory = new MqttUnsubAckPacketFactory(); - var unsubAckPacket = unsubAckPacketFactory.Create(unsubscribePacket, unsubscribeResult); - + var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult); + Session.EnqueuePacket(new MqttPacketBusItem(unsubAckPacket)); if (unsubscribeResult.CloseConnection) @@ -439,12 +418,8 @@ namespace MQTTnet.Server.Internal { HandleTopicAlias(publishPacket); - var applicationMessageFactory = new MqttApplicationMessageFactory(); - var applicationMessage = applicationMessageFactory.Create(publishPacket); + _sessionsManager.DispatchPublishPacket(publishPacket, this); - _sessionsManager.DispatchApplicationMessage(applicationMessage, this); - - // TODO: Check! switch (publishPacket.QualityOfServiceLevel) { case MqttQualityOfServiceLevel.AtMostOnce: @@ -454,15 +429,13 @@ namespace MQTTnet.Server.Internal } case MqttQualityOfServiceLevel.AtLeastOnce: { - var pubAckPacketFactory = new MqttPubAckPacketFactory(); - var pubAckPacket = pubAckPacketFactory.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); Session.EnqueuePacket(new MqttPacketBusItem(pubAckPacket)); break; } case MqttQualityOfServiceLevel.ExactlyOnce: { - var pubRecPacketFactory = new MqttPubRecPacketFactory(); - var pubRecPacket = pubRecPacketFactory.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); Session.EnqueuePacket(new MqttPacketBusItem(pubRecPacket)); break; } @@ -500,7 +473,7 @@ namespace MQTTnet.Server.Internal } } } - + async Task TrySendDisconnectPacket(MqttClientDisconnectReason reason) { try @@ -511,8 +484,7 @@ namespace MQTTnet.Server.Internal ReasonString = reason.ToString() }; - var disconnectPacketFactory = new MqttDisconnectPacketFactory(); - var disconnectPacket = disconnectPacketFactory.Create(disconnectOptions); + var disconnectPacket = _packetFactories.Disconnect.Create(disconnectOptions); using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout)) { @@ -521,10 +493,11 @@ namespace MQTTnet.Server.Internal } catch (Exception exception) { - _logger.Warning(exception, "Client '{{0}}': Error while sending DISCONNECT packet (Reason = {1}).", ClientId, reason); + _logger.Warning(exception, "Client '{{0}}': Error while sending DISCONNECT packet (Reason = {1}).", Id, reason); } } + // TODO: Fix! async Task InvokeClientMessageQueueInterceptor(MqttPublishPacket publishPacket, MqttQueuedApplicationMessage queuedApplicationMessage) { if (_serverOptions.ClientMessageQueueInterceptor == null) @@ -535,7 +508,7 @@ namespace MQTTnet.Server.Internal var context = new MqttClientMessageQueueInterceptorContext { SenderClientId = queuedApplicationMessage.SenderClientId, - ReceiverClientId = ClientId, + ReceiverClientId = Id, ApplicationMessage = queuedApplicationMessage.ApplicationMessage, SubscriptionQualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel }; diff --git a/Source/MQTTnet/Server/Internal/MqttClientSession.cs b/Source/MQTTnet/Server/Internal/MqttClientSession.cs index 8a189c51..c0894a24 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSession.cs @@ -1,10 +1,12 @@ using System; using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using MQTTnet.Client; using MQTTnet.Internal; using MQTTnet.Packets; +using MQTTnet.Protocol; namespace MQTTnet.Server.Internal { @@ -12,6 +14,8 @@ namespace MQTTnet.Server.Internal { readonly MqttPacketBus _packetBus = new MqttPacketBus(); + readonly Dictionary _publishPacketsWithoutAcknowledge = new Dictionary(); + readonly IMqttServerOptions _serverOptions; readonly MqttClientSessionsManager _clientSessionsManager; @@ -24,52 +28,70 @@ namespace MQTTnet.Server.Internal { _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager)); - ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); + Id = clientId ?? throw new ArgumentNullException(nameof(clientId)); Items = items ?? throw new ArgumentNullException(nameof(items)); SubscriptionsManager = new MqttClientSubscriptionsManager(this, serverOptions, eventDispatcher, retainedMessagesManager); - //ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions); } public event EventHandler Deleted; - public string ClientId { get; } + public string Id { get; } public MqttPacketIdentifierProvider PacketIdentifierProvider { get; } = new MqttPacketIdentifierProvider(); public DateTime CreatedTimestamp { get; } = DateTime.UtcNow; - public bool IsCleanSession { get; set; } = true; + //public bool IsCleanSession { get; set; } = true; - public MqttApplicationMessage WillMessage { get; set; } + public MqttConnectPacket LatestConnectPacket { get; set; } public MqttClientSubscriptionsManager SubscriptionsManager { get; } - //public MqttClientSessionApplicationMessagesQueue ApplicationMessagesQueue { get; } - public IDictionary Items { get; } - public void EnqueuePacket(MqttPacketBusItem packet) + public bool WillMessageSent { get; set; } + + public long PendingDataPacketsCount => _packetBus.DataPacketsCount; + + public void AcknowledgePublishPacket(ushort packetIdentifier) { - if (packet == null) throw new ArgumentNullException(nameof(packet)); + _publishPacketsWithoutAcknowledge.Remove(packetIdentifier); + } + + public void EnqueuePacket(MqttPacketBusItem packetBusItem) + { + if (packetBusItem == null) throw new ArgumentNullException(nameof(packetBusItem)); if (_packetBus.PacketsCount >= _serverOptions.MaxPendingMessagesPerClient) { - // TODO: Log! - return; + if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage) + { + // TODO: Log! + return; + } + else + { + // TODO: Implement. + } } - if (packet.Packet is MqttPublishPacket) + if (packetBusItem.Packet is MqttPublishPacket publishPacket) { - _packetBus.Enqueue(packet, MqttPacketBusPartition.Data); + if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) + { + _publishPacketsWithoutAcknowledge[publishPacket.PacketIdentifier] = publishPacket; + } + + _packetBus.Enqueue(packetBusItem, MqttPacketBusPartition.Data); } - else if (packet.Packet is MqttPingReqPacket || packet.Packet is MqttPingRespPacket) + else if (packetBusItem.Packet is MqttPingReqPacket || packetBusItem.Packet is MqttPingRespPacket) { - _packetBus.Enqueue(packet, MqttPacketBusPartition.Health); + _packetBus.Enqueue(packetBusItem, MqttPacketBusPartition.Health); } else { - _packetBus.Enqueue(packet, MqttPacketBusPartition.Control); + _packetBus.Enqueue(packetBusItem, MqttPacketBusPartition.Control); } } @@ -80,13 +102,25 @@ namespace MQTTnet.Server.Internal public Task DeleteAsync() { - return _clientSessionsManager.DeleteSessionAsync(ClientId); + return _clientSessionsManager.DeleteSessionAsync(Id); } public void OnDeleted() { Deleted?.Invoke(this, EventArgs.Empty); } + + public void Recover() + { + // TODO: Keep the bus and only insert pending items again. + // TODO: Check if packet identifier must be restarted or not. + _packetBus.Clear(); + + foreach (var publishPacket in _publishPacketsWithoutAcknowledge.Values.ToList()) + { + EnqueuePacket(new MqttPacketBusItem(publishPacket)); + } + } public void Dispose() { diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index a18e3df6..8b10367c 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -19,7 +19,8 @@ namespace MQTTnet.Server.Internal public sealed class MqttClientSessionsManager : IDisposable { readonly BlockingCollection _messageQueue = new BlockingCollection(); - + readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); + readonly AsyncLock _createConnectionSyncRoot = new AsyncLock(); readonly Dictionary _clientConnections = new Dictionary(4096); readonly Dictionary _clientSessions = new Dictionary(4096); @@ -129,36 +130,39 @@ namespace MQTTnet.Server.Internal { if (clientConnection != null) { - if (clientConnection.ClientId != null) + if (clientConnection.Id != null) { // in case it is a takeover _clientConnections already contains the new connection if (!clientConnection.IsTakenOver) { lock (_clientConnections) { - _clientConnections.Remove(clientConnection.ClientId); + _clientConnections.Remove(clientConnection.Id); } if (!_options.EnablePersistentSessions) { - await DeleteSessionAsync(clientConnection.ClientId).ConfigureAwait(false); + await DeleteSessionAsync(clientConnection.Id).ConfigureAwait(false); } } } var endpoint = clientConnection.Endpoint; - if (clientConnection.ClientId != null && !clientConnection.IsTakenOver) + if (clientConnection.Id != null && !clientConnection.IsTakenOver) { // The event is fired at a separate place in case of a handover! await _eventDispatcher.SafeNotifyClientDisconnectedAsync( - clientConnection.ClientId, + clientConnection.Id, clientConnection.IsCleanDisconnect ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean, endpoint).ConfigureAwait(false); } } - await channelAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false); + using (var timeout = new CancellationTokenSource(_options.DefaultCommunicationTimeout)) + { + await channelAdapter.DisconnectAsync(timeout.Token).ConfigureAwait(false); + } } } @@ -221,11 +225,11 @@ namespace MQTTnet.Server.Internal return Task.FromResult((IList)result); } - public void DispatchApplicationMessage(MqttApplicationMessage applicationMessage, MqttClientConnection sender) + public void DispatchPublishPacket(MqttPublishPacket publishPacket, MqttClientConnection sender) { - if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - _messageQueue.Add(new MqttPendingApplicationMessage(applicationMessage, sender)); + _messageQueue.Add(new MqttPendingApplicationMessage(publishPacket, sender)); } public async Task SubscribeAsync(string clientId, ICollection topicFilters) @@ -242,11 +246,7 @@ namespace MQTTnet.Server.Internal foreach (var retainedApplicationMessage in subscribeResult.RetainedApplicationMessages) { - //clientSession.ApplicationMessagesQueue.Enqueue(retainedApplicationMessage); - - // TODO: Review! - var publishPacketFactory = new MqttPublishPacketFactory(); - var publishPacket = publishPacketFactory.Create(retainedApplicationMessage.ApplicationMessage); + var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage); clientSession.EnqueuePacket(new MqttPacketBusItem(publishPacket)); } } @@ -351,8 +351,10 @@ namespace MQTTnet.Server.Internal } var clientConnection = pendingApplicationMessage.Sender; - var senderClientId = clientConnection?.ClientId ?? _options.ClientId; - var applicationMessage = pendingApplicationMessage.ApplicationMessage; + var senderClientId = clientConnection?.Id ?? _options.ClientId; + + // TODO: Only use PUBLISH packets internally! + var applicationMessage = new MqttApplicationMessageFactory().Create(pendingApplicationMessage.PublishPacket); var interceptor = _options.ApplicationMessageInterceptor; if (interceptor != null) @@ -403,7 +405,7 @@ namespace MQTTnet.Server.Internal continue; } - _logger.Verbose("Client '{0}': Queued application message with topic '{1}'.", clientSession.ClientId, applicationMessage.Topic); + _logger.Verbose("Client '{0}': Queued application message with topic '{1}'.", clientSession.Id, applicationMessage.Topic); // var queuedApplicationMessage = new MqttQueuedApplicationMessage // { @@ -411,16 +413,12 @@ namespace MQTTnet.Server.Internal // SubscriptionQualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel, // SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers // }; - - - - var publishPacketFactory = new MqttPublishPacketFactory(); - var publishPacket = publishPacketFactory.Create(applicationMessage); + + var publishPacket = _packetFactories.Publish.Create(applicationMessage); publishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel; publishPacket.Properties.SubscriptionIdentifiers.AddRange(checkSubscriptionsResult.SubscriptionIdentifiers); - //clientSession.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage); - + if (publishPacket.QualityOfServiceLevel > 0) { publishPacket.PacketIdentifier = clientSession.PacketIdentifierProvider.GetNextPacketIdentifier(); @@ -458,34 +456,6 @@ namespace MQTTnet.Server.Internal _logger.Error(exception, "Unhandled exception while processing next queued application message."); } } - - // public async Task CreatePublishPacket(MqttQueuedApplicationMessage queuedApplicationMessage) - // { - // var publishPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(queuedApplicationMessage.ApplicationMessage); - // publishPacket.QualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel; - // - // if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500) - // { - // publishPacket.Properties.SubscriptionIdentifiers.AddRange(queuedApplicationMessage.SubscriptionIdentifiers ?? new List()); - // } - // - // // Set the retain flag to true according to [MQTT-3.3.1-8] and [MQTT-3.3.1-9]. - // publishPacket.Retain = queuedApplicationMessage.IsRetainedMessage; - // - // publishPacket = await InvokeClientMessageQueueInterceptor(publishPacket, queuedApplicationMessage).ConfigureAwait(false); - // if (publishPacket == null) - // { - // // The interceptor has decided that the message is not relevant and will be fully ignored. - // return null; - // } - // - // if (publishPacket.QualityOfServiceLevel > 0) - // { - // publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); - // } - // - // return publishPacket; - // } async Task ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter) { @@ -547,6 +517,7 @@ namespace MQTTnet.Server.Internal { _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId); connAckPacket.IsSessionPresent = true; + session.Recover(); } } @@ -565,7 +536,7 @@ namespace MQTTnet.Server.Internal if (existingConnection != null) { - await _eventDispatcher.SafeNotifyClientDisconnectedAsync(existingConnection.ClientId, MqttClientDisconnectType.Takeover, existingConnection.Endpoint); + await _eventDispatcher.SafeNotifyClientDisconnectedAsync(existingConnection.Id, MqttClientDisconnectType.Takeover, existingConnection.Endpoint); existingConnection.IsTakenOver = true; await existingConnection.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false); @@ -591,7 +562,7 @@ namespace MQTTnet.Server.Internal } else { - senderClientId = clientConnection.ClientId; + senderClientId = clientConnection.Id; sessionItems = clientConnection.Session.Items; } @@ -599,7 +570,7 @@ namespace MQTTnet.Server.Internal { ClientId = senderClientId, SessionItems = sessionItems, - AcceptPublish = true, + ProcessPublish = true, ApplicationMessage = applicationMessage, CloseConnection = false }; diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index 5c895cdc..6695b442 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -16,7 +16,7 @@ namespace MQTTnet.Server.Internal // completely by swapping references etc. readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim(); readonly Dictionary _subscriptions = new Dictionary(4096); - + readonly MqttClientSession _clientSession; readonly IMqttServerOptions _options; readonly MqttServerEventDispatcher _eventDispatcher; @@ -67,7 +67,7 @@ namespace MQTTnet.Server.Internal // must be closed. So do not revert to false! result.CloseConnection = true; } - + if (!acceptSubscription || string.IsNullOrEmpty(finalTopicFilter.Topic)) { continue; @@ -81,7 +81,7 @@ namespace MQTTnet.Server.Internal interceptorContext.ReasonCode); await _eventDispatcher - .SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter) + .SafeNotifyClientSubscribedTopicAsync(_clientSession.Id, finalTopicFilter) .ConfigureAwait(false); FilterRetainedApplicationMessages(retainedApplicationMessages, subscription, result); @@ -103,7 +103,7 @@ namespace MQTTnet.Server.Internal foreach (var topicFilter in unsubscribePacket.TopicFilters) { _subscriptions.TryGetValue(topicFilter, out var existingSubscription); - + var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription, cancellationToken).ConfigureAwait(false); var acceptUnsubscription = interceptorContext.ReasonCode == MqttUnsubscribeReasonCode.Success; @@ -115,7 +115,7 @@ namespace MQTTnet.Server.Internal // must be closed. So do not revert to false! result.CloseConnection = true; } - + if (!acceptUnsubscription) { continue; @@ -134,7 +134,7 @@ namespace MQTTnet.Server.Internal foreach (var topicFilter in unsubscribePacket.TopicFilters) { - await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false); + await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.Id, topicFilter).ConfigureAwait(false); } return result; @@ -154,7 +154,7 @@ namespace MQTTnet.Server.Internal _subscriptionsLock.ExitReadLock(); } - var senderIsReceiver = string.Equals(senderClientId, _clientSession.ClientId); + var senderIsReceiver = string.Equals(senderClientId, _clientSession.Id); var qosLevels = new HashSet(); var subscriptionIdentifiers = new HashSet(); @@ -296,7 +296,7 @@ namespace MQTTnet.Server.Internal { var context = new MqttSubscriptionInterceptorContext { - ClientId = _clientSession.ClientId, + ClientId = _clientSession.Id, TopicFilter = topicFilter, SessionItems = _clientSession.Items, Session = new MqttSessionStatus(_clientSession), @@ -343,7 +343,7 @@ namespace MQTTnet.Server.Internal { var context = new MqttUnsubscriptionInterceptorContext { - ClientId = _clientSession.ClientId, + ClientId = _clientSession.Id, Topic = topicFilter, SessionItems = _clientSession.Items, CancellationToken = cancellationToken @@ -369,8 +369,7 @@ namespace MQTTnet.Server.Internal return context; } - static MqttQualityOfServiceLevel GetEffectiveQoS(MqttQualityOfServiceLevel qosLevel, - HashSet subscribedQoSLevels) + static MqttQualityOfServiceLevel GetEffectiveQoS(MqttQualityOfServiceLevel qosLevel, ICollection subscribedQoSLevels) { MqttQualityOfServiceLevel effectiveQoS; if (subscribedQoSLevels.Contains(qosLevel)) diff --git a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs index 66211c2d..b7a848b1 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs @@ -101,7 +101,7 @@ namespace MQTTnet.Server.Internal return; } - _logger.Warning("Client '{0}': Did not receive any packet or keep alive signal.", connection.ClientId); + _logger.Warning("Client '{0}': Did not receive any packet or keep alive signal.", connection.Id); // Execute the disconnection in background so that the keep alive monitor can continue // with checking other connections. @@ -112,7 +112,7 @@ namespace MQTTnet.Server.Internal } catch (Exception exception) { - _logger.Error(exception, "Client {0}: Unhandled exception while checking keep alive timeouts.", connection.ClientId); + _logger.Error(exception, "Client {0}: Unhandled exception while checking keep alive timeouts.", connection.Id); } } } diff --git a/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs b/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs index c48b35ab..c0f0135d 100644 --- a/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs +++ b/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs @@ -41,12 +41,6 @@ namespace MQTTnet.Server public string Password => Encoding.UTF8.GetString(RawPassword ?? PlatformAbstractionLayer.EmptyByteArray); - /// - /// Gets or sets the will delay interval. - /// This is the time between the client disconnect and the time the will message will be sent. - /// - public MqttApplicationMessage WillMessage => _connectPacket.WillMessage; - /// /// Gets or sets a value indicating whether clean sessions are used or not. /// When a client connects to a broker it can connect using either a non persistent connection (clean session) or a persistent connection. @@ -123,7 +117,7 @@ namespace MQTTnet.Server /// Gets or sets the will delay interval. /// This is the time between the client disconnect and the time the will message will be sent. /// - public uint? WillDelayInterval => _connectPacket.Properties?.WillDelayInterval; + public uint? WillDelayInterval => _connectPacket.WillProperties?.WillDelayInterval; /// /// Gets or sets a key/value collection that can be used to share data within the scope of this session. diff --git a/Source/MQTTnet/Server/MqttPendingApplicationMessage.cs b/Source/MQTTnet/Server/MqttPendingApplicationMessage.cs index 52ed9c76..2c12e4b0 100644 --- a/Source/MQTTnet/Server/MqttPendingApplicationMessage.cs +++ b/Source/MQTTnet/Server/MqttPendingApplicationMessage.cs @@ -1,17 +1,18 @@ -using MQTTnet.Server.Internal; +using MQTTnet.Packets; +using MQTTnet.Server.Internal; namespace MQTTnet.Server { public sealed class MqttPendingApplicationMessage { - public MqttPendingApplicationMessage(MqttApplicationMessage applicationMessage, MqttClientConnection sender) + public MqttPendingApplicationMessage(MqttPublishPacket publishPacket, MqttClientConnection sender) { Sender = sender; - ApplicationMessage = applicationMessage; + PublishPacket = publishPacket; } public MqttClientConnection Sender { get; } - public MqttApplicationMessage ApplicationMessage { get; } + public MqttPublishPacket PublishPacket { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 54827e27..48338716 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -1,7 +1,6 @@ using MQTTnet.Adapter; using MQTTnet.Client.Publishing; using MQTTnet.Client.Receiving; -using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Protocol; using MQTTnet.Server.Status; @@ -11,6 +10,7 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; using MQTTnet.Diagnostics.Logger; +using MQTTnet.Formatter; using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.Server.Internal; @@ -19,6 +19,7 @@ namespace MQTTnet.Server { public class MqttServer : Disposable, IMqttServer { + readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); readonly MqttServerEventDispatcher _eventDispatcher; readonly ICollection _adapters; readonly IMqttNetLogger _rootLogger; @@ -148,7 +149,8 @@ namespace MQTTnet.Server ThrowIfNotStarted(); - _clientSessionsManager.DispatchApplicationMessage(applicationMessage, null); + var publishPacket = _packetFactories.Publish.Create(applicationMessage); + _clientSessionsManager.DispatchPublishPacket(publishPacket, null); return Task.FromResult(new MqttClientPublishResult()); } diff --git a/Source/MQTTnet/Server/Status/MqttClientStatus.cs b/Source/MQTTnet/Server/Status/MqttClientStatus.cs index ca729de0..3d9b9eef 100644 --- a/Source/MQTTnet/Server/Status/MqttClientStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttClientStatus.cs @@ -19,7 +19,7 @@ namespace MQTTnet.Server.Status /// Gets or sets the client identifier. /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// - public string ClientId => _connection.ClientId; + public string ClientId => _connection.Id; public string Endpoint => _connection.Endpoint; diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index f13b6826..abb62047 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -24,10 +24,9 @@ namespace MQTTnet.Server.Status remove => _session.Deleted += value; } - public string ClientId => _session.ClientId; + public string ClientId => _session.Id; - // TODO: Fix! - public long PendingApplicationMessagesCount => 0; // _session.ApplicationMessagesQueue.Count; + public long PendingApplicationMessagesCount => _session.PendingDataPacketsCount; public DateTime CreatedTimestamp => _session.CreatedTimestamp; diff --git a/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs index 8ee99b8a..5e96c0cd 100644 --- a/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs @@ -23,10 +23,8 @@ using MQTTnet.Tests.Mockups; namespace MQTTnet.Tests.Client { [TestClass] - public class Client_Tests + public class Client_Tests : BaseTestClass { - public TestContext TestContext { get; set; } - [TestMethod] public async Task Ensure_Queue_Drain() { @@ -690,7 +688,7 @@ namespace MQTTnet.Tests.Client [TestMethod] public async Task Message_Send_Retry() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { testEnvironment.IgnoreClientLogErrors = true; testEnvironment.IgnoreServerLogErrors = true; @@ -914,40 +912,5 @@ namespace MQTTnet.Tests.Client Assert.IsTrue(success); } } - - [DataTestMethod] - [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)] - [DataRow(MqttQualityOfServiceLevel.AtLeastOnce)] - public async Task Retry_If_Not_PubAck(MqttQualityOfServiceLevel qos) - { - long count = 0; - using (var testEnvironment = new TestEnvironment(TestContext)) - { - await testEnvironment.StartServer(); - var publisher = await testEnvironment.ConnectClient(); - var subscriber = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId(qos.ToString())); - await subscriber.SubscribeAsync("#", qos); - - subscriber.UseApplicationMessageReceivedHandler(c => - { - c.AutoAcknowledge = false; - ++count; - Console.WriteLine("process"); - return Task.CompletedTask; - }); - - var pub = publisher.PublishAsync("a", null, qos); - - await Task.Delay(100); - await subscriber.DisconnectAsync(); - await subscriber.ReconnectAsync(); - await Task.Delay(100); - - var res = await pub; - - Assert.AreEqual(MqttClientPublishReasonCode.Success, res.ReasonCode); - Assert.AreEqual(2, count); - } - } } } diff --git a/Tests/MQTTnet.Core.Tests/Extension_Tests.cs b/Tests/MQTTnet.Core.Tests/Extension_Tests.cs index 68951000..3be5aad8 100644 --- a/Tests/MQTTnet.Core.Tests/Extension_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Extension_Tests.cs @@ -27,16 +27,7 @@ namespace MQTTnet.Tests ct => Task.Delay(TimeSpan.FromMilliseconds(500), ct).ContinueWith(t => 5, ct), TimeSpan.FromMilliseconds(100), CancellationToken.None); } - - [TestMethod] - public async Task TimeoutAfterCompleteInTime() - { - var result = await MqttTaskTimeout.WaitAsync( - ct => Task.Delay(TimeSpan.FromMilliseconds(100), ct).ContinueWith(t => 5, ct), - TimeSpan.FromMilliseconds(500), CancellationToken.None); - Assert.AreEqual(5, result); - } - + [TestMethod] public async Task TimeoutAfterWithInnerException() { diff --git a/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs b/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs index e2fe4832..bc30aba5 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs @@ -118,7 +118,7 @@ namespace MQTTnet.Tests.MQTTv5 await client.DisconnectAsync(); Assert.AreEqual(1, result.Items.Count); - Assert.AreEqual(MqttClientUnsubscribeResultCode.Success, result.Items[0].ReasonCode); + Assert.AreEqual(MqttClientUnsubscribeResultCode.Success, result.Items[0].ResultCode); } } diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapter.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapter.cs index 2a5be2e3..33658425 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapter.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapter.cs @@ -37,7 +37,7 @@ namespace MQTTnet.Tests.Mockups return Task.FromResult(0); } - public Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken) + public Task DisconnectAsync(CancellationToken cancellationToken) { return Task.FromResult(0); } diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs index 511f3dae..689fb3c2 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs @@ -9,7 +9,6 @@ using MQTTnet.Diagnostics.Logger; using MQTTnet.Exceptions; using MQTTnet.Formatter; using MQTTnet.Formatter.V3; -using MQTTnet.Formatter.V5; using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -94,13 +93,11 @@ namespace MQTTnet.Tests Username = "USER", KeepAlivePeriod = 123, CleanSession = true, - WillMessage = new MqttApplicationMessage - { - Topic = "My/last/will", - Payload = Encoding.UTF8.GetBytes("Good byte."), - QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, - Retain = true - } + WillFlag = true, + WillTopic = "My/last/will", + WillMessage = Encoding.UTF8.GetBytes("Good byte."), + WillQoS = MqttQualityOfServiceLevel.AtLeastOnce, + WillRetain = true }; SerializeAndCompare(p, "EDUABE1RVFQE7gB7AANYWVoADE15L2xhc3Qvd2lsbAAKR29vZCBieXRlLgAEVVNFUgAEUEFTUw=="); @@ -131,13 +128,11 @@ namespace MQTTnet.Tests Username = "USER", KeepAlivePeriod = 123, CleanSession = true, - WillMessage = new MqttApplicationMessage - { - Topic = "My/last/will", - Payload = Encoding.UTF8.GetBytes("Good byte."), - QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, - Retain = true - } + WillFlag = true, + WillTopic = "My/last/will", + WillMessage = Encoding.UTF8.GetBytes("Good byte."), + WillQoS = MqttQualityOfServiceLevel.AtLeastOnce, + WillRetain = true }; DeserializeAndCompare(p, "EDUABE1RVFQE7gB7AANYWVoADE15L2xhc3Qvd2lsbAAKR29vZCBieXRlLgAEVVNFUgAEUEFTUw=="); diff --git a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs index 69cb66e5..2a59c717 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs @@ -5,6 +5,7 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Client.Options; +using MQTTnet.Client.Publishing; using MQTTnet.Client.Subscribing; using MQTTnet.Protocol; using MQTTnet.Server; @@ -165,6 +166,46 @@ namespace MQTTnet.Tests.Server Assert.AreEqual("injected_one", messageReceivedHandler.ReceivedApplicationMessages[0].Topic); } } + + [DataTestMethod] + [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)] + [DataRow(MqttQualityOfServiceLevel.AtLeastOnce)] + public async Task Retry_If_Not_PubAck(MqttQualityOfServiceLevel qos) + { + long count = 0; + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(o => o.WithPersistentSessions()); + + var publisher = await testEnvironment.ConnectClient(); + + var subscriber = await testEnvironment.ConnectClient(o => o + .WithClientId(qos.ToString()) + .WithCleanSession(false)); + + subscriber.UseApplicationMessageReceivedHandler(c => + { + c.AutoAcknowledge = false; + ++count; + Console.WriteLine("process"); + return Task.CompletedTask; + }); + + await subscriber.SubscribeAsync("#", qos); + + var pub = publisher.PublishAsync("a", null, qos); + + await Task.Delay(100); + await subscriber.DisconnectAsync(); + await subscriber.ReconnectAsync(); + await Task.Delay(100); + + var res = await pub; + + Assert.AreEqual(MqttClientPublishReasonCode.Success, res.ReasonCode); + Assert.AreEqual(2, count); + } + } async Task TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options) { diff --git a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs index 00951789..e321fd29 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs @@ -64,7 +64,7 @@ namespace MQTTnet.TestApp.NetCore if (context.ApplicationMessage.Topic == "not_allowed_topic") { - context.AcceptPublish = false; + context.ProcessPublish = false; context.CloseConnection = true; } }),