diff --git a/Source/MQTTnet/Server/IMqttClientSession.cs b/Source/MQTTnet/Server/IMqttClientSession.cs index 6e29a2f2..51341e9a 100644 --- a/Source/MQTTnet/Server/IMqttClientSession.cs +++ b/Source/MQTTnet/Server/IMqttClientSession.cs @@ -1,22 +1,11 @@ using System; -using System.Collections.Generic; -using System.Threading.Tasks; -using MQTTnet.Adapter; namespace MQTTnet.Server { public interface IMqttClientSession : IDisposable { string ClientId { get; } - void FillStatus(MqttClientSessionStatus status); - void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage); - void ClearPendingApplicationMessages(); - - Task RunAsync(MqttApplicationMessage willMessage, int keepAliveInterval, IMqttChannelAdapter adapter); void Stop(MqttClientDisconnectType disconnectType); - - Task SubscribeAsync(IList topicFilters); - Task UnsubscribeAsync(IList topicFilters); } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index c915dbc0..860467cf 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -6,10 +6,10 @@ using MQTTnet.Adapter; using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; +using MQTTnet.Formatter; using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; -using MQTTnet.Serializer; namespace MQTTnet.Server { @@ -35,7 +35,8 @@ namespace MQTTnet.Server private string _adapterEndpoint; private MqttProtocolVersion? _adapterProtocolVersion; - + private IMqttChannelAdapter _channelAdapter; + public MqttClientSession( string clientId, IMqttServerOptions options, @@ -73,9 +74,9 @@ namespace MQTTnet.Server status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived; } - public Task RunAsync(MqttApplicationMessage willMessage, int keepAlivePeriod, IMqttChannelAdapter adapter) + public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) { - _workerTask = RunInternalAsync(willMessage, keepAlivePeriod, adapter); + _workerTask = RunInternalAsync(connectPacket, adapter); return _workerTask; } @@ -84,52 +85,6 @@ namespace MQTTnet.Server Stop(type, false); } - public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket) - { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - - var checkSubscriptionsResult = _subscriptionsManager.CheckSubscriptions(publishPacket.Topic, publishPacket.QualityOfServiceLevel); - if (!checkSubscriptionsResult.IsSubscribed) - { - return; - } - - publishPacket = new MqttPublishPacket - { - Topic = publishPacket.Topic, - Payload = publishPacket.Payload, - QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel, - Retain = publishPacket.Retain, - Dup = false - }; - - if (publishPacket.QualityOfServiceLevel > 0) - { - publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); - } - - if (_options.ClientMessageQueueInterceptor != null) - { - var context = new MqttClientMessageQueueInterceptorContext( - senderClientSession?.ClientId, - ClientId, - publishPacket.ToApplicationMessage()); - - _options.ClientMessageQueueInterceptor?.Invoke(context); - - if (!context.AcceptEnqueue || context.ApplicationMessage == null) - { - return; - } - - publishPacket.Topic = context.ApplicationMessage.Topic; - publishPacket.Payload = context.ApplicationMessage.Payload; - publishPacket.QualityOfServiceLevel = context.ApplicationMessage.QualityOfServiceLevel; - } - - _pendingPacketsQueue.Enqueue(publishPacket); - } - public Task SubscribeAsync(IList topicFilters) { if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); @@ -164,44 +119,11 @@ namespace MQTTnet.Server { _pendingPacketsQueue?.Dispose(); - _cancellationTokenSource?.Cancel (); + _cancellationTokenSource?.Cancel(); _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; } - - private void Stop(MqttClientDisconnectType type, bool isInsideSession) - { - try - { - var cts = _cancellationTokenSource; - if (cts == null || cts.IsCancellationRequested) - { - return; - } - - _cancellationTokenSource?.Cancel(false); - - _wasCleanDisconnect = type == MqttClientDisconnectType.Clean; - - if (_willMessage != null && !_wasCleanDisconnect) - { - _sessionsManager.EnqueueApplicationMessage(this, _willMessage.ToPublishPacket()); - } - - _willMessage = null; - - if (!isInsideSession) - { - _workerTask?.GetAwaiter().GetResult(); - } - } - finally - { - _logger.Info("Client '{0}': Disconnected (clean={1}).", ClientId, _wasCleanDisconnect); - _eventDispatcher.OnClientDisconnected(ClientId, _wasCleanDisconnect); - } - } - + private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) { if (adapter == null) throw new ArgumentNullException(nameof(adapter)); @@ -229,13 +151,14 @@ namespace MQTTnet.Server //end workaround _wasCleanDisconnect = false; - _willMessage = willMessage; + _willMessage = connectPacket.WillMessage; _pendingPacketsQueue.Start(adapter, _cancellationTokenSource.Token); - _keepAliveMonitor.Start(keepAlivePeriod, _cancellationTokenSource.Token); + _keepAliveMonitor.Start(connectPacket.KeepAlivePeriod, _cancellationTokenSource.Token); _adapterEndpoint = adapter.Endpoint; - _adapterProtocolVersion = adapter.PacketSerializer.ProtocolVersion; + _adapterProtocolVersion = adapter.PacketFormatterAdapter.ProtocolVersion; + _channelAdapter = adapter; while (!_cancellationTokenSource.IsCancellationRequested) { @@ -308,15 +231,10 @@ namespace MQTTnet.Server } finally { - adapter?.Dispose(); + adapter.Dispose(); } } - public void Stop(MqttClientDisconnectType type) - { - Stop(type, false); - } - private void Stop(MqttClientDisconnectType type, bool isInsideSession) { try @@ -346,12 +264,11 @@ namespace MQTTnet.Server finally { _logger.Info("Client '{0}': Session stopped.", ClientId); - - _sessionsManager.Server.OnClientDisconnected(ClientId, _wasCleanDisconnect); + _eventDispatcher.OnClientDisconnected(ClientId, _wasCleanDisconnect); } } - public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) + public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage, bool isRetainedApplicationMessage) { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); @@ -361,16 +278,38 @@ namespace MQTTnet.Server return; } - var publishPacket = _adapter.PacketFormatterAdapter.ConvertApplicationMessageToPublishPacket(applicationMessage); - publishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel; - publishPacket.Retain = false; - publishPacket.Dup = false; - + var publishPacket = _channelAdapter.PacketFormatterAdapter.ConvertApplicationMessageToPublishPacket(applicationMessage); + + // Set the retain flag to true according to [MQTT-3.3.1-8] and [MQTT-3.3.1-9]. + publishPacket.Retain = isRetainedApplicationMessage; + if (publishPacket.QualityOfServiceLevel > 0) { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); } + if (_options.ClientMessageQueueInterceptor != null) + { + var context = new MqttClientMessageQueueInterceptorContext( + senderClientSession?.ClientId, + ClientId, + applicationMessage); + + _options.ClientMessageQueueInterceptor?.Invoke(context); + + if (!context.AcceptEnqueue || context.ApplicationMessage == null) + { + return; + } + + publishPacket.Topic = context.ApplicationMessage.Topic; + publishPacket.Payload = context.ApplicationMessage.Payload; + publishPacket.QualityOfServiceLevel = context.ApplicationMessage.QualityOfServiceLevel; + } + + _pendingPacketsQueue.Enqueue(publishPacket); + } + private async Task TryDisconnectAdapterAsync(IMqttChannelAdapter adapter) { if (adapter == null) @@ -454,6 +393,7 @@ namespace MQTTnet.Server } _logger.Warning(null, "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); + Stop(MqttClientDisconnectType.NotClean, true); } @@ -462,12 +402,7 @@ namespace MQTTnet.Server var retainedMessages = _retainedMessagesManager.GetSubscribedMessages(topicFilters); foreach (var applicationMessage in retainedMessages) { - var publishPacket = applicationMessage.ToPublishPacket(); - - // Set the retain flag to true according to [MQTT-3.3.1-8]. - publishPacket.Retain = true; - - EnqueueApplicationMessage(null, publishPacket); + EnqueueApplicationMessage(null, applicationMessage, true); } } diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 0eb6c9f1..a453915b 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -7,7 +7,6 @@ using System.Threading.Tasks; using MQTTnet.Adapter; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; -using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -190,12 +189,7 @@ namespace MQTTnet.Server foreach (var clientSession in GetSessions()) { - var publishPacket = applicationMessage.ToPublishPacket(); - - // Set the retain flag to true according to [MQTT-3.3.1-9]. - publishPacket.Retain = false; - - clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, publishPacket); + clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, enqueuedApplicationMessage.ApplicationMessage, false); } } catch (OperationCanceledException)