Fix breaking changes.

This commit is contained in:
Christian
2022-07-31 11:10:38 +02:00
parent 5a21939e29
commit 29a9941d86
+31 -33
View File
@@ -20,19 +20,19 @@ namespace MQTTnet.Server
{ {
public class MqttServer : Disposable public class MqttServer : Disposable
{ {
readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer();
readonly IDictionary _sessionItems = new ConcurrentDictionary<object, object>();
readonly ICollection<IMqttServerAdapter> _adapters; readonly ICollection<IMqttServerAdapter> _adapters;
readonly MqttClientSessionsManager _clientSessionsManager;
readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer();
readonly MqttServerKeepAliveMonitor _keepAliveMonitor;
readonly MqttNetSourceLogger _logger; readonly MqttNetSourceLogger _logger;
readonly MqttServerOptions _options; readonly MqttServerOptions _options;
readonly IMqttNetLogger _rootLogger;
readonly MqttRetainedMessagesManager _retainedMessagesManager; readonly MqttRetainedMessagesManager _retainedMessagesManager;
readonly MqttServerKeepAliveMonitor _keepAliveMonitor; readonly IMqttNetLogger _rootLogger;
readonly MqttClientSessionsManager _clientSessionsManager;
readonly IDictionary _sessionItems = new ConcurrentDictionary<object, object>();
CancellationTokenSource _cancellationTokenSource; CancellationTokenSource _cancellationTokenSource;
public MqttServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) public MqttServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
{ {
_options = options ?? throw new ArgumentNullException(nameof(options)); _options = options ?? throw new ArgumentNullException(nameof(options));
@@ -58,18 +58,18 @@ namespace MQTTnet.Server
remove => _eventContainer.ApplicationMessageNotConsumedEvent.RemoveHandler(value); remove => _eventContainer.ApplicationMessageNotConsumedEvent.RemoveHandler(value);
} }
public event Func<ClientConnectedEventArgs, Task> ClientConnectedAsync
{
add => _eventContainer.ClientConnectedEvent.AddHandler(value);
remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value);
}
public event Func<ClientAcknowledgedPublishPacketEventArgs, Task> ClientAcknowledgedPublishPacketAsync public event Func<ClientAcknowledgedPublishPacketEventArgs, Task> ClientAcknowledgedPublishPacketAsync
{ {
add => _eventContainer.ClientAcknowledgedPublishPacketEvent.AddHandler(value); add => _eventContainer.ClientAcknowledgedPublishPacketEvent.AddHandler(value);
remove => _eventContainer.ClientAcknowledgedPublishPacketEvent.RemoveHandler(value); remove => _eventContainer.ClientAcknowledgedPublishPacketEvent.RemoveHandler(value);
} }
public event Func<ClientConnectedEventArgs, Task> ClientConnectedAsync
{
add => _eventContainer.ClientConnectedEvent.AddHandler(value);
remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value);
}
public event Func<ClientDisconnectedEventArgs, Task> ClientDisconnectedAsync public event Func<ClientDisconnectedEventArgs, Task> ClientDisconnectedAsync
{ {
add => _eventContainer.ClientDisconnectedEvent.AddHandler(value); add => _eventContainer.ClientDisconnectedEvent.AddHandler(value);
@@ -167,7 +167,7 @@ namespace MQTTnet.Server
} }
public bool IsStarted => _cancellationTokenSource != null; public bool IsStarted => _cancellationTokenSource != null;
public Task DeleteRetainedMessagesAsync() public Task DeleteRetainedMessagesAsync()
{ {
ThrowIfNotStarted(); ThrowIfNotStarted();
@@ -183,10 +183,10 @@ namespace MQTTnet.Server
} }
ThrowIfNotStarted(); ThrowIfNotStarted();
return _clientSessionsManager.GetClient(id).StopAsync(reasonCode); return _clientSessionsManager.GetClient(id).StopAsync(reasonCode);
} }
public Task<IList<MqttClientStatus>> GetClientsAsync() public Task<IList<MqttClientStatus>> GetClientsAsync()
{ {
ThrowIfNotStarted(); ThrowIfNotStarted();
@@ -226,17 +226,15 @@ namespace MQTTnet.Server
var processPublish = true; var processPublish = true;
var applicationMessage = injectedApplicationMessage.ApplicationMessage; var applicationMessage = injectedApplicationMessage.ApplicationMessage;
if (_eventContainer.InterceptingPublishEvent.HasHandlers) if (_eventContainer.InterceptingPublishEvent.HasHandlers)
{ {
var interceptingPublishEventArgs = new InterceptingPublishEventArgs var interceptingPublishEventArgs = new InterceptingPublishEventArgs(
{ applicationMessage,
ApplicationMessage = applicationMessage, _cancellationTokenSource.Token,
CancellationToken = _cancellationTokenSource.Token, injectedApplicationMessage.SenderClientId,
ClientId = injectedApplicationMessage.SenderClientId, _sessionItems);
SessionItems = _sessionItems
};
await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false); await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false);
applicationMessage = interceptingPublishEventArgs.ApplicationMessage; applicationMessage = interceptingPublishEventArgs.ApplicationMessage;
@@ -247,12 +245,12 @@ namespace MQTTnet.Server
{ {
return; return;
} }
if (string.IsNullOrEmpty(applicationMessage.Topic)) if (string.IsNullOrEmpty(applicationMessage.Topic))
{ {
throw new NotSupportedException("Injected application messages must contain a topic. Topic alias is not supported."); throw new NotSupportedException("Injected application messages must contain a topic. Topic alias is not supported.");
} }
await _clientSessionsManager.DispatchApplicationMessage(injectedApplicationMessage.SenderClientId, applicationMessage).ConfigureAwait(false); await _clientSessionsManager.DispatchApplicationMessage(injectedApplicationMessage.SenderClientId, applicationMessage).ConfigureAwait(false);
} }
@@ -274,7 +272,7 @@ namespace MQTTnet.Server
} }
await _eventContainer.StartedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); await _eventContainer.StartedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
_logger.Info("Started."); _logger.Info("Started.");
} }
@@ -304,7 +302,7 @@ namespace MQTTnet.Server
} }
await _eventContainer.StoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); await _eventContainer.StoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
_logger.Info("Stopped."); _logger.Info("Stopped.");
} }
@@ -348,7 +346,7 @@ namespace MQTTnet.Server
return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters); return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters);
} }
public Task UpdateRetainedMessageAsync(MqttApplicationMessage retainedMessage) public Task UpdateRetainedMessageAsync(MqttApplicationMessage retainedMessage)
{ {
if (retainedMessage == null) if (retainedMessage == null)
@@ -385,7 +383,7 @@ namespace MQTTnet.Server
void ThrowIfNotStarted() void ThrowIfNotStarted()
{ {
ThrowIfDisposed(); ThrowIfDisposed();
if (_cancellationTokenSource == null) if (_cancellationTokenSource == null)
{ {
throw new InvalidOperationException("The MQTT server is not started."); throw new InvalidOperationException("The MQTT server is not started.");
@@ -395,7 +393,7 @@ namespace MQTTnet.Server
void ThrowIfStarted() void ThrowIfStarted()
{ {
ThrowIfDisposed(); ThrowIfDisposed();
if (_cancellationTokenSource != null) if (_cancellationTokenSource != null)
{ {
throw new InvalidOperationException("The MQTT server is already started."); throw new InvalidOperationException("The MQTT server is already started.");