Expose custom session items in session status.

This commit is contained in:
Christian Kratky
2020-04-04 12:54:33 +02:00
parent 6e0d3ec0fb
commit 67d56644b2
7 changed files with 54 additions and 21 deletions
+1
View File
@@ -25,6 +25,7 @@
* [Server] Added support for assigned client IDs (MQTTv5 only) (thanks to @bcrosnier).
* [Server] Added interceptor for unsubscriptions.
* [Server] Removed exceptions when user properties are set with MQTT protocol version 3.1
* [Server] Added custom session items to the client status.
* [MQTTnet.AspNetCore] improved compatibility with AspNetCore 3.1
* [MQTTnet.Server] Added interceptor for unsubscriptions.
</releaseNotes>
+3 -2
View File
@@ -8,9 +8,9 @@ namespace MQTTnet.Server
{
public class MqttClientSession
{
private readonly IMqttNetLogger _logger;
readonly IMqttNetLogger _logger;
private readonly DateTime _createdTimestamp = DateTime.UtcNow;
readonly DateTime _createdTimestamp = DateTime.UtcNow;
public MqttClientSession(string clientId, IDictionary<object, object> items, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttNetLogger logger)
{
@@ -73,6 +73,7 @@ namespace MQTTnet.Server
status.ClientId = ClientId;
status.CreatedTimestamp = _createdTimestamp;
status.PendingApplicationMessagesCount = ApplicationMessagesQueue.Count;
status.Items = Items;
}
}
}
@@ -1,6 +1,6 @@
using System;
using MQTTnet.Formatter;
using System;
using System.Threading.Tasks;
using MQTTnet.Formatter;
namespace MQTTnet.Server.Status
{
@@ -9,7 +9,7 @@ namespace MQTTnet.Server.Status
string ClientId { get; }
string Endpoint { get; }
MqttProtocolVersion ProtocolVersion { get; }
DateTime LastPacketReceivedTimestamp { get; }
@@ -29,7 +29,7 @@ namespace MQTTnet.Server.Status
long BytesSent { get; }
long BytesReceived { get; }
Task DisconnectAsync();
void ResetStatistics();
@@ -1,12 +1,15 @@
using System.Threading.Tasks;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace MQTTnet.Server.Status
{
public interface IMqttSessionStatus
{
string ClientId { get; set; }
string ClientId { get; }
long PendingApplicationMessagesCount { get; set; }
long PendingApplicationMessagesCount { get; }
IDictionary<object, object> Items { get; }
Task ClearPendingApplicationMessagesAsync();
@@ -1,16 +1,16 @@
using System;
using MQTTnet.Formatter;
using System;
using System.Threading.Tasks;
using MQTTnet.Formatter;
namespace MQTTnet.Server.Status
{
public class MqttClientStatus : IMqttClientStatus
{
private readonly MqttClientConnection _connection;
readonly MqttClientConnection _connection;
public MqttClientStatus(MqttClientConnection connection)
{
_connection = connection;
_connection = connection ?? throw new ArgumentNullException(nameof(connection));
}
public string ClientId { get; set; }
@@ -1,12 +1,13 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace MQTTnet.Server.Status
{
public class MqttSessionStatus : IMqttSessionStatus
{
private readonly MqttClientSession _session;
private readonly MqttClientSessionsManager _sessionsManager;
readonly MqttClientSession _session;
readonly MqttClientSessionsManager _sessionsManager;
public MqttSessionStatus(MqttClientSession session, MqttClientSessionsManager sessionsManager)
{
@@ -17,14 +18,16 @@ namespace MQTTnet.Server.Status
public string ClientId { get; set; }
public long PendingApplicationMessagesCount { get; set; }
public DateTime CreatedTimestamp { get; set; }
public IDictionary<object, object> Items { get; set; }
public Task DeleteAsync()
{
return _sessionsManager.DeleteSessionAsync(ClientId);
}
public Task ClearPendingApplicationMessagesAsync()
{
_session.ApplicationMessagesQueue.Clear();
+29 -4
View File
@@ -1,10 +1,11 @@
using System.Text;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Client.Subscribing;
using MQTTnet.Server;
using MQTTnet.Tests.Mockups;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace MQTTnet.Tests
{
@@ -42,7 +43,7 @@ namespace MQTTnet.Tests
string receivedPayload = null;
var client = await testEnvironment.ConnectClientAsync();
client.UseApplicationMessageReceivedHandler(delegate(MqttApplicationMessageReceivedEventArgs args)
client.UseApplicationMessageReceivedHandler(delegate (MqttApplicationMessageReceivedEventArgs args)
{
receivedPayload = args.ApplicationMessage.ConvertPayloadToString();
});
@@ -59,5 +60,29 @@ namespace MQTTnet.Tests
Assert.AreEqual("Hello World", receivedPayload);
}
}
[TestMethod]
public async Task Get_Session_Items_In_Status()
{
using (var testEnvironment = new TestEnvironment(TestContext))
{
var serverOptions = new MqttServerOptionsBuilder()
.WithConnectionValidator(delegate (MqttConnectionValidatorContext context)
{
// Don't validate anything. Just set some session items.
context.SessionItems["can_subscribe_x"] = true;
context.SessionItems["default_payload"] = "Hello World";
});
await testEnvironment.StartServerAsync(serverOptions);
var client = await testEnvironment.ConnectClientAsync();
var sessionStatus = await testEnvironment.Server.GetSessionStatusAsync();
var session = sessionStatus.First();
Assert.AreEqual(true, session.Items["can_subscribe_x"]);
}
}
}
}