From 29a9941d86e6d82d5505d11a25911cb8ce34df0e Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sun, 31 Jul 2022 11:10:38 +0200 Subject: [PATCH] Fix breaking changes. --- Source/MQTTnet/Server/MqttServer.cs | 64 ++++++++++++++--------------- 1 file changed, 31 insertions(+), 33 deletions(-) diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 054467a7..37115b99 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -20,19 +20,19 @@ namespace MQTTnet.Server { public class MqttServer : Disposable { - readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer(); - - readonly IDictionary _sessionItems = new ConcurrentDictionary(); readonly ICollection _adapters; + readonly MqttClientSessionsManager _clientSessionsManager; + readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer(); + readonly MqttServerKeepAliveMonitor _keepAliveMonitor; readonly MqttNetSourceLogger _logger; readonly MqttServerOptions _options; - readonly IMqttNetLogger _rootLogger; readonly MqttRetainedMessagesManager _retainedMessagesManager; - readonly MqttServerKeepAliveMonitor _keepAliveMonitor; - readonly MqttClientSessionsManager _clientSessionsManager; - + readonly IMqttNetLogger _rootLogger; + + readonly IDictionary _sessionItems = new ConcurrentDictionary(); + CancellationTokenSource _cancellationTokenSource; - + public MqttServer(MqttServerOptions options, IEnumerable adapters, IMqttNetLogger logger) { _options = options ?? throw new ArgumentNullException(nameof(options)); @@ -58,18 +58,18 @@ namespace MQTTnet.Server remove => _eventContainer.ApplicationMessageNotConsumedEvent.RemoveHandler(value); } - public event Func ClientConnectedAsync - { - add => _eventContainer.ClientConnectedEvent.AddHandler(value); - remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value); - } - public event Func ClientAcknowledgedPublishPacketAsync { add => _eventContainer.ClientAcknowledgedPublishPacketEvent.AddHandler(value); remove => _eventContainer.ClientAcknowledgedPublishPacketEvent.RemoveHandler(value); } + public event Func ClientConnectedAsync + { + add => _eventContainer.ClientConnectedEvent.AddHandler(value); + remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value); + } + public event Func ClientDisconnectedAsync { add => _eventContainer.ClientDisconnectedEvent.AddHandler(value); @@ -167,7 +167,7 @@ namespace MQTTnet.Server } public bool IsStarted => _cancellationTokenSource != null; - + public Task DeleteRetainedMessagesAsync() { ThrowIfNotStarted(); @@ -183,10 +183,10 @@ namespace MQTTnet.Server } ThrowIfNotStarted(); - + return _clientSessionsManager.GetClient(id).StopAsync(reasonCode); } - + public Task> GetClientsAsync() { ThrowIfNotStarted(); @@ -226,17 +226,15 @@ namespace MQTTnet.Server var processPublish = true; var applicationMessage = injectedApplicationMessage.ApplicationMessage; - + if (_eventContainer.InterceptingPublishEvent.HasHandlers) { - var interceptingPublishEventArgs = new InterceptingPublishEventArgs - { - ApplicationMessage = applicationMessage, - CancellationToken = _cancellationTokenSource.Token, - ClientId = injectedApplicationMessage.SenderClientId, - SessionItems = _sessionItems - }; - + var interceptingPublishEventArgs = new InterceptingPublishEventArgs( + applicationMessage, + _cancellationTokenSource.Token, + injectedApplicationMessage.SenderClientId, + _sessionItems); + await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false); applicationMessage = interceptingPublishEventArgs.ApplicationMessage; @@ -247,12 +245,12 @@ namespace MQTTnet.Server { return; } - + if (string.IsNullOrEmpty(applicationMessage.Topic)) { throw new NotSupportedException("Injected application messages must contain a topic. Topic alias is not supported."); } - + await _clientSessionsManager.DispatchApplicationMessage(injectedApplicationMessage.SenderClientId, applicationMessage).ConfigureAwait(false); } @@ -274,7 +272,7 @@ namespace MQTTnet.Server } await _eventContainer.StartedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); - + _logger.Info("Started."); } @@ -304,7 +302,7 @@ namespace MQTTnet.Server } await _eventContainer.StoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); - + _logger.Info("Stopped."); } @@ -348,7 +346,7 @@ namespace MQTTnet.Server return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters); } - + public Task UpdateRetainedMessageAsync(MqttApplicationMessage retainedMessage) { if (retainedMessage == null) @@ -385,7 +383,7 @@ namespace MQTTnet.Server void ThrowIfNotStarted() { ThrowIfDisposed(); - + if (_cancellationTokenSource == null) { throw new InvalidOperationException("The MQTT server is not started."); @@ -395,7 +393,7 @@ namespace MQTTnet.Server void ThrowIfStarted() { ThrowIfDisposed(); - + if (_cancellationTokenSource != null) { throw new InvalidOperationException("The MQTT server is already started.");