mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-15 17:00:03 +08:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
be0c75439c |
@@ -69,6 +69,11 @@ namespace MQTTnet.Extensions.Rpc
|
|||||||
throw new ArgumentNullException(nameof(methodName));
|
throw new ArgumentNullException(nameof(methodName));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (!_mqttClient.IsConnected)
|
||||||
|
{
|
||||||
|
throw new MqttClientDisconnectedException(null);
|
||||||
|
}
|
||||||
|
|
||||||
var context = new TopicGenerationContext(_mqttClient, _options, methodName, qualityOfServiceLevel);
|
var context = new TopicGenerationContext(_mqttClient, _options, methodName, qualityOfServiceLevel);
|
||||||
var topicNames = _options.TopicGenerationStrategy.CreateRpcTopics(context);
|
var topicNames = _options.TopicGenerationStrategy.CreateRpcTopics(context);
|
||||||
|
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ using System.Threading.Tasks;
|
|||||||
using Microsoft.VisualStudio.TestTools.UnitTesting;
|
using Microsoft.VisualStudio.TestTools.UnitTesting;
|
||||||
using MQTTnet.Client;
|
using MQTTnet.Client;
|
||||||
using MQTTnet.Exceptions;
|
using MQTTnet.Exceptions;
|
||||||
|
using MQTTnet.Extensions.Rpc;
|
||||||
using MQTTnet.Formatter;
|
using MQTTnet.Formatter;
|
||||||
using MQTTnet.Internal;
|
using MQTTnet.Internal;
|
||||||
using MQTTnet.Packets;
|
using MQTTnet.Packets;
|
||||||
@@ -20,6 +21,38 @@ namespace MQTTnet.Tests.Clients.MqttClient
|
|||||||
[TestClass]
|
[TestClass]
|
||||||
public sealed class MqttClient_Connection_Tests : BaseTestClass
|
public sealed class MqttClient_Connection_Tests : BaseTestClass
|
||||||
{
|
{
|
||||||
|
[TestMethod]
|
||||||
|
public async Task Connect_Disconnect_Connect()
|
||||||
|
{
|
||||||
|
using (var testEnvironment = CreateTestEnvironment())
|
||||||
|
{
|
||||||
|
await testEnvironment.StartServer();
|
||||||
|
|
||||||
|
var clientOptions = testEnvironment.CreateDefaultClientOptions();
|
||||||
|
var client = testEnvironment.CreateClient();
|
||||||
|
|
||||||
|
await client.ConnectAsync(clientOptions);
|
||||||
|
await client.DisconnectAsync();
|
||||||
|
await client.ConnectAsync(clientOptions);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
[TestMethod]
|
||||||
|
[ExpectedException(typeof(InvalidOperationException))]
|
||||||
|
public async Task Connect_Multiple_Times_Should_Fail()
|
||||||
|
{
|
||||||
|
using (var testEnvironment = CreateTestEnvironment())
|
||||||
|
{
|
||||||
|
await testEnvironment.StartServer();
|
||||||
|
|
||||||
|
var clientOptions = testEnvironment.CreateDefaultClientOptions();
|
||||||
|
var client = testEnvironment.CreateClient();
|
||||||
|
|
||||||
|
await client.ConnectAsync(clientOptions);
|
||||||
|
await client.ConnectAsync(clientOptions);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
[TestMethod]
|
[TestMethod]
|
||||||
[ExpectedException(typeof(MqttCommunicationException))]
|
[ExpectedException(typeof(MqttCommunicationException))]
|
||||||
public async Task Connect_To_Invalid_Server_Port_Not_Opened()
|
public async Task Connect_To_Invalid_Server_Port_Not_Opened()
|
||||||
@@ -196,5 +229,17 @@ namespace MQTTnet.Tests.Clients.MqttClient
|
|||||||
Assert.AreEqual(response.UserProperties[0].Value, "Value");
|
Assert.AreEqual(response.UserProperties[0].Value, "Value");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[TestMethod]
|
||||||
|
[ExpectedException(typeof(MqttClientDisconnectedException))]
|
||||||
|
public async Task Rpc_Client_Not_Connected()
|
||||||
|
{
|
||||||
|
var factory = new MqttFactory();
|
||||||
|
using (var client = factory.CreateMqttClient())
|
||||||
|
{
|
||||||
|
var rpcClient = factory.CreateMqttRpcClient(client);
|
||||||
|
await rpcClient.ExecuteAsync("aaa", new byte[] { 1, 2, 3 }, MqttQualityOfServiceLevel.AtLeastOnce);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -0,0 +1,59 @@
|
|||||||
|
// 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;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using Microsoft.VisualStudio.TestTools.UnitTesting;
|
||||||
|
using MQTTnet.Client;
|
||||||
|
using MQTTnet.Protocol;
|
||||||
|
using MQTTnet.Server;
|
||||||
|
|
||||||
|
namespace MQTTnet.Tests.Clients.MqttClient
|
||||||
|
{
|
||||||
|
[TestClass]
|
||||||
|
public sealed class MqttClient_Publish_Tests : BaseTestClass
|
||||||
|
{
|
||||||
|
[TestMethod]
|
||||||
|
public async Task Acknowledge_After_Disconnect_Should_Fail()
|
||||||
|
{
|
||||||
|
using (var testEnvironment = CreateTestEnvironment())
|
||||||
|
{
|
||||||
|
var server = await testEnvironment.StartServer();
|
||||||
|
|
||||||
|
var client = await testEnvironment.ConnectClient();
|
||||||
|
await client.SubscribeAsync("TEST", MqttQualityOfServiceLevel.AtLeastOnce);
|
||||||
|
|
||||||
|
var flag = false;
|
||||||
|
|
||||||
|
client.ApplicationMessageReceivedAsync += async args =>
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
args.AutoAcknowledge = false;
|
||||||
|
|
||||||
|
await client.DisconnectAsync();
|
||||||
|
await args.AcknowledgeAsync();
|
||||||
|
|
||||||
|
flag = true;
|
||||||
|
}
|
||||||
|
catch (Exception exception)
|
||||||
|
{
|
||||||
|
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
await server.InjectApplicationMessageAsync(
|
||||||
|
new MqttApplicationMessage
|
||||||
|
{
|
||||||
|
Topic = "TEST"
|
||||||
|
});
|
||||||
|
|
||||||
|
SpinWait.SpinUntil(() => flag, TimeSpan.FromSeconds(10));
|
||||||
|
|
||||||
|
Assert.IsTrue(flag);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -69,38 +69,8 @@ namespace MQTTnet.Tests.Clients.MqttClient
|
|||||||
Assert.IsTrue(success);
|
Assert.IsTrue(success);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
[TestMethod]
|
|
||||||
public async Task Connect_Disconnect_Connect()
|
|
||||||
{
|
|
||||||
using (var testEnvironment = CreateTestEnvironment())
|
|
||||||
{
|
|
||||||
await testEnvironment.StartServer();
|
|
||||||
|
|
||||||
var clientOptions = testEnvironment.CreateDefaultClientOptions();
|
|
||||||
var client = testEnvironment.CreateClient();
|
|
||||||
|
|
||||||
await client.ConnectAsync(clientOptions);
|
|
||||||
await client.DisconnectAsync();
|
|
||||||
await client.ConnectAsync(clientOptions);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
[TestMethod]
|
|
||||||
[ExpectedException(typeof(InvalidOperationException))]
|
|
||||||
public async Task Connect_Multiple_Times_Should_Fail()
|
|
||||||
{
|
|
||||||
using (var testEnvironment = CreateTestEnvironment())
|
|
||||||
{
|
|
||||||
await testEnvironment.StartServer();
|
|
||||||
|
|
||||||
var clientOptions = testEnvironment.CreateDefaultClientOptions();
|
|
||||||
var client = testEnvironment.CreateClient();
|
|
||||||
|
|
||||||
await client.ConnectAsync(clientOptions);
|
|
||||||
await client.ConnectAsync(clientOptions);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
[TestMethod]
|
[TestMethod]
|
||||||
public async Task Disconnect_Event_Contains_Exception()
|
public async Task Disconnect_Event_Contains_Exception()
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ using MQTTnet.Client;
|
|||||||
namespace MQTTnet.Tests.Clients
|
namespace MQTTnet.Tests.Clients
|
||||||
{
|
{
|
||||||
[TestClass]
|
[TestClass]
|
||||||
public class MqttClientOptionsBuilder_Tests
|
public sealed class MqttClientOptionsBuilder_Tests
|
||||||
{
|
{
|
||||||
[TestMethod]
|
[TestMethod]
|
||||||
public void WithConnectionUri_Credential_Test()
|
public void WithConnectionUri_Credential_Test()
|
||||||
|
|||||||
@@ -881,6 +881,7 @@ namespace MQTTnet.Client
|
|||||||
|
|
||||||
_lastPacketSentTimestamp = DateTime.UtcNow;
|
_lastPacketSentTimestamp = DateTime.UtcNow;
|
||||||
|
|
||||||
|
//ThrowIfNotConnected();
|
||||||
return _adapter.SendPacketAsync(packet, cancellationToken);
|
return _adapter.SendPacketAsync(packet, cancellationToken);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -77,7 +77,7 @@ namespace MQTTnet.Client
|
|||||||
|
|
||||||
internal MqttPublishPacket PublishPacket { get; set; }
|
internal MqttPublishPacket PublishPacket { get; set; }
|
||||||
|
|
||||||
public Task AcknowledgeAsync(CancellationToken cancellationToken)
|
public Task AcknowledgeAsync(CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
if (_acknowledgeHandler == null)
|
if (_acknowledgeHandler == null)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -229,7 +229,13 @@ namespace MQTTnet.Server
|
|||||||
return _clientSessionsManager.GetSessionsStatus();
|
return _clientSessionsManager.GetSessionsStatus();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Obsolete("Use InjectApplicationMessageAsync instead.")]
|
||||||
public Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage, CancellationToken cancellationToken = default)
|
public Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
return InjectApplicationMessageAsync(injectedApplicationMessage, cancellationToken);
|
||||||
|
}
|
||||||
|
|
||||||
|
public Task InjectApplicationMessageAsync(InjectedMqttApplicationMessage injectedApplicationMessage, CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
if (injectedApplicationMessage == null)
|
if (injectedApplicationMessage == null)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -13,12 +13,38 @@ namespace MQTTnet.Server
|
|||||||
{
|
{
|
||||||
public static class MqttServerExtensions
|
public static class MqttServerExtensions
|
||||||
{
|
{
|
||||||
|
[Obsolete("Use InjectApplicationMessageAsync instead.")]
|
||||||
public static Task InjectApplicationMessage(
|
public static Task InjectApplicationMessage(
|
||||||
this MqttServer server,
|
this MqttServer server,
|
||||||
string topic,
|
string topic,
|
||||||
string payload = null,
|
string payload = null,
|
||||||
MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
|
MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
|
||||||
bool retain = false)
|
bool retain = false)
|
||||||
|
{
|
||||||
|
return InjectApplicationMessageAsync(server, topic, payload, qualityOfServiceLevel, retain);
|
||||||
|
}
|
||||||
|
|
||||||
|
public static Task InjectApplicationMessageAsync(this MqttServer server, MqttApplicationMessage applicationMessage)
|
||||||
|
{
|
||||||
|
if (server == null)
|
||||||
|
{
|
||||||
|
throw new ArgumentNullException(nameof(server));
|
||||||
|
}
|
||||||
|
|
||||||
|
if (applicationMessage == null)
|
||||||
|
{
|
||||||
|
throw new ArgumentNullException(nameof(applicationMessage));
|
||||||
|
}
|
||||||
|
|
||||||
|
return server.InjectApplicationMessageAsync(new InjectedMqttApplicationMessage(applicationMessage));
|
||||||
|
}
|
||||||
|
|
||||||
|
public static Task InjectApplicationMessageAsync(
|
||||||
|
this MqttServer server,
|
||||||
|
string topic,
|
||||||
|
string payload = null,
|
||||||
|
MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
|
||||||
|
bool retain = false)
|
||||||
{
|
{
|
||||||
if (server == null)
|
if (server == null)
|
||||||
{
|
{
|
||||||
@@ -36,7 +62,7 @@ namespace MQTTnet.Server
|
|||||||
payloadBuffer = Encoding.UTF8.GetBytes(stringPayload);
|
payloadBuffer = Encoding.UTF8.GetBytes(stringPayload);
|
||||||
}
|
}
|
||||||
|
|
||||||
return server.InjectApplicationMessage(
|
return server.InjectApplicationMessageAsync(
|
||||||
new InjectedMqttApplicationMessage(
|
new InjectedMqttApplicationMessage(
|
||||||
new MqttApplicationMessage
|
new MqttApplicationMessage
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user