Migrate handlers to events in ManagedClient.

This commit is contained in:
Christian
2022-04-21 20:47:53 +02:00
parent 1d4794b071
commit 920728e77a
7 changed files with 23 additions and 114 deletions
@@ -6,7 +6,7 @@ using System;
namespace MQTTnet.Extensions.ManagedClient namespace MQTTnet.Extensions.ManagedClient
{ {
public class ApplicationMessageSkippedEventArgs : EventArgs public sealed class ApplicationMessageSkippedEventArgs : EventArgs
{ {
public ApplicationMessageSkippedEventArgs(ManagedMqttApplicationMessage applicationMessage) public ApplicationMessageSkippedEventArgs(ManagedMqttApplicationMessage applicationMessage)
{ {
@@ -1,35 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Threading.Tasks;
namespace MQTTnet.Extensions.ManagedClient
{
public class ApplicationMessageSkippedHandlerDelegate : IApplicationMessageSkippedHandler
{
private readonly Func<ApplicationMessageSkippedEventArgs, Task> _handler;
public ApplicationMessageSkippedHandlerDelegate(Action<ApplicationMessageSkippedEventArgs> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handler = eventArgs =>
{
handler(eventArgs);
return Task.FromResult(0);
};
}
public ApplicationMessageSkippedHandlerDelegate(Func<ApplicationMessageSkippedEventArgs, Task> handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
public Task HandleApplicationMessageSkippedAsync(ApplicationMessageSkippedEventArgs eventArgs)
{
return _handler(eventArgs);
}
}
}
@@ -1,13 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Threading.Tasks;
namespace MQTTnet.Extensions.ManagedClient
{
public interface IApplicationMessageSkippedHandler
{
Task HandleApplicationMessageSkippedAsync(ApplicationMessageSkippedEventArgs eventArgs);
}
}
@@ -13,6 +13,8 @@ namespace MQTTnet.Extensions.ManagedClient
event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync; event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync;
event Func<ApplicationMessageSkippedEventArgs, Task> ApplicationMessageSkippedAsync;
event Func<EventArgs, Task> ConnectedAsync; event Func<EventArgs, Task> ConnectedAsync;
event Func<ConnectingFailedEventArgs, Task> ConnectingFailedAsync; event Func<ConnectingFailedEventArgs, Task> ConnectingFailedAsync;
@@ -21,7 +23,7 @@ namespace MQTTnet.Extensions.ManagedClient
event Func<EventArgs, Task> DisconnectedAsync; event Func<EventArgs, Task> DisconnectedAsync;
IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; } event Func<ManagedProcessFailedEventArgs, Task> SynchronizingSubscriptionsFailedAsync;
IMqttClient InternalClient { get; } IMqttClient InternalClient { get; }
@@ -33,8 +35,6 @@ namespace MQTTnet.Extensions.ManagedClient
int PendingApplicationMessagesCount { get; } int PendingApplicationMessagesCount { get; }
ISynchronizingSubscriptionsFailedHandler SynchronizingSubscriptionsFailedHandler { get; set; }
Task EnqueueAsync(MqttApplicationMessage applicationMessage); Task EnqueueAsync(MqttApplicationMessage applicationMessage);
Task EnqueueAsync(ManagedMqttApplicationMessage applicationMessage); Task EnqueueAsync(ManagedMqttApplicationMessage applicationMessage);
@@ -1,13 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Threading.Tasks;
namespace MQTTnet.Extensions.ManagedClient
{
public interface ISynchronizingSubscriptionsFailedHandler
{
Task HandleSynchronizingSubscriptionsFailedAsync(ManagedProcessFailedEventArgs eventArgs);
}
}
@@ -20,9 +20,11 @@ namespace MQTTnet.Extensions.ManagedClient
public sealed class ManagedMqttClient : Disposable, IManagedMqttClient public sealed class ManagedMqttClient : Disposable, IManagedMqttClient
{ {
readonly AsyncEvent<ApplicationMessageProcessedEventArgs> _applicationMessageProcessedEvent = new AsyncEvent<ApplicationMessageProcessedEventArgs>(); readonly AsyncEvent<ApplicationMessageProcessedEventArgs> _applicationMessageProcessedEvent = new AsyncEvent<ApplicationMessageProcessedEventArgs>();
readonly AsyncEvent<ApplicationMessageSkippedEventArgs> _applicationMessageSkippedEvent = new AsyncEvent<ApplicationMessageSkippedEventArgs>();
readonly AsyncEvent<ConnectingFailedEventArgs> _connectingFailedEvent = new AsyncEvent<ConnectingFailedEventArgs>(); readonly AsyncEvent<ConnectingFailedEventArgs> _connectingFailedEvent = new AsyncEvent<ConnectingFailedEventArgs>();
readonly AsyncEvent<EventArgs> _connectionStateChangedEvent = new AsyncEvent<EventArgs>(); readonly AsyncEvent<EventArgs> _connectionStateChangedEvent = new AsyncEvent<EventArgs>();
readonly AsyncEvent<ManagedProcessFailedEventArgs> _synchronizingSubscriptionsFailedEvent = new AsyncEvent<ManagedProcessFailedEventArgs>();
readonly MqttNetSourceLogger _logger; readonly MqttNetSourceLogger _logger;
readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>(); readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
@@ -73,6 +75,12 @@ namespace MQTTnet.Extensions.ManagedClient
_logger = logger.WithSource(nameof(ManagedMqttClient)); _logger = logger.WithSource(nameof(ManagedMqttClient));
} }
public event Func<ApplicationMessageSkippedEventArgs, Task> ApplicationMessageSkippedAsync
{
add => _applicationMessageSkippedEvent.AddHandler(value);
remove => _applicationMessageSkippedEvent.RemoveHandler(value);
}
public event Func<ApplicationMessageProcessedEventArgs, Task> ApplicationMessageProcessedAsync public event Func<ApplicationMessageProcessedEventArgs, Task> ApplicationMessageProcessedAsync
{ {
add => _applicationMessageProcessedEvent.AddHandler(value); add => _applicationMessageProcessedEvent.AddHandler(value);
@@ -109,7 +117,11 @@ namespace MQTTnet.Extensions.ManagedClient
remove => InternalClient.DisconnectedAsync -= value; remove => InternalClient.DisconnectedAsync -= value;
} }
public IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; } public event Func<ManagedProcessFailedEventArgs, Task> SynchronizingSubscriptionsFailedAsync
{
add => _synchronizingSubscriptionsFailedEvent.AddHandler(value);
remove => _synchronizingSubscriptionsFailedEvent.RemoveHandler(value);
}
public IMqttClient InternalClient { get; } public IMqttClient InternalClient { get; }
@@ -120,9 +132,7 @@ namespace MQTTnet.Extensions.ManagedClient
public ManagedMqttClientOptions Options { get; private set; } public ManagedMqttClientOptions Options { get; private set; }
public int PendingApplicationMessagesCount => _messageQueue.Count; public int PendingApplicationMessagesCount => _messageQueue.Count;
public ISynchronizingSubscriptionsFailedHandler SynchronizingSubscriptionsFailedHandler { get; set; }
public async Task EnqueueAsync(MqttApplicationMessage applicationMessage) public async Task EnqueueAsync(MqttApplicationMessage applicationMessage)
{ {
ThrowIfDisposed(); ThrowIfDisposed();
@@ -191,13 +201,9 @@ namespace MQTTnet.Extensions.ManagedClient
} }
finally finally
{ {
if (applicationMessageSkippedEventArgs != null) if (_applicationMessageSkippedEvent.HasHandlers)
{ {
var applicationMessageSkippedHandler = ApplicationMessageSkippedHandler; await _applicationMessageSkippedEvent.InvokeAsync(applicationMessageSkippedEventArgs).ConfigureAwait(false);
if (applicationMessageSkippedHandler != null)
{
await applicationMessageSkippedHandler.HandleApplicationMessageSkippedAsync(applicationMessageSkippedEventArgs).ConfigureAwait(false);
}
} }
} }
} }
@@ -348,10 +354,9 @@ namespace MQTTnet.Extensions.ManagedClient
{ {
_logger.Warning(exception, "Synchronizing subscriptions failed."); _logger.Warning(exception, "Synchronizing subscriptions failed.");
var synchronizingSubscriptionsFailedHandler = SynchronizingSubscriptionsFailedHandler; if (_synchronizingSubscriptionsFailedEvent.HasHandlers)
if (SynchronizingSubscriptionsFailedHandler != null)
{ {
await synchronizingSubscriptionsFailedHandler.HandleSynchronizingSubscriptionsFailedAsync(new ManagedProcessFailedEventArgs(exception)).ConfigureAwait(false); await _synchronizingSubscriptionsFailedEvent.InvokeAsync(new ManagedProcessFailedEventArgs(exception)).ConfigureAwait(false);
} }
} }
@@ -1,35 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Threading.Tasks;
namespace MQTTnet.Extensions.ManagedClient
{
public class SynchronizingSubscriptionsFailedHandlerDelegate : ISynchronizingSubscriptionsFailedHandler
{
private readonly Func<ManagedProcessFailedEventArgs, Task> _handler;
public SynchronizingSubscriptionsFailedHandlerDelegate(Action<ManagedProcessFailedEventArgs> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handler = context =>
{
handler(context);
return Task.FromResult(0);
};
}
public SynchronizingSubscriptionsFailedHandlerDelegate(Func<ManagedProcessFailedEventArgs, Task> handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
public Task HandleSynchronizingSubscriptionsFailedAsync(ManagedProcessFailedEventArgs eventArgs)
{
return _handler(eventArgs);
}
}
}