YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-24 14:43:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add extension methods.

Browse Source
This commit is contained in:
Christian Kratky
2019-04-07 19:58:01 +02:00
parent dbee941dcc
commit 61b10a8b56
11 changed files with 300 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/MqttClientConnection.cs
+1 -10
View File
@@ -121,15 +121,6 @@ namespace MQTTnet.Server
return _packageReceiverTask;
}
public void EnqueueApplicationMessage(MqttApplicationMessage applicationMessage, string senderClientId, bool isRetainedApplicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
Session.EnqueueApplicationMessage(applicationMessage, senderClientId, isRetainedApplicationMessage);
_logger.Verbose("Queued application message (ClientId: {0}).", ClientId);
}
private async Task<MqttClientDisconnectType> RunInternalAsync()
{
var disconnectType = MqttClientDisconnectType.NotClean;
@@ -273,7 +264,7 @@ namespace MQTTnet.Server
var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false);
foreach (var applicationMessage in retainedMessages)
{
EnqueueApplicationMessage(applicationMessage, ClientId, true);
Session.EnqueueApplicationMessage(applicationMessage, ClientId, true);
}
}
Source/MQTTnet/Server/MqttClientSession.cs
+10 -1
View File
@@ -1,20 +1,26 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Server.Status;
namespace MQTTnet.Server
{
public class MqttClientSession
{
private readonly IMqttNetChildLogger _logger;
private readonly DateTime _createdTimestamp = DateTime.UtcNow;
public MqttClientSession(string clientId, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions)
public MqttClientSession(string clientId, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttNetChildLogger logger)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
SubscriptionsManager = new MqttClientSubscriptionsManager(clientId, eventDispatcher, serverOptions);
ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions);
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttClientSession));
}
public string ClientId { get; }
@@ -35,12 +41,15 @@ namespace MQTTnet.Server
return;
}
_logger.Verbose("Queued application message with topic '{0}' (ClientId: {1}).", applicationMessage.Topic, ClientId);
ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage);
}
public async Task SubscribeAsync(ICollection<TopicFilter> topicFilters, MqttRetainedMessagesManager retainedMessagesManager)
{
await SubscriptionsManager.SubscribeAsync(topicFilters).ConfigureAwait(false);
var matchingRetainedMessages = await retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false);
foreach (var matchingRetainedMessage in matchingRetainedMessages)
{
Source/MQTTnet/Server/MqttClientSessionsManager.cs
+1 -3
View File
@@ -332,8 +332,6 @@ namespace MQTTnet.Server
{
if (connectPacket.CleanSession)
{
// TODO: Check if required.
//session.Dispose();
session = null;
_logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId);
@@ -346,7 +344,7 @@ namespace MQTTnet.Server
if (session == null)
{
session = new MqttClientSession(connectPacket.ClientId, _eventDispatcher, _options);
session = new MqttClientSession(connectPacket.ClientId, _eventDispatcher, _options, _logger);
_logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
}
Source/MQTTnet/Server/MqttServerExtensions.cs
+37 -1
View File
@@ -2,14 +2,46 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client;
using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving;
using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public static class MqttServerExtensions
{
public static IMqttServer UseApplicationMessageReceivedHandler(this IMqttServer server, Func<MqttApplicationMessageReceivedEventArgs, Task> handler)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (handler == null)
{
return server.UseApplicationMessageReceivedHandler((IMqttApplicationMessageReceivedHandler)null);
}
return server.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(handler));
}
public static IMqttServer UseApplicationMessageReceivedHandler(this IMqttServer server, Action<MqttApplicationMessageReceivedEventArgs> handler)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (handler == null)
{
return server.UseApplicationMessageReceivedHandler((IMqttApplicationMessageReceivedHandler)null);
}
return server.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(handler));
}
public static IMqttServer UseApplicationMessageReceivedHandler(this IMqttServer server, IMqttApplicationMessageReceivedHandler handler)
{
if (server == null) throw new ArgumentNullException(nameof(server));
server.ApplicationMessageReceivedHandler = handler;
return server;
}
public static Task SubscribeAsync(this IMqttServer server, string clientId, params TopicFilter[] topicFilters)
{
if (server == null) throw new ArgumentNullException(nameof(server));
@@ -120,12 +152,16 @@ namespace MQTTnet.Server
public static Task<MqttClientPublishResult> PublishAsync(this IMqttServer server, Func<MqttApplicationMessageBuilder, MqttApplicationMessageBuilder> builder, CancellationToken cancellationToken)
{
if (server == null) throw new ArgumentNullException(nameof(server));
var message = builder(new MqttApplicationMessageBuilder()).Build();
return server.PublishAsync(message, cancellationToken);
}
public static Task<MqttClientPublishResult> PublishAsync(this IMqttServer server, Func<MqttApplicationMessageBuilder, MqttApplicationMessageBuilder> builder)
{
if (server == null) throw new ArgumentNullException(nameof(server));
var message = builder(new MqttApplicationMessageBuilder()).Build();
return server.PublishAsync(message, CancellationToken.None);
}