diff --git a/MQTTnet.sln.DotSettings b/MQTTnet.sln.DotSettings
index 2dc673c8..49d8306b 100644
--- a/MQTTnet.sln.DotSettings
+++ b/MQTTnet.sln.DotSettings
@@ -243,6 +243,11 @@ See the LICENSE file in the project root for more information.
True
True
True
+ True
True
+ True
+ True
+ True
True
- True
\ No newline at end of file
+ True
+ True
\ No newline at end of file
diff --git a/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs b/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs
index a9b08a87..865a49ce 100644
--- a/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs
+++ b/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs
@@ -18,7 +18,7 @@ using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Packets;
using MQTTnet.Protocol;
-using MQTTnet.Tests.Extensions;
+using MQTTnet.Tests.Helpers;
namespace MQTTnet.AspNetCore.Tests
{
@@ -35,7 +35,7 @@ namespace MQTTnet.AspNetCore.Tests
connection.Transport = pipe;
var ctx = new MqttConnectionContext(serializer, connection);
- await pipe.Receive.Writer.WriteAsync(writer.AddMqttHeader(MqttControlPacketType.Connect, new byte[0]));
+ await pipe.Receive.Writer.WriteAsync(writer.AddMqttHeader(MqttControlPacketType.Connect, Array.Empty()));
await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(CancellationToken.None));
diff --git a/Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs b/Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs
index 1d50b904..4ac0d4f6 100644
--- a/Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs
+++ b/Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs
@@ -19,8 +19,8 @@ namespace MQTTnet.Benchmarks
[MemoryDiagnoser]
public class ReaderExtensionsBenchmark
{
- private MqttPacketFormatterAdapter mqttPacketFormatter;
- private MemoryStream stream;
+ MqttPacketFormatterAdapter mqttPacketFormatter;
+ MemoryStream stream;
[GlobalSetup]
public void GlobalSetup()
@@ -31,8 +31,7 @@ namespace MQTTnet.Benchmarks
.WithPayload(new byte[10 * 1024])
.Build();
- var packetFactories = new MqttPacketFactories();
- var packet = packetFactories.Publish.Create(mqttMessage);
+ var packet = MqttPacketFactories.Publish.Create(mqttMessage);
var buffer = mqttPacketFormatter.Encode(packet);
stream = new MemoryStream();
diff --git a/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj b/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj
index d1d76589..8100c0e3 100644
--- a/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj
+++ b/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj
@@ -34,6 +34,7 @@
true
true
1591
+ 7.3
diff --git a/Source/MQTTnet.Extensions.Rpc/MqttFactoryExtensions.cs b/Source/MQTTnet.Extensions.Rpc/MqttFactoryExtensions.cs
index 3d4f5170..a8818461 100644
--- a/Source/MQTTnet.Extensions.Rpc/MqttFactoryExtensions.cs
+++ b/Source/MQTTnet.Extensions.Rpc/MqttFactoryExtensions.cs
@@ -11,10 +11,12 @@ namespace MQTTnet.Extensions.Rpc
{
public static IMqttRpcClient CreateMqttRpcClient(this MqttFactory factory, IMqttClient mqttClient)
{
- return factory.CreateMqttRpcClient(mqttClient, new MqttRpcClientOptions
- {
- TopicGenerationStrategy = new DefaultMqttRpcClientTopicGenerationStrategy()
- });
+ return factory.CreateMqttRpcClient(
+ mqttClient,
+ new MqttRpcClientOptions
+ {
+ TopicGenerationStrategy = new DefaultMqttRpcClientTopicGenerationStrategy()
+ });
}
public static IMqttRpcClient CreateMqttRpcClient(this MqttFactory _, IMqttClient mqttClient, MqttRpcClientOptions rpcClientOptions)
diff --git a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs
index e3ac567a..5784b814 100644
--- a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs
+++ b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs
@@ -17,8 +17,9 @@ namespace MQTTnet.Extensions.Rpc
{
readonly IMqttClient _mqttClient;
readonly MqttRpcClientOptions _options;
+
readonly ConcurrentDictionary> _waitingCalls = new ConcurrentDictionary>();
-
+
public MqttRpcClient(IMqttClient mqttClient, MqttRpcClientOptions options)
{
_mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
@@ -66,14 +67,8 @@ namespace MQTTnet.Extensions.Rpc
throw new ArgumentNullException(nameof(methodName));
}
- var topicNames = _options.TopicGenerationStrategy.CreateRpcTopics(
- new TopicGenerationContext
- {
- MethodName = methodName,
- QualityOfServiceLevel = qualityOfServiceLevel,
- MqttClient = _mqttClient,
- Options = _options
- });
+ var context = new TopicGenerationContext(_mqttClient, _options, methodName, qualityOfServiceLevel);
+ var topicNames = _options.TopicGenerationStrategy.CreateRpcTopics(context);
var requestTopic = topicNames.RequestTopic;
var responseTopic = topicNames.ResponseTopic;
diff --git a/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptionsBuilder.cs b/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptionsBuilder.cs
index dc53ddc0..5fb0ea2e 100644
--- a/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptionsBuilder.cs
+++ b/Source/MQTTnet.Extensions.Rpc/Options/MqttRpcClientOptionsBuilder.cs
@@ -10,13 +10,6 @@ namespace MQTTnet.Extensions.Rpc
{
IMqttRpcClientTopicGenerationStrategy _topicGenerationStrategy = new DefaultMqttRpcClientTopicGenerationStrategy();
- public MqttRpcClientOptionsBuilder WithTopicGenerationStrategy(IMqttRpcClientTopicGenerationStrategy value)
- {
- _topicGenerationStrategy = value ?? throw new ArgumentNullException(nameof(value));
-
- return this;
- }
-
public MqttRpcClientOptions Build()
{
return new MqttRpcClientOptions
@@ -24,5 +17,12 @@ namespace MQTTnet.Extensions.Rpc
TopicGenerationStrategy = _topicGenerationStrategy
};
}
+
+ public MqttRpcClientOptionsBuilder WithTopicGenerationStrategy(IMqttRpcClientTopicGenerationStrategy value)
+ {
+ _topicGenerationStrategy = value ?? throw new ArgumentNullException(nameof(value));
+
+ return this;
+ }
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/DefaultMqttRpcClientTopicGenerationStrategy.cs b/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/DefaultMqttRpcClientTopicGenerationStrategy.cs
index 11c087d8..e1af7473 100644
--- a/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/DefaultMqttRpcClientTopicGenerationStrategy.cs
+++ b/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/DefaultMqttRpcClientTopicGenerationStrategy.cs
@@ -10,6 +10,11 @@ namespace MQTTnet.Extensions.Rpc
{
public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
{
+ if (context == null)
+ {
+ throw new ArgumentNullException(nameof(context));
+ }
+
if (context.MethodName.Contains("/") || context.MethodName.Contains("+") || context.MethodName.Contains("#"))
{
throw new ArgumentException("The method name cannot contain /, + or #.");
diff --git a/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs b/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs
index 7185c1ba..03898bb8 100644
--- a/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs
+++ b/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs
@@ -2,6 +2,7 @@
// 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 MQTTnet.Client;
using MQTTnet.Protocol;
@@ -9,20 +10,29 @@ namespace MQTTnet.Extensions.Rpc
{
public sealed class TopicGenerationContext
{
- public string MethodName { get; set; }
+ public TopicGenerationContext(IMqttClient mqttClient, MqttRpcClientOptions options, string methodName, MqttQualityOfServiceLevel qualityOfServiceLevel)
+ {
+ MethodName = methodName ?? throw new ArgumentNullException(nameof(methodName));
+ QualityOfServiceLevel = qualityOfServiceLevel;
+ MqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
+ Options = options ?? throw new ArgumentNullException(nameof(options));
+ }
+
+ public string MethodName { get; }
+
+ public IMqttClient MqttClient { get; }
+
+ public MqttRpcClientOptions Options { get; }
///
- /// Gets or sets the quality of service level.
- /// The Quality of Service (QoS) level is an agreement between the sender of a message and the receiver of a message that defines the guarantee of delivery for a specific message.
- /// There are 3 QoS levels in MQTT:
- /// - At most once (0): Message gets delivered no time, once or multiple times.
- /// - At least once (1): Message gets delivered at least once (one time or more often).
- /// - Exactly once (2): Message gets delivered exactly once (It's ensured that the message only comes once).
+ /// Gets or sets the quality of service level.
+ /// The Quality of Service (QoS) level is an agreement between the sender of a message and the receiver of a message
+ /// that defines the guarantee of delivery for a specific message.
+ /// There are 3 QoS levels in MQTT:
+ /// - At most once (0): Message gets delivered no time, once or multiple times.
+ /// - At least once (1): Message gets delivered at least once (one time or more often).
+ /// - Exactly once (2): Message gets delivered exactly once (It's ensured that the message only comes once).
///
- public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
-
- public IMqttClient MqttClient { get; set; }
-
- public MqttRpcClientOptions Options { get; set; }
+ public MqttQualityOfServiceLevel QualityOfServiceLevel { get; }
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Extensions.WebSocket4Net/MQTTnet.Extensions.WebSocket4Net.csproj b/Source/MQTTnet.Extensions.WebSocket4Net/MQTTnet.Extensions.WebSocket4Net.csproj
index 04581cd0..ba3433ee 100644
--- a/Source/MQTTnet.Extensions.WebSocket4Net/MQTTnet.Extensions.WebSocket4Net.csproj
+++ b/Source/MQTTnet.Extensions.WebSocket4Net/MQTTnet.Extensions.WebSocket4Net.csproj
@@ -34,6 +34,7 @@
true
true
1591
+ 7.3
diff --git a/Source/MQTTnet.Extensions.WebSocket4Net/MqttFactoryExtensions.cs b/Source/MQTTnet.Extensions.WebSocket4Net/MqttFactoryExtensions.cs
index d23286f2..1b056f1b 100644
--- a/Source/MQTTnet.Extensions.WebSocket4Net/MqttFactoryExtensions.cs
+++ b/Source/MQTTnet.Extensions.WebSocket4Net/MqttFactoryExtensions.cs
@@ -10,9 +10,12 @@ namespace MQTTnet.Extensions.WebSocket4Net
{
public static MqttFactory UseWebSocket4Net(this MqttFactory mqttFactory)
{
- if (mqttFactory == null) throw new ArgumentNullException(nameof(mqttFactory));
+ if (mqttFactory == null)
+ {
+ throw new ArgumentNullException(nameof(mqttFactory));
+ }
return mqttFactory.UseClientAdapterFactory(new WebSocket4NetMqttClientAdapterFactory());
}
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs
index d656a92f..859bd2a3 100644
--- a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs
+++ b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs
@@ -5,6 +5,7 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
+using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Security.Authentication;
@@ -17,14 +18,14 @@ using MQTTnet.Exceptions;
using MQTTnet.Internal;
using SuperSocket.ClientEngine;
using WebSocket4Net;
+using DataReceivedEventArgs = WebSocket4Net.DataReceivedEventArgs;
namespace MQTTnet.Extensions.WebSocket4Net
{
public sealed class WebSocket4NetMqttChannel : IMqttChannel
{
- readonly BlockingCollection _receiveBuffer = new BlockingCollection();
-
readonly MqttClientOptions _clientOptions;
+ readonly BlockingCollection _receiveBuffer = new BlockingCollection();
readonly MqttClientWebSocketOptions _webSocketOptions;
WebSocket _webSocket;
@@ -35,13 +36,13 @@ namespace MQTTnet.Extensions.WebSocket4Net
_webSocketOptions = webSocketOptions ?? throw new ArgumentNullException(nameof(webSocketOptions));
}
+ public X509Certificate2 ClientCertificate { get; }
+
public string Endpoint => _webSocketOptions.Uri;
public bool IsSecureConnection { get; private set; }
- public X509Certificate2 ClientCertificate { get; }
-
- public async Task ConnectAsync(CancellationToken cancellationToken)
+ public Task ConnectAsync(CancellationToken cancellationToken)
{
var uri = _webSocketOptions.Uri;
if (!uri.StartsWith("ws://", StringComparison.OrdinalIgnoreCase) && !uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase))
@@ -55,11 +56,13 @@ namespace MQTTnet.Extensions.WebSocket4Net
uri = "wss://" + uri;
}
}
+
+ IsSecureConnection = uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase);
-#if NET48 || NETCOREAPP3_1 || NET5 || NET6
- var sslProtocols = _webSocketOptions?.TlsOptions?.SslProtocol ?? SslProtocols.Tls12 | SslProtocols.Tls13;
+#if NET48 || NETCOREAPP3_0_OR_GREATER
+ var sslProtocols = _webSocketOptions.TlsOptions?.SslProtocol ?? SslProtocols.Tls12 | SslProtocols.Tls13;
#else
- var sslProtocols = _webSocketOptions?.TlsOptions?.SslProtocol ?? SslProtocols.Tls12 | (SslProtocols)0x00003000 /*Tls13*/;
+ var sslProtocols = _webSocketOptions.TlsOptions?.SslProtocol ?? SslProtocols.Tls12 | (SslProtocols)0x00003000 /*Tls13*/;
#endif
var subProtocol = _webSocketOptions.SubProtocols.FirstOrDefault() ?? string.Empty;
@@ -90,7 +93,7 @@ namespace MQTTnet.Extensions.WebSocket4Net
var receiveBufferSize = 0;
var certificates = new X509CertificateCollection();
- if (_webSocketOptions?.TlsOptions?.Certificates != null)
+ if (_webSocketOptions.TlsOptions?.Certificates != null)
{
foreach (var certificate in _webSocketOptions.TlsOptions.Certificates)
{
@@ -107,15 +110,13 @@ namespace MQTTnet.Extensions.WebSocket4Net
NoDelay = true,
Security =
{
- AllowUnstrustedCertificate = _webSocketOptions?.TlsOptions?.AllowUntrustedCertificates == true,
- AllowCertificateChainErrors = _webSocketOptions?.TlsOptions?.IgnoreCertificateChainErrors == true,
+ AllowUnstrustedCertificate = _webSocketOptions.TlsOptions?.AllowUntrustedCertificates == true,
+ AllowCertificateChainErrors = _webSocketOptions.TlsOptions?.IgnoreCertificateChainErrors == true,
Certificates = certificates
}
};
- await ConnectInternalAsync(cancellationToken).ConfigureAwait(false);
-
- IsSecureConnection = uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase);
+ return ConnectInternalAsync(cancellationToken);
}
public Task DisconnectAsync(CancellationToken cancellationToken)
@@ -124,10 +125,23 @@ namespace MQTTnet.Extensions.WebSocket4Net
{
_webSocket.Close();
}
-
+
return CompletedTask.Instance;
}
+ public void Dispose()
+ {
+ if (_webSocket == null)
+ {
+ return;
+ }
+
+ _webSocket.DataReceived -= OnDataReceived;
+ _webSocket.Error -= OnError;
+ _webSocket.Dispose();
+ _webSocket = null;
+ }
+
public Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
var readBytes = 0;
@@ -161,75 +175,55 @@ namespace MQTTnet.Extensions.WebSocket4Net
return CompletedTask.Instance;
}
- public void Dispose()
- {
- if (_webSocket == null)
- {
- return;
- }
-
- _webSocket.DataReceived -= OnDataReceived;
- _webSocket.Error -= OnError;
- _webSocket.Dispose();
- _webSocket = null;
- }
-
- static void OnError(object sender, ErrorEventArgs e)
- {
- System.Diagnostics.Debug.Write(e.Exception.ToString());
- }
-
- void OnDataReceived(object sender, DataReceivedEventArgs e)
- {
- foreach (var @byte in e.Data)
- {
- _receiveBuffer.Add(@byte);
- }
- }
-
async Task ConnectInternalAsync(CancellationToken cancellationToken)
{
_webSocket.Error += OnError;
_webSocket.DataReceived += OnDataReceived;
- var taskCompletionSource = new TaskCompletionSource();
+ var promise = new AsyncTaskCompletionSource();
void ErrorHandler(object sender, ErrorEventArgs e)
{
- taskCompletionSource.TrySetResult(e.Exception);
+ promise.TrySetException(e.Exception);
}
- void SuccessHandler(object sender, EventArgs e)
+ void OpenedHandler(object sender, EventArgs e)
{
- taskCompletionSource.TrySetResult(null);
+ promise.TrySetResult(true);
+ }
+
+ void ClosedHandler(object sender, EventArgs e)
+ {
+ promise.TrySetException(new MqttCommunicationException("Connection closed."));
}
try
{
- _webSocket.Opened += SuccessHandler;
+ _webSocket.Opened += OpenedHandler;
+ _webSocket.Closed += ClosedHandler;
_webSocket.Error += ErrorHandler;
-#pragma warning disable AsyncFixer02 // Long-running or blocking operations inside an async method
- _webSocket.Open();
-#pragma warning restore AsyncFixer02 // Long-running or blocking operations inside an async method
-
- using (var timeoutCts = new CancellationTokenSource(_clientOptions.Timeout))
- using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(timeoutCts.Token, cancellationToken))
+ using (var timeout = new CancellationTokenSource(_clientOptions.Timeout))
+ using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, cancellationToken))
{
- using (linkedCts.Token.Register(() => taskCompletionSource.TrySetCanceled()))
+ using (linkedCts.Token.Register(() => promise.TrySetCanceled()))
{
try
{
- await taskCompletionSource.Task.ConfigureAwait(false);
+ // OpenAsync does basically the same but without support for a cancellation token.
+ // So we do it on our own!
+ _webSocket.Open();
+
+ await promise.Task.ConfigureAwait(false);
}
catch (Exception exception)
{
- var timeoutReached = timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested;
+ var timeoutReached = timeout.IsCancellationRequested && !cancellationToken.IsCancellationRequested;
if (timeoutReached)
{
throw new MqttCommunicationTimedOutException(exception);
}
-
+
if (exception is AuthenticationException authenticationException)
{
throw new MqttCommunicationException(authenticationException.InnerException);
@@ -251,9 +245,23 @@ namespace MQTTnet.Extensions.WebSocket4Net
}
finally
{
- _webSocket.Opened -= SuccessHandler;
+ _webSocket.Opened -= OpenedHandler;
+ _webSocket.Closed -= ClosedHandler;
_webSocket.Error -= ErrorHandler;
}
}
+
+ void OnDataReceived(object sender, DataReceivedEventArgs e)
+ {
+ foreach (var @byte in e.Data)
+ {
+ _receiveBuffer.Add(@byte);
+ }
+ }
+
+ static void OnError(object sender, ErrorEventArgs e)
+ {
+ Debug.Write(e.Exception.ToString());
+ }
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs
index 69ece623..27359a1a 100644
--- a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs
+++ b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs
@@ -2,12 +2,12 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
-using MQTTnet.Adapter;
-using MQTTnet.Formatter;
-using MQTTnet.Implementations;
using System;
+using MQTTnet.Adapter;
using MQTTnet.Client;
using MQTTnet.Diagnostics;
+using MQTTnet.Formatter;
+using MQTTnet.Implementations;
namespace MQTTnet.Extensions.WebSocket4Net
{
@@ -15,7 +15,10 @@ namespace MQTTnet.Extensions.WebSocket4Net
{
public IMqttChannelAdapter CreateClientAdapter(MqttClientOptions options, MqttPacketInspector packetInspector, IMqttNetLogger logger)
{
- if (options == null) throw new ArgumentNullException(nameof(options));
+ if (options == null)
+ {
+ throw new ArgumentNullException(nameof(options));
+ }
switch (options.ChannelOptions)
{
@@ -23,7 +26,7 @@ namespace MQTTnet.Extensions.WebSocket4Net
{
return new MqttChannelAdapter(
new MqttTcpChannel(options),
- new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(4096, 65535)),
+ new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(options.WriterBufferSize, options.WriterBufferSizeMax)),
packetInspector,
logger);
}
@@ -32,16 +35,16 @@ namespace MQTTnet.Extensions.WebSocket4Net
{
return new MqttChannelAdapter(
new WebSocket4NetMqttChannel(options, webSocketOptions),
- new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(4068, 65535)),
+ new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(options.WriterBufferSize, options.WriterBufferSizeMax)),
packetInspector,
logger);
}
default:
- {
- throw new NotSupportedException();
- }
+ {
+ throw new NotSupportedException();
+ }
}
}
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.TestApp/PublicBrokerTest.cs b/Source/MQTTnet.TestApp/PublicBrokerTest.cs
index 768ba874..f95c332e 100644
--- a/Source/MQTTnet.TestApp/PublicBrokerTest.cs
+++ b/Source/MQTTnet.TestApp/PublicBrokerTest.cs
@@ -2,11 +2,12 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
-using MQTTnet.Client;
using System;
using System.Security.Authentication;
using System.Threading;
using System.Threading.Tasks;
+using MQTTnet.Client;
+using MQTTnet.Extensions.WebSocket4Net;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Protocol;
@@ -17,10 +18,6 @@ namespace MQTTnet.TestApp
{
public static async Task RunAsync()
{
- // MqttNetConsoleLogger.ForwardToConsole();
-
- // For most of these connections to work, set output target to Net5.0.
-
#if NET5_0_OR_GREATER
// TLS13 is only available in Net5.0
var unsafeTls13 = new MqttClientOptionsBuilderTlsParameters
@@ -31,7 +28,6 @@ namespace MQTTnet.TestApp
CertificateValidationHandler = w => true
};
#endif
-
// Also defining TLS12 for servers that don't seem no to support TLS13.
var unsafeTls12 = new MqttClientOptionsBuilderTlsParameters
{
@@ -42,33 +38,37 @@ namespace MQTTnet.TestApp
};
// mqtt.eclipseprojects.io
- await ExecuteTestAsync("mqtt.eclipseprojects.io TCP",
- new MqttClientOptionsBuilder().WithTcpServer("mqtt.eclipseprojects.io", 1883)
- .WithProtocolVersion(MqttProtocolVersion.V311).Build());
+ await ExecuteTestAsync(
+ "mqtt.eclipseprojects.io TCP",
+ new MqttClientOptionsBuilder().WithTcpServer("mqtt.eclipseprojects.io", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
- await ExecuteTestAsync("mqtt.eclipseprojects.io WS",
- new MqttClientOptionsBuilder().WithWebSocketServer("mqtt.eclipseprojects.io:80/mqtt")
- .WithProtocolVersion(MqttProtocolVersion.V311).Build());
+ await ExecuteTestAsync(
+ "mqtt.eclipseprojects.io WS",
+ new MqttClientOptionsBuilder().WithWebSocketServer("mqtt.eclipseprojects.io:80/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build());
#if NET5_0_OR_GREATER
await ExecuteTestAsync("mqtt.eclipseprojects.io WS TLS13",
new MqttClientOptionsBuilder().WithWebSocketServer("mqtt.eclipseprojects.io:443/mqtt")
.WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls13).Build());
+
+ await ExecuteTestAsync("mqtt.eclipseprojects.io WS TLS13 (WebSocket4Net)",
+ new MqttClientOptionsBuilder().WithWebSocketServer("mqtt.eclipseprojects.io:443/mqtt")
+ .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls13).Build(),
+ true);
#endif
// test.mosquitto.org
- await ExecuteTestAsync("test.mosquitto.org TCP",
- new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1883)
- .WithProtocolVersion(MqttProtocolVersion.V311).Build());
+ await ExecuteTestAsync(
+ "test.mosquitto.org TCP",
+ new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
- await ExecuteTestAsync("test.mosquitto.org TCP - Authenticated",
- new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1884)
- .WithCredentials("rw", "readwrite")
- .WithProtocolVersion(MqttProtocolVersion.V311).Build());
+ await ExecuteTestAsync(
+ "test.mosquitto.org TCP - Authenticated",
+ new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1884).WithCredentials("rw", "readwrite").WithProtocolVersion(MqttProtocolVersion.V311).Build());
- await ExecuteTestAsync("test.mosquitto.org TCP TLS12",
- new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883)
- .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
+ await ExecuteTestAsync(
+ "test.mosquitto.org TCP TLS12",
+ new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883).WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
#if NET5_0_OR_GREATER
await ExecuteTestAsync("test.mosquitto.org TCP TLS13",
@@ -76,27 +76,40 @@ namespace MQTTnet.TestApp
.WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls13).Build());
#endif
- await ExecuteTestAsync("test.mosquitto.org TCP TLS12 - Authenticated",
+ await ExecuteTestAsync(
+ "test.mosquitto.org TCP TLS12 - Authenticated",
new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8885)
.WithCredentials("rw", "readwrite")
- .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
+ .WithProtocolVersion(MqttProtocolVersion.V311)
+ .WithTls(unsafeTls12)
+ .Build());
- await ExecuteTestAsync("test.mosquitto.org WS",
- new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8080/mqtt")
- .WithProtocolVersion(MqttProtocolVersion.V311).Build());
+ await ExecuteTestAsync(
+ "test.mosquitto.org WS",
+ new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8080/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build());
- await ExecuteTestAsync("test.mosquitto.org WS TLS12",
- new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8081/mqtt")
- .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
+ await ExecuteTestAsync(
+ "test.mosquitto.org WS (WebSocket4Net)",
+ new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8080/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build(),
+ true);
+
+ await ExecuteTestAsync(
+ "test.mosquitto.org WS TLS12",
+ new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8081/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
+
+ // await ExecuteTestAsync(
+ // "test.mosquitto.org WS TLS12 (WebSocket4Net)",
+ // new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8081/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build(),
+ // true);
// broker.emqx.io
- await ExecuteTestAsync("broker.emqx.io TCP",
- new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 1883)
- .WithProtocolVersion(MqttProtocolVersion.V311).Build());
+ await ExecuteTestAsync(
+ "broker.emqx.io TCP",
+ new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
- await ExecuteTestAsync("broker.emqx.io TCP TLS12",
- new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883)
- .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
+ await ExecuteTestAsync(
+ "broker.emqx.io TCP TLS12",
+ new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883).WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
#if NET5_0_OR_GREATER
await ExecuteTestAsync("broker.emqx.io TCP TLS13",
@@ -104,23 +117,37 @@ namespace MQTTnet.TestApp
.WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls13).Build());
#endif
- await ExecuteTestAsync("broker.emqx.io WS",
- new MqttClientOptionsBuilder().WithWebSocketServer("broker.emqx.io:8083/mqtt")
- .WithProtocolVersion(MqttProtocolVersion.V311).Build());
+ await ExecuteTestAsync(
+ "broker.emqx.io WS",
+ new MqttClientOptionsBuilder().WithWebSocketServer("broker.emqx.io:8083/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build());
- await ExecuteTestAsync("broker.emqx.io WS TLS12",
- new MqttClientOptionsBuilder().WithWebSocketServer("broker.emqx.io:8084/mqtt")
- .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
+ await ExecuteTestAsync(
+ "broker.emqx.io WS (WebSocket4Net)",
+ new MqttClientOptionsBuilder().WithWebSocketServer("broker.emqx.io:8084/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build(),
+ true);
+ await ExecuteTestAsync(
+ "broker.emqx.io WS TLS12",
+ new MqttClientOptionsBuilder().WithWebSocketServer("broker.emqx.io:8084/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
+
+ await ExecuteTestAsync(
+ "broker.emqx.io WS TLS12 (WebSocket4Net)",
+ new MqttClientOptionsBuilder().WithWebSocketServer("broker.emqx.io:8084/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build(),
+ true);
// broker.hivemq.com
- await ExecuteTestAsync("broker.hivemq.com TCP",
- new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com", 1883)
- .WithProtocolVersion(MqttProtocolVersion.V311).Build());
+ await ExecuteTestAsync(
+ "broker.hivemq.com TCP",
+ new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
- await ExecuteTestAsync("broker.hivemq.com WS",
- new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt")
- .WithProtocolVersion(MqttProtocolVersion.V311).Build());
+ await ExecuteTestAsync(
+ "broker.hivemq.com WS",
+ new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build());
+
+ await ExecuteTestAsync(
+ "broker.hivemq.com WS (WebSocket4Net)",
+ new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build(),
+ true);
// mqtt.swifitch.cz: Does not seem to operate any more
@@ -130,36 +157,43 @@ namespace MQTTnet.TestApp
Console.ReadLine();
}
- static async Task ExecuteTestAsync(string name, MqttClientOptions options)
+ static async Task ExecuteTestAsync(string name, MqttClientOptions options, bool useWebSocket4Net = false)
{
try
{
Write("Testing '" + name + "'... ", ConsoleColor.Gray);
+
var factory = new MqttFactory();
- //factory.UseWebSocket4Net();
- var client = factory.CreateMqttClient();
- var topic = Guid.NewGuid().ToString();
-
- MqttApplicationMessage receivedMessage = null;
- client.ApplicationMessageReceivedAsync += e =>
+ if (useWebSocket4Net)
{
- receivedMessage = e.ApplicationMessage;
- return CompletedTask.Instance;
- };
-
- await client.ConnectAsync(options);
- await client.SubscribeAsync(topic, MqttQualityOfServiceLevel.AtLeastOnce);
- await client.PublishStringAsync(topic, "Hello_World", MqttQualityOfServiceLevel.AtLeastOnce);
-
- SpinWait.SpinUntil(() => receivedMessage != null, 5000);
-
- if (receivedMessage?.Topic != topic || receivedMessage?.ConvertPayloadToString() != "Hello_World")
- {
- throw new Exception("Message invalid.");
+ factory.UseWebSocket4Net();
}
- await client.UnsubscribeAsync(topic);
- await client.DisconnectAsync();
+ using (var client = factory.CreateMqttClient())
+ {
+ MqttApplicationMessage receivedMessage = null;
+ client.ApplicationMessageReceivedAsync += e =>
+ {
+ receivedMessage = e.ApplicationMessage;
+ return CompletedTask.Instance;
+ };
+
+ await client.ConnectAsync(options).ConfigureAwait(false);
+
+ var topic = Guid.NewGuid().ToString();
+ await client.SubscribeAsync(topic, MqttQualityOfServiceLevel.AtLeastOnce).ConfigureAwait(false);
+ await client.PublishStringAsync(topic, "Hello_World", MqttQualityOfServiceLevel.AtLeastOnce).ConfigureAwait(false);
+
+ SpinWait.SpinUntil(() => receivedMessage != null, 5000);
+
+ if (receivedMessage?.Topic != topic || receivedMessage?.ConvertPayloadToString() != "Hello_World")
+ {
+ throw new Exception("Message invalid.");
+ }
+
+ await client.UnsubscribeAsync(topic).ConfigureAwait(false);
+ await client.DisconnectAsync().ConfigureAwait(false);
+ }
Write("[OK]\n", ConsoleColor.Green);
}
@@ -174,6 +208,5 @@ namespace MQTTnet.TestApp
Console.ForegroundColor = color;
Console.Write(message);
}
-
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Tests/RPC_Tests.cs b/Source/MQTTnet.Tests/Extensions/Rpc_Tests.cs
similarity index 95%
rename from Source/MQTTnet.Tests/RPC_Tests.cs
rename to Source/MQTTnet.Tests/Extensions/Rpc_Tests.cs
index f266f5a6..b48d5053 100644
--- a/Source/MQTTnet.Tests/RPC_Tests.cs
+++ b/Source/MQTTnet.Tests/Extensions/Rpc_Tests.cs
@@ -6,68 +6,93 @@ using System;
using System.Text;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
-using MQTTnet.Tests.Mockups;
using MQTTnet.Client;
using MQTTnet.Exceptions;
using MQTTnet.Extensions.Rpc;
-using MQTTnet.Protocol;
using MQTTnet.Formatter;
-using MQTTnet.Implementations;
using MQTTnet.Internal;
+using MQTTnet.Protocol;
+using MQTTnet.Tests.Mockups;
-namespace MQTTnet.Tests
+namespace MQTTnet.Tests.Extensions
{
[TestClass]
- public sealed class RPC_Tests : BaseTestClass
+ public sealed class Rpc_Tests : BaseTestClass
{
+ [TestMethod]
+ public async Task Execute_Success_MQTT_V5_Mixed_Clients()
+ {
+ using (var testEnvironment = CreateTestEnvironment())
+ {
+ await testEnvironment.StartServer();
+ var responseSender = await testEnvironment.ConnectClient();
+ await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping");
+
+ responseSender.ApplicationMessageReceivedAsync += async e =>
+ {
+ Assert.IsNull(e.ApplicationMessage.ResponseTopic);
+ await responseSender.PublishStringAsync(e.ApplicationMessage.Topic + "/response", "pong");
+ };
+
+ var requestSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
+
+ using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
+ {
+ var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
+
+ Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
+ }
+ }
+ }
+
[TestMethod]
public Task Execute_Success_With_QoS_0()
{
return Execute_Success(MqttQualityOfServiceLevel.AtMostOnce, MqttProtocolVersion.V311);
}
- [TestMethod]
- public Task Execute_Success_With_QoS_1()
- {
- return Execute_Success(MqttQualityOfServiceLevel.AtLeastOnce, MqttProtocolVersion.V311);
- }
-
- [TestMethod]
- public Task Execute_Success_With_QoS_2()
- {
- return Execute_Success(MqttQualityOfServiceLevel.ExactlyOnce, MqttProtocolVersion.V311);
- }
-
[TestMethod]
public Task Execute_Success_With_QoS_0_MQTT_V5()
{
return Execute_Success(MqttQualityOfServiceLevel.AtMostOnce, MqttProtocolVersion.V500);
}
- [TestMethod]
- public Task Execute_Success_With_QoS_1_MQTT_V5()
- {
- return Execute_Success(MqttQualityOfServiceLevel.AtLeastOnce, MqttProtocolVersion.V500);
- }
-
- [TestMethod]
- public Task Execute_Success_With_QoS_2_MQTT_V5()
- {
- return Execute_Success(MqttQualityOfServiceLevel.ExactlyOnce, MqttProtocolVersion.V500);
- }
-
[TestMethod]
public Task Execute_Success_With_QoS_0_MQTT_V5_Use_ResponseTopic()
{
return Execute_Success_MQTT_V5(MqttQualityOfServiceLevel.AtMostOnce);
}
+ [TestMethod]
+ public Task Execute_Success_With_QoS_1()
+ {
+ return Execute_Success(MqttQualityOfServiceLevel.AtLeastOnce, MqttProtocolVersion.V311);
+ }
+
+ [TestMethod]
+ public Task Execute_Success_With_QoS_1_MQTT_V5()
+ {
+ return Execute_Success(MqttQualityOfServiceLevel.AtLeastOnce, MqttProtocolVersion.V500);
+ }
+
[TestMethod]
public Task Execute_Success_With_QoS_1_MQTT_V5_Use_ResponseTopic()
{
return Execute_Success_MQTT_V5(MqttQualityOfServiceLevel.AtLeastOnce);
}
+ [TestMethod]
+ public Task Execute_Success_With_QoS_2()
+ {
+ return Execute_Success(MqttQualityOfServiceLevel.ExactlyOnce, MqttProtocolVersion.V311);
+ }
+
+ [TestMethod]
+ public Task Execute_Success_With_QoS_2_MQTT_V5()
+ {
+ return Execute_Success(MqttQualityOfServiceLevel.ExactlyOnce, MqttProtocolVersion.V500);
+ }
+
[TestMethod]
public Task Execute_Success_With_QoS_2_MQTT_V5_Use_ResponseTopic()
{
@@ -90,6 +115,31 @@ namespace MQTTnet.Tests
}
}
+ [TestMethod]
+ [ExpectedException(typeof(MqttCommunicationTimedOutException))]
+ public async Task Execute_Timeout_MQTT_V5_Mixed_Clients()
+ {
+ using (var testEnvironment = new TestEnvironment(TestContext))
+ {
+ await testEnvironment.StartServer();
+ var responseSender = await testEnvironment.ConnectClient();
+ await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping");
+
+ responseSender.ApplicationMessageReceivedAsync += async e =>
+ {
+ Assert.IsNull(e.ApplicationMessage.ResponseTopic);
+ await CompletedTask.Instance;
+ };
+
+ var requestSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
+
+ using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
+ {
+ var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
+ }
+ }
+ }
+
[TestMethod]
[ExpectedException(typeof(MqttCommunicationTimedOutException))]
public async Task Execute_With_Custom_Topic_Names()
@@ -98,14 +148,24 @@ namespace MQTTnet.Tests
{
await testEnvironment.StartServer();
- var requestSender = await testEnvironment.ConnectClient();
-
var rpcClient = await testEnvironment.ConnectRpcClient(new MqttRpcClientOptionsBuilder().WithTopicGenerationStrategy(new TestTopicStrategy()).Build());
await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
}
}
+ [TestMethod]
+ public void Use_Factory()
+ {
+ var factory = new MqttFactory();
+ using (var client = factory.CreateMqttClient())
+ {
+ var rpcClient = factory.CreateMqttRpcClient(client);
+
+ Assert.IsNotNull(rpcClient);
+ }
+ }
+
async Task Execute_Success(MqttQualityOfServiceLevel qosLevel, MqttProtocolVersion protocolVersion)
{
using (var testEnvironment = CreateTestEnvironment())
@@ -151,57 +211,6 @@ namespace MQTTnet.Tests
}
}
- [TestMethod]
- public async Task Execute_Success_MQTT_V5_Mixed_Clients()
- {
- using (var testEnvironment = CreateTestEnvironment())
- {
- await testEnvironment.StartServer();
- var responseSender = await testEnvironment.ConnectClient();
- await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping", MqttQualityOfServiceLevel.AtMostOnce);
-
- responseSender.ApplicationMessageReceivedAsync += async e =>
- {
- Assert.IsNull(e.ApplicationMessage.ResponseTopic);
- await responseSender.PublishStringAsync(e.ApplicationMessage.Topic + "/response", "pong");
- };
-
- var requestSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
-
- using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
- {
- var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
-
- Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
- }
- }
- }
-
- [TestMethod]
- [ExpectedException(typeof(MqttCommunicationTimedOutException))]
- public async Task Execute_Timeout_MQTT_V5_Mixed_Clients()
- {
- using (var testEnvironment = new TestEnvironment(TestContext))
- {
- await testEnvironment.StartServer();
- var responseSender = await testEnvironment.ConnectClient();
- await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping", MqttQualityOfServiceLevel.AtMostOnce);
-
- responseSender.ApplicationMessageReceivedAsync += async e =>
- {
- Assert.IsNull(e.ApplicationMessage.ResponseTopic);
- await CompletedTask.Instance;
- };
-
- var requestSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
-
- using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
- {
- var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
- }
- }
- }
-
class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
{
public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
@@ -214,4 +223,4 @@ namespace MQTTnet.Tests
}
}
}
-}
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Tests/Extensions/WebSocket4Net_Tests.cs b/Source/MQTTnet.Tests/Extensions/WebSocket4Net_Tests.cs
new file mode 100644
index 00000000..8c70fef0
--- /dev/null
+++ b/Source/MQTTnet.Tests/Extensions/WebSocket4Net_Tests.cs
@@ -0,0 +1,51 @@
+// 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;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Client;
+using MQTTnet.Exceptions;
+using MQTTnet.Extensions.WebSocket4Net;
+using MQTTnet.Tests.Helpers;
+
+namespace MQTTnet.Tests.Extensions
+{
+ [TestClass]
+ public sealed class WebSocket4Net_Tests
+ {
+ [TestMethod]
+ [ExpectedException(typeof(MqttCommunicationException))]
+ public async Task Connect_Failed_With_Invalid_Server()
+ {
+ var factory = new MqttFactory().UseWebSocket4Net();
+
+ using (var client = factory.CreateMqttClient())
+ {
+ var options = new MqttClientOptionsBuilder().WithWebSocketServer("ws://a.b/mqtt").WithTimeout(TimeSpan.FromSeconds(2)).Build();
+ await client.ConnectAsync(options).ConfigureAwait(false);
+ }
+ }
+
+ [TestMethod]
+ public void Use_Correct_Adapter()
+ {
+ // Should not throw exception
+ var factory = new MqttFactory().UseWebSocket4Net();
+ using (var client = factory.CreateMqttClient())
+ {
+ var adapterFactory = client.GetFieldValue("_adapterFactory");
+
+ Assert.IsInstanceOfType(adapterFactory, typeof(WebSocket4NetMqttClientAdapterFactory));
+ }
+ }
+
+ [TestMethod]
+ public void Use_WebSocket4Net()
+ {
+ // Should not throw exception
+ new MqttFactory().UseWebSocket4Net();
+ }
+ }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs
index 673e46b3..6c32b24a 100644
--- a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs
+++ b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs
@@ -15,7 +15,7 @@ using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Packets;
using MQTTnet.Protocol;
-using MQTTnet.Tests.Extensions;
+using MQTTnet.Tests.Helpers;
using MQTTnet.Tests.Mockups;
namespace MQTTnet.Tests.Formatter
diff --git a/Source/MQTTnet.Tests/Extensions/MqttPacketWriterExtensions.cs b/Source/MQTTnet.Tests/Helpers/MqttPacketWriterExtensions.cs
similarity index 95%
rename from Source/MQTTnet.Tests/Extensions/MqttPacketWriterExtensions.cs
rename to Source/MQTTnet.Tests/Helpers/MqttPacketWriterExtensions.cs
index e254354f..1c6e1bf1 100644
--- a/Source/MQTTnet.Tests/Extensions/MqttPacketWriterExtensions.cs
+++ b/Source/MQTTnet.Tests/Helpers/MqttPacketWriterExtensions.cs
@@ -5,7 +5,7 @@
using MQTTnet.Formatter;
using MQTTnet.Protocol;
-namespace MQTTnet.Tests.Extensions
+namespace MQTTnet.Tests.Helpers
{
public static class MqttPacketWriterExtensions
{
diff --git a/Source/MQTTnet.Tests/Helpers/ReflectionExtensions.cs b/Source/MQTTnet.Tests/Helpers/ReflectionExtensions.cs
new file mode 100644
index 00000000..2639153f
--- /dev/null
+++ b/Source/MQTTnet.Tests/Helpers/ReflectionExtensions.cs
@@ -0,0 +1,28 @@
+// 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.Reflection;
+
+namespace MQTTnet.Tests.Helpers
+{
+ public static class ReflectionExtensions
+ {
+ public static object GetFieldValue(this object source, string fieldName)
+ {
+ if (source == null)
+ {
+ throw new ArgumentNullException(nameof(source));
+ }
+
+ var field = source.GetType().GetField(fieldName, BindingFlags.Instance | BindingFlags.NonPublic);
+ if (field == null)
+ {
+ throw new ArgumentException($"Field {fieldName} not found.");
+ }
+
+ return field.GetValue(source);
+ }
+ }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet.Tests/MQTTnet.Tests.csproj b/Source/MQTTnet.Tests/MQTTnet.Tests.csproj
index d9a7ce1e..daf55189 100644
--- a/Source/MQTTnet.Tests/MQTTnet.Tests.csproj
+++ b/Source/MQTTnet.Tests/MQTTnet.Tests.csproj
@@ -22,6 +22,7 @@
+
\ No newline at end of file
diff --git a/Source/MQTTnet.Tests/Server/QoS_Tests.cs b/Source/MQTTnet.Tests/Server/QoS_Tests.cs
index d053cac5..2b5e59d2 100644
--- a/Source/MQTTnet.Tests/Server/QoS_Tests.cs
+++ b/Source/MQTTnet.Tests/Server/QoS_Tests.cs
@@ -27,6 +27,7 @@ namespace MQTTnet.Tests.Server
await dummyClient.SubscribeAsync("#", MqttQualityOfServiceLevel.AtLeastOnce);
dummyClient.Dispose();
+ await LongTestDelay();
await LongTestDelay();
// Now inject messages which are appended to the queue of the client.
diff --git a/Source/MQTTnet.Tests/Server/Session_Tests.cs b/Source/MQTTnet.Tests/Server/Session_Tests.cs
index 6b9af994..b8c5527f 100644
--- a/Source/MQTTnet.Tests/Server/Session_Tests.cs
+++ b/Source/MQTTnet.Tests/Server/Session_Tests.cs
@@ -121,10 +121,9 @@ namespace MQTTnet.Tests.Server
Assert.AreEqual(true, session.Items["can_subscribe_x"]);
}
}
-
- #if !NET452
[TestMethod]
+ [DataRow(MqttProtocolVersion.V310)]
[DataRow(MqttProtocolVersion.V311)]
[DataRow(MqttProtocolVersion.V500)]
public async Task Handle_Parallel_Connection_Attempts(MqttProtocolVersion protocolVersion)
@@ -150,8 +149,6 @@ namespace MQTTnet.Tests.Server
// Try to connect 50 clients at the same time.
var clients = await Task.WhenAll(Enumerable.Range(0, 50).Select(i => ConnectAndSubscribe(testEnvironment, options, OnReceive)));
- await Task.Delay(TimeSpan.FromSeconds(10));
-
var connectedClients = clients.Where(c => c?.TryPingAsync().GetAwaiter().GetResult() == true).ToList();
await LongTestDelay();
@@ -162,13 +159,12 @@ namespace MQTTnet.Tests.Server
var sendClient = await testEnvironment.ConnectClient(option2);
await sendClient.PublishStringAsync("aaa", "1");
- await Task.Delay(TimeSpan.FromSeconds(5));
+ await LongTestDelay();
Assert.AreEqual(true, hasReceive);
}
}
-#endif
-
+
[DataTestMethod]
[DataRow(MqttQualityOfServiceLevel.ExactlyOnce)]
[DataRow(MqttQualityOfServiceLevel.AtLeastOnce)]
diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs
index 50e9223a..47ee9b6b 100644
--- a/Source/MQTTnet/Client/MqttClient.cs
+++ b/Source/MQTTnet/Client/MqttClient.cs
@@ -19,7 +19,6 @@ namespace MQTTnet.Client
public sealed class MqttClient : Disposable, IMqttClient
{
readonly IMqttClientAdapterFactory _adapterFactory;
- readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory();
readonly AsyncEvent _applicationMessageReceivedEvent = new AsyncEvent();
readonly MqttClientPublishResultFactory _clientPublishResultFactory = new MqttClientPublishResultFactory();
@@ -32,7 +31,6 @@ namespace MQTTnet.Client
readonly object _disconnectLock = new object();
readonly AsyncEvent _inspectPacketEvent = new AsyncEvent();
readonly MqttNetSourceLogger _logger;
- readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
readonly IMqttNetLogger _rootLogger;
@@ -207,7 +205,7 @@ namespace MQTTnet.Client
if (clientWasConnected)
{
- var disconnectPacket = _packetFactories.Disconnect.Create(options);
+ var disconnectPacket = MqttPacketFactories.Disconnect.Create(options);
await SendAsync(disconnectPacket, cancellationToken).ConfigureAwait(false);
}
}
@@ -241,7 +239,7 @@ namespace MQTTnet.Client
ThrowIfDisposed();
ThrowIfNotConnected();
- var publishPacket = _packetFactories.Publish.Create(applicationMessage);
+ var publishPacket = MqttPacketFactories.Publish.Create(applicationMessage);
switch (applicationMessage.QualityOfServiceLevel)
{
@@ -301,7 +299,7 @@ namespace MQTTnet.Client
ThrowIfDisposed();
ThrowIfNotConnected();
- var subscribePacket = _packetFactories.Subscribe.Create(options);
+ var subscribePacket = MqttPacketFactories.Subscribe.Create(options);
subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
var subAckPacket = await SendAndReceiveAsync(subscribePacket, cancellationToken).ConfigureAwait(false);
@@ -319,7 +317,7 @@ namespace MQTTnet.Client
ThrowIfDisposed();
ThrowIfNotConnected();
- var unsubscribePacket = _packetFactories.Unsubscribe.Create(options);
+ var unsubscribePacket = MqttPacketFactories.Unsubscribe.Create(options);
unsubscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
var unsubAckPacket = await SendAndReceiveAsync(unsubscribePacket, cancellationToken).ConfigureAwait(false);
@@ -347,7 +345,7 @@ namespace MQTTnet.Client
{
if (!eventArgs.ProcessingFailed)
{
- var pubAckPacket = _packetFactories.PubAck.Create(eventArgs);
+ var pubAckPacket = MqttPacketFactories.PubAck.Create(eventArgs);
return SendAsync(pubAckPacket, cancellationToken);
}
}
@@ -355,7 +353,7 @@ namespace MQTTnet.Client
{
if (!eventArgs.ProcessingFailed)
{
- var pubRecPacket = _packetFactories.PubRec.Create(eventArgs);
+ var pubRecPacket = MqttPacketFactories.PubRec.Create(eventArgs);
return SendAsync(pubRecPacket, cancellationToken);
}
}
@@ -373,7 +371,7 @@ namespace MQTTnet.Client
try
{
- var connectPacket = _packetFactories.Connect.Create(options);
+ var connectPacket = MqttPacketFactories.Connect.Create(options);
var connAckPacket = await SendAndReceiveAsync(connectPacket, cancellationToken).ConfigureAwait(false);
@@ -535,7 +533,7 @@ namespace MQTTnet.Client
async Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
{
- var applicationMessage = _applicationMessageFactory.Create(publishPacket);
+ var applicationMessage = MqttApplicationMessageFactory.Create(publishPacket);
var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage, publishPacket, AcknowledgeReceivedPublishPacket);
await _applicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
@@ -601,7 +599,7 @@ namespace MQTTnet.Client
{
// The packet is unknown. Probably due to a restart of the client.
// So wen send this to the server to trigger a full resend of the message.
- var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
+ var pubRelPacket = MqttPacketFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
return SendAsync(pubRelPacket, cancellationToken);
}
@@ -610,7 +608,7 @@ namespace MQTTnet.Client
Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
{
- var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
+ var pubCompPacket = MqttPacketFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
return SendAsync(pubCompPacket, cancellationToken);
}
@@ -637,7 +635,7 @@ namespace MQTTnet.Client
var pubRecPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false);
- var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
+ var pubRelPacket = MqttPacketFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
var pubCompPacket = await SendAndReceiveAsync(pubRelPacket, cancellationToken).ConfigureAwait(false);
diff --git a/Source/MQTTnet/Diagnostics/Logger/MqttNetSourceLoggerExtensions.cs b/Source/MQTTnet/Diagnostics/Logger/MqttNetSourceLoggerExtensions.cs
index 18d70a5f..47874aa0 100644
--- a/Source/MQTTnet/Diagnostics/Logger/MqttNetSourceLoggerExtensions.cs
+++ b/Source/MQTTnet/Diagnostics/Logger/MqttNetSourceLoggerExtensions.cs
@@ -6,17 +6,100 @@ using System;
namespace MQTTnet.Diagnostics
{
+ /*
+ * The logger uses generic parameters in order to avoid boxing of parameter values like integers etc.
+ */
+
public static class MqttNetSourceLoggerExtensions
{
- /*
- * The logger uses generic parameters in order to avoid boxing of parameter values like integers etc.
- */
-
- public static MqttNetSourceLogger WithSource(this IMqttNetLogger logger, string source)
+ public static void Error(this MqttNetSourceLogger logger, Exception exception, string message, TParameter1 parameter1)
{
- if (logger == null) throw new ArgumentNullException(nameof(logger));
+ if (!logger.IsEnabled)
+ {
+ return;
+ }
- return new MqttNetSourceLogger(logger, source);
+ logger.Publish(MqttNetLogLevel.Error, message, new object[] { parameter1 }, exception);
+ }
+
+ public static void Error(this MqttNetSourceLogger logger, Exception exception, string message, TParameter1 parameter1, TParameter2 parameter2)
+ {
+ if (!logger.IsEnabled)
+ {
+ return;
+ }
+
+ logger.Publish(MqttNetLogLevel.Error, message, new object[] { parameter1, parameter2 }, exception);
+ }
+
+ public static void Error(this MqttNetSourceLogger logger, Exception exception, string message)
+ {
+ if (!logger.IsEnabled)
+ {
+ return;
+ }
+
+ logger.Publish(MqttNetLogLevel.Error, message, null, exception);
+ }
+
+ public static void Error(this MqttNetSourceLogger logger, string message)
+ {
+ if (!logger.IsEnabled)
+ {
+ return;
+ }
+
+ logger.Publish(MqttNetLogLevel.Error, message, null, null);
+ }
+
+ public static void Info(this MqttNetSourceLogger logger, string message, TParameter1 parameter1)
+ {
+ if (!logger.IsEnabled)
+ {
+ return;
+ }
+
+ logger.Publish(MqttNetLogLevel.Info, message, new object[] { parameter1 }, null);
+ }
+
+ public static void Info(this MqttNetSourceLogger logger, string message, TParameter1 parameter1, TParameter2 parameter2)
+ {
+ if (!logger.IsEnabled)
+ {
+ return;
+ }
+
+ logger.Publish(MqttNetLogLevel.Info, message, new object[] { parameter1, parameter2 }, null);
+ }
+
+ public static void Info(this MqttNetSourceLogger logger, string message)
+ {
+ if (!logger.IsEnabled)
+ {
+ return;
+ }
+
+ logger.Publish(MqttNetLogLevel.Info, message, null, null);
+ }
+
+ public static void Publish(this MqttNetSourceLogger logger, MqttNetLogLevel logLevel, Exception exception, string message, TParameter1 parameter1)
+ {
+ if (!logger.IsEnabled)
+ {
+ return;
+ }
+
+ logger.Publish(logLevel, message, new object[] { parameter1 }, exception);
+ }
+
+ public static void Publish(this MqttNetSourceLogger logger, MqttNetLogLevel logLevel, Exception exception, string message, TParameter1 parameter1, TParameter2 parameter2)
+ {
+ if (!logger.IsEnabled)
+ {
+ return;
+ }
+
+ logger.Publish(logLevel, message, new object[] { parameter1, parameter2 }, exception);
}
public static void Verbose(this MqttNetSourceLogger logger, string message, TParameter1 parameter1)
@@ -26,7 +109,7 @@ namespace MQTTnet.Diagnostics
return;
}
- logger.Publish(MqttNetLogLevel.Verbose, message, new object[] {parameter1}, null);
+ logger.Publish(MqttNetLogLevel.Verbose, message, new object[] { parameter1 }, null);
}
public static void Verbose(this MqttNetSourceLogger logger, string message, TParameter1 parameter1, TParameter2 parameter2)
@@ -36,10 +119,14 @@ namespace MQTTnet.Diagnostics
return;
}
- logger.Publish(MqttNetLogLevel.Verbose, message, new object[] {parameter1, parameter2}, null);
+ logger.Publish(MqttNetLogLevel.Verbose, message, new object[] { parameter1, parameter2 }, null);
}
- public static void Verbose(this MqttNetSourceLogger logger, string message, TParameter1 parameter1, TParameter2 parameter2,
+ public static void Verbose(
+ this MqttNetSourceLogger logger,
+ string message,
+ TParameter1 parameter1,
+ TParameter2 parameter2,
TParameter3 parameter3)
{
if (!logger.IsEnabled)
@@ -47,7 +134,7 @@ namespace MQTTnet.Diagnostics
return;
}
- logger.Publish(MqttNetLogLevel.Verbose, message, new object[] {parameter1, parameter2, parameter3}, null);
+ logger.Publish(MqttNetLogLevel.Verbose, message, new object[] { parameter1, parameter2, parameter3 }, null);
}
public static void Verbose(this MqttNetSourceLogger logger, string message)
@@ -60,134 +147,74 @@ namespace MQTTnet.Diagnostics
logger.Publish(MqttNetLogLevel.Verbose, message, null, null);
}
- public static void Info(this MqttNetSourceLogger logger, string message, TParameter1 parameter1)
- {
- if (!logger.IsEnabled)
- {
- return;
- }
-
- logger.Publish(MqttNetLogLevel.Info, message, new object[] {parameter1}, null);
- }
-
- public static void Info(this MqttNetSourceLogger logger, string message, TParameter1 parameter1, TParameter2 parameter2)
- {
- if (!logger.IsEnabled)
- {
- return;
- }
-
- logger.Publish(MqttNetLogLevel.Info, message, new object[] {parameter1, parameter2}, null);
- }
-
- public static void Info(this MqttNetSourceLogger logger, string message)
- {
- if (!logger.IsEnabled)
- {
- return;
- }
-
- logger.Publish(MqttNetLogLevel.Info, message, null, null);
- }
-
public static void Warning(this MqttNetSourceLogger logger, Exception exception, string message, TParameter1 parameter1)
{
if (!logger.IsEnabled)
{
return;
}
-
- logger.Publish(MqttNetLogLevel.Warning, message, new object[] {parameter1}, exception);
+
+ logger.Publish(MqttNetLogLevel.Warning, message, new object[] { parameter1 }, exception);
}
-
+
public static void Warning(this MqttNetSourceLogger logger, Exception exception, string message, TParameter1 parameter1, TParameter2 parameter2)
{
if (!logger.IsEnabled)
{
return;
}
-
- logger.Publish(MqttNetLogLevel.Warning, message, new object[] {parameter1, parameter2}, exception);
+
+ logger.Publish(MqttNetLogLevel.Warning, message, new object[] { parameter1, parameter2 }, exception);
}
-
+
public static void Warning(this MqttNetSourceLogger logger, Exception exception, string message)
{
if (!logger.IsEnabled)
{
return;
}
-
+
logger.Publish(MqttNetLogLevel.Warning, message, null, exception);
}
-
+
public static void Warning(this MqttNetSourceLogger logger, string message, TParameter1 parameter1)
{
if (!logger.IsEnabled)
{
return;
}
-
- logger.Publish(MqttNetLogLevel.Warning, message, new object[] {parameter1}, null);
+
+ logger.Publish(MqttNetLogLevel.Warning, message, new object[] { parameter1 }, null);
}
-
+
public static void Warning(this MqttNetSourceLogger logger, string message, TParameter1 parameter1, TParameter2 parameter2)
{
if (!logger.IsEnabled)
{
return;
}
-
- logger.Publish(MqttNetLogLevel.Warning, message, new object[] {parameter1, parameter2}, null);
+
+ logger.Publish(MqttNetLogLevel.Warning, message, new object[] { parameter1, parameter2 }, null);
}
-
+
public static void Warning(this MqttNetSourceLogger logger, string message)
{
if (!logger.IsEnabled)
{
return;
}
-
+
logger.Publish(MqttNetLogLevel.Warning, message, null, null);
}
- public static void Error(this MqttNetSourceLogger logger, Exception exception, string message, TParameter1 parameter1)
+ public static MqttNetSourceLogger WithSource(this IMqttNetLogger logger, string source)
{
- if (!logger.IsEnabled)
+ if (logger == null)
{
- return;
+ throw new ArgumentNullException(nameof(logger));
}
-
- logger.Publish(MqttNetLogLevel.Error, message, new object[] {parameter1}, exception);
- }
-
- public static void Error(this MqttNetSourceLogger logger, Exception exception, string message, TParameter1 parameter1, TParameter2 parameter2)
- {
- if (!logger.IsEnabled)
- {
- return;
- }
-
- logger.Publish(MqttNetLogLevel.Error, message, new object[] {parameter1, parameter2}, exception);
- }
- public static void Error(this MqttNetSourceLogger logger, Exception exception, string message)
- {
- if (!logger.IsEnabled)
- {
- return;
- }
-
- logger.Publish(MqttNetLogLevel.Error, message, null, exception);
- }
-
- public static void Error(this MqttNetSourceLogger logger, string message)
- {
- if (!logger.IsEnabled)
- {
- return;
- }
-
- logger.Publish(MqttNetLogLevel.Error, message, null, null);
+ return new MqttNetSourceLogger(logger, source);
}
}
}
\ No newline at end of file
diff --git a/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs b/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs
index 0e80f1df..1fd23bf7 100644
--- a/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs
+++ b/Source/MQTTnet/Formatter/MqttApplicationMessageFactory.cs
@@ -7,9 +7,9 @@ using MQTTnet.Packets;
namespace MQTTnet.Formatter
{
- public sealed class MqttApplicationMessageFactory
+ public static class MqttApplicationMessageFactory
{
- public MqttApplicationMessage Create(MqttPublishPacket publishPacket)
+ public static MqttApplicationMessage Create(MqttPublishPacket publishPacket)
{
if (publishPacket == null)
{
diff --git a/Source/MQTTnet/Formatter/MqttPacketFactories.cs b/Source/MQTTnet/Formatter/MqttPacketFactories.cs
index 70de69d2..19264230 100644
--- a/Source/MQTTnet/Formatter/MqttPacketFactories.cs
+++ b/Source/MQTTnet/Formatter/MqttPacketFactories.cs
@@ -4,30 +4,30 @@
namespace MQTTnet.Formatter
{
- public sealed class MqttPacketFactories
+ public static class MqttPacketFactories
{
- public MqttConnAckPacketFactory ConnAck { get; } = new MqttConnAckPacketFactory();
+ public static MqttConnAckPacketFactory ConnAck { get; } = new MqttConnAckPacketFactory();
- public MqttConnectPacketFactory Connect { get; } = new MqttConnectPacketFactory();
+ public static MqttConnectPacketFactory Connect { get; } = new MqttConnectPacketFactory();
- public MqttDisconnectPacketFactory Disconnect { get; } = new MqttDisconnectPacketFactory();
+ public static MqttDisconnectPacketFactory Disconnect { get; } = new MqttDisconnectPacketFactory();
- public MqttPubAckPacketFactory PubAck { get; } = new MqttPubAckPacketFactory();
+ public static MqttPubAckPacketFactory PubAck { get; } = new MqttPubAckPacketFactory();
- public MqttPubCompPacketFactory PubComp { get; } = new MqttPubCompPacketFactory();
+ public static MqttPubCompPacketFactory PubComp { get; } = new MqttPubCompPacketFactory();
- public MqttPublishPacketFactory Publish { get; } = new MqttPublishPacketFactory();
+ public static MqttPublishPacketFactory Publish { get; } = new MqttPublishPacketFactory();
- public MqttPubRecPacketFactory PubRec { get; } = new MqttPubRecPacketFactory();
+ public static MqttPubRecPacketFactory PubRec { get; } = new MqttPubRecPacketFactory();
- public MqttPubRelPacketFactory PubRel { get; } = new MqttPubRelPacketFactory();
+ public static MqttPubRelPacketFactory PubRel { get; } = new MqttPubRelPacketFactory();
- public MqttSubAckPacketFactory SubAck { get; } = new MqttSubAckPacketFactory();
+ public static MqttSubAckPacketFactory SubAck { get; } = new MqttSubAckPacketFactory();
- public MqttSubscribePacketFactory Subscribe { get; } = new MqttSubscribePacketFactory();
+ public static MqttSubscribePacketFactory Subscribe { get; } = new MqttSubscribePacketFactory();
- public MqttUnsubAckPacketFactory UnsubAck { get; } = new MqttUnsubAckPacketFactory();
+ public static MqttUnsubAckPacketFactory UnsubAck { get; } = new MqttUnsubAckPacketFactory();
- public MqttUnsubscribePacketFactory Unsubscribe { get; } = new MqttUnsubscribePacketFactory();
+ public static MqttUnsubscribePacketFactory Unsubscribe { get; } = new MqttUnsubscribePacketFactory();
}
}
\ No newline at end of file
diff --git a/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs
index f9651d81..2924d86e 100644
--- a/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs
+++ b/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs
@@ -78,7 +78,7 @@ namespace MQTTnet.Formatter.V3
return DecodeSubscribePacket(receivedMqttPacket.Body);
case MqttControlPacketType.SubAck:
return DecodeSubAckPacket(receivedMqttPacket.Body);
- case MqttControlPacketType.Unsubscibe:
+ case MqttControlPacketType.Unsubscribe:
return DecodeUnsubscribePacket(receivedMqttPacket.Body);
case MqttControlPacketType.UnsubAck:
return DecodeUnsubAckPacket(receivedMqttPacket.Body);
@@ -783,7 +783,7 @@ namespace MQTTnet.Formatter.V3
}
}
- return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
+ return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Unsubscribe, 0x02);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs b/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs
index 5f875851..9d460411 100644
--- a/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs
+++ b/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs
@@ -54,7 +54,7 @@ namespace MQTTnet.Formatter.V5
return DecodeSubscribePacket(receivedMqttPacket.Body);
case MqttControlPacketType.SubAck:
return DecodeSubAckPacket(receivedMqttPacket.Body);
- case MqttControlPacketType.Unsubscibe:
+ case MqttControlPacketType.Unsubscribe:
return DecodeUnsubscribePacket(receivedMqttPacket.Body);
case MqttControlPacketType.UnsubAck:
return DecodeUnsubAckPacket(receivedMqttPacket.Body);
diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs
index 31092fa1..ed94c0e4 100644
--- a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs
+++ b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs
@@ -524,7 +524,7 @@ namespace MQTTnet.Formatter.V5
_bufferWriter.WriteString(topicFilter);
}
- return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
+ return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Unsubscribe, 0x02);
}
static void ThrowIfPacketIdentifierIsInvalid(ushort packetIdentifier, MqttPacket packet)
diff --git a/Source/MQTTnet/Internal/CancellationTokenSourceExtensions.cs b/Source/MQTTnet/Internal/CancellationTokenSourceExtensions.cs
new file mode 100644
index 00000000..94c3febc
--- /dev/null
+++ b/Source/MQTTnet/Internal/CancellationTokenSourceExtensions.cs
@@ -0,0 +1,37 @@
+// 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;
+
+namespace MQTTnet.Internal
+{
+ public static class CancellationTokenSourceExtensions
+ {
+ public static bool TryCancel(this CancellationTokenSource cancellationTokenSource, bool throwOnFirstException = false)
+ {
+ if (cancellationTokenSource == null)
+ {
+ return false;
+ }
+
+ try
+ {
+ // Checking the _IsCancellationRequested_ here will not throw an
+ // "ObjectDisposedException" as the getter of the property "Token" will do!
+ if (cancellationTokenSource.IsCancellationRequested)
+ {
+ return false;
+ }
+
+ cancellationTokenSource.Cancel(throwOnFirstException);
+ return true;
+ }
+ catch (ObjectDisposedException)
+ {
+ return false;
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet/Packets/MqttPacketExtensions.cs b/Source/MQTTnet/Packets/MqttPacketExtensions.cs
new file mode 100644
index 00000000..42d2288d
--- /dev/null
+++ b/Source/MQTTnet/Packets/MqttPacketExtensions.cs
@@ -0,0 +1,102 @@
+// 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;
+
+namespace MQTTnet.Packets
+{
+ public static class MqttPacketExtensions
+ {
+ public static string GetRfcName(this MqttPacket packet)
+ {
+ if (packet == null)
+ {
+ throw new ArgumentNullException(nameof(packet));
+ }
+
+ switch (packet)
+ {
+ case MqttConnectPacket _:
+ {
+ return "CONNECT";
+ }
+
+ case MqttConnAckPacket _:
+ {
+ return "CONNACK";
+ }
+
+ case MqttAuthPacket _:
+ {
+ return "AUTH";
+ }
+
+ case MqttDisconnectPacket _:
+ {
+ return "DISCONNECT";
+ }
+
+ case MqttPingReqPacket _:
+ {
+ return "PINGREQ";
+ }
+
+ case MqttPingRespPacket _:
+ {
+ return "PINGRESP";
+ }
+
+ case MqttSubscribePacket _:
+ {
+ return "SUBSCRIBE";
+ }
+
+ case MqttSubAckPacket _:
+ {
+ return "SUBACK";
+ }
+
+ case MqttUnsubscribePacket _:
+ {
+ return "UNSUBSCRIBE";
+ }
+
+ case MqttUnsubAckPacket _:
+ {
+ return "UNSUBACK";
+ }
+
+ case MqttPublishPacket _:
+ {
+ return "PUBLISH";
+ }
+
+ case MqttPubAckPacket _:
+ {
+ return "PUBACK";
+ }
+
+ case MqttPubRelPacket _:
+ {
+ return "PUBREL";
+ }
+
+ case MqttPubRecPacket _:
+ {
+ return "PUBREC";
+ }
+
+ case MqttPubCompPacket _:
+ {
+ return "PUBCOMP";
+ }
+
+ default:
+ {
+ return packet.GetType().Name;
+ }
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet/Protocol/MqttControlPacketType.cs b/Source/MQTTnet/Protocol/MqttControlPacketType.cs
index 22b48e03..37a3fc5d 100644
--- a/Source/MQTTnet/Protocol/MqttControlPacketType.cs
+++ b/Source/MQTTnet/Protocol/MqttControlPacketType.cs
@@ -15,7 +15,7 @@ namespace MQTTnet.Protocol
PubComp = 7,
Subscribe = 8,
SubAck = 9,
- Unsubscibe = 10,
+ Unsubscribe = 10,
UnsubAck = 11,
PingReq = 12,
PingResp = 13,
diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs
index 463ca2ec..6c4b9bc8 100644
--- a/Source/MQTTnet/Server/Internal/MqttClient.cs
+++ b/Source/MQTTnet/Server/Internal/MqttClient.cs
@@ -12,27 +12,21 @@ using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Internal;
-using MQTTnet.PacketDispatcher;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Server
{
- public sealed class MqttClient
+ public sealed class MqttClient : IDisposable
{
- readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory();
readonly MqttConnectPacket _connectPacket;
readonly MqttServerEventContainer _eventContainer;
readonly MqttNetSourceLogger _logger;
- readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
- readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
-
readonly MqttServerOptions _serverOptions;
-
readonly MqttClientSessionsManager _sessionsManager;
readonly Dictionary _topicAlias = new Dictionary();
- CancellationTokenSource _cancellationToken;
+ CancellationTokenSource _cancellationToken = new CancellationTokenSource();
bool _disconnectPacketSent;
public MqttClient(
@@ -45,14 +39,13 @@ namespace MQTTnet.Server
IMqttNetLogger logger)
{
_serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
- _eventContainer = eventContainer;
+ _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
_sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
+ _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
Endpoint = channelAdapter.Endpoint;
-
Session = session ?? throw new ArgumentNullException(nameof(session));
- _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
if (logger == null)
{
@@ -80,6 +73,11 @@ namespace MQTTnet.Server
public MqttClientStatistics Statistics { get; } = new MqttClientStatistics();
+ public void Dispose()
+ {
+ _cancellationToken?.Dispose();
+ }
+
public void ResetStatistics()
{
ChannelAdapter.ResetStatistics();
@@ -88,46 +86,43 @@ namespace MQTTnet.Server
public async Task RunAsync()
{
- _logger.Info("Client '{0}': Session started.", Id);
+ _logger.Info("Client '{0}': Session started", Id);
Session.LatestConnectPacket = _connectPacket;
Session.WillMessageSent = false;
- using (_cancellationToken = new CancellationTokenSource())
+ try
{
var cancellationToken = _cancellationToken.Token;
- try
- {
- _ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default).ConfigureAwait(false);
+ _ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default)
+ .ConfigureAwait(false);
- IsRunning = true;
+ IsRunning = true;
- await ReceivePackagesLoop(cancellationToken).ConfigureAwait(false);
- }
- finally
- {
- IsRunning = false;
-
- _cancellationToken?.Cancel();
- _cancellationToken = null;
- }
+ await ReceivePackagesLoop(cancellationToken).ConfigureAwait(false);
}
+ finally
+ {
+ IsRunning = false;
- _packetDispatcher.CancelAll();
+ _cancellationToken?.TryCancel();
+ _cancellationToken?.Dispose();
+ _cancellationToken = null;
+ }
if (!IsTakenOver && !IsCleanDisconnect && Session.LatestConnectPacket.WillFlag && !Session.WillMessageSent)
{
- var willPublishPacket = _packetFactories.Publish.Create(Session.LatestConnectPacket);
- var willApplicationMessage = _applicationMessageFactory.Create(willPublishPacket);
+ var willPublishPacket = MqttPacketFactories.Publish.Create(Session.LatestConnectPacket);
+ var willApplicationMessage = MqttApplicationMessageFactory.Create(willPublishPacket);
_ = _sessionsManager.DispatchApplicationMessage(Id, willApplicationMessage);
Session.WillMessageSent = true;
- _logger.Info("Client '{0}': Published will message.", Id);
+ _logger.Info("Client '{0}': Published will message", Id);
}
- _logger.Info("Client '{0}': Connection stopped.", Id);
+ _logger.Info("Client '{0}': Connection stopped", Id);
}
public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
@@ -181,6 +176,12 @@ namespace MQTTnet.Server
return CompletedTask.Instance;
}
+ void HandleIncomingPingReqPacket()
+ {
+ // See: The Server MUST send a PINGRESP packet in response to a PINGREQ packet [MQTT-3.12.4-1].
+ Session.EnqueueHealthPacket(new MqttPacketBusItem(MqttPingRespPacket.Instance));
+ }
+
Task HandleIncomingPubAckPacket(MqttPubAckPacket pubAckPacket)
{
var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier);
@@ -210,7 +211,7 @@ namespace MQTTnet.Server
HandleTopicAlias(publishPacket);
InterceptingPublishEventArgs interceptingPublishEventArgs = null;
- var applicationMessage = _applicationMessageFactory.Create(publishPacket);
+ var applicationMessage = MqttApplicationMessageFactory.Create(publishPacket);
var closeConnection = false;
var processPublish = true;
@@ -253,19 +254,19 @@ namespace MQTTnet.Server
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
- var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs, dispatchResult);
+ var pubAckPacket = MqttPacketFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs, dispatchResult);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket));
break;
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
- var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs, dispatchResult);
+ var pubRecPacket = MqttPacketFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs, dispatchResult);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket));
break;
}
default:
{
- throw new MqttCommunicationException("Received a not supported QoS level.");
+ throw new MqttCommunicationException("Received a not supported QoS level");
}
}
}
@@ -274,7 +275,7 @@ namespace MQTTnet.Server
{
// Do not fire the event _ClientAcknowledgedPublishPacket_ here because the QoS 2 process is only finished
// properly when the client has sent the PUBCOMP packet.
- var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
+ var pubRelPacket = MqttPacketFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
return CompletedTask.Instance;
@@ -282,7 +283,7 @@ namespace MQTTnet.Server
void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket)
{
- var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
+ var pubCompPacket = MqttPacketFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubCompPacket));
}
@@ -290,7 +291,7 @@ namespace MQTTnet.Server
{
var subscribeResult = await Session.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false);
- var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult);
+ var subAckPacket = MqttPacketFactories.SubAck.Create(subscribePacket, subscribeResult);
Session.EnqueueControlPacket(new MqttPacketBusItem(subAckPacket));
@@ -304,7 +305,7 @@ namespace MQTTnet.Server
{
foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
{
- var publishPacket = _packetFactories.Publish.Create(retainedMessageMatch);
+ var publishPacket = MqttPacketFactories.Publish.Create(retainedMessageMatch);
Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
}
@@ -314,7 +315,7 @@ namespace MQTTnet.Server
{
var unsubscribeResult = await Session.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false);
- var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult);
+ var unsubAckPacket = MqttPacketFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult);
Session.EnqueueControlPacket(new MqttPacketBusItem(unsubAckPacket));
@@ -345,7 +346,7 @@ namespace MQTTnet.Server
}
else
{
- _logger.Warning("Client '{0}': Received invalid topic alias ({1}).", Id, publishPacket.TopicAlias);
+ _logger.Warning("Client '{0}': Received invalid topic alias ({1})", Id, publishPacket.TopicAlias);
}
}
}
@@ -371,6 +372,7 @@ namespace MQTTnet.Server
async Task ReceivePackagesLoop(CancellationToken cancellationToken)
{
+ MqttPacket currentPacket = null;
try
{
// We do not listen for the cancellation token here because the internal buffer might still
@@ -380,12 +382,12 @@ namespace MQTTnet.Server
{
await Task.Yield();
- var packet = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
- if (packet == null)
+ currentPacket = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
+ if (currentPacket == null)
{
return;
}
-
+
// Check for cancellation again because receive packet might block some time.
if (cancellationToken.IsCancellationRequested)
{
@@ -398,73 +400,69 @@ namespace MQTTnet.Server
{
return;
}
-
+
var processPacket = true;
if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers)
{
- var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, packet, Session.Items);
+ var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, currentPacket, Session.Items);
await _eventContainer.InterceptingInboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
- packet = interceptingPacketEventArgs.Packet;
+ currentPacket = interceptingPacketEventArgs.Packet;
processPacket = interceptingPacketEventArgs.ProcessPacket;
}
- if (!processPacket || packet == null)
+ if (!processPacket || currentPacket == null)
{
// Restart the receiving process to get the next packet ignoring the current one..
continue;
}
- Statistics.HandleReceivedPacket(packet);
+ Statistics.HandleReceivedPacket(currentPacket);
- if (packet is MqttPublishPacket publishPacket)
+ if (currentPacket is MqttPublishPacket publishPacket)
{
await HandleIncomingPublishPacket(publishPacket, cancellationToken).ConfigureAwait(false);
}
- else if (packet is MqttPubAckPacket pubAckPacket)
+ else if (currentPacket is MqttPubAckPacket pubAckPacket)
{
await HandleIncomingPubAckPacket(pubAckPacket).ConfigureAwait(false);
}
- else if (packet is MqttPubCompPacket pubCompPacket)
+ else if (currentPacket is MqttPubCompPacket pubCompPacket)
{
await HandleIncomingPubCompPacket(pubCompPacket).ConfigureAwait(false);
}
- else if (packet is MqttPubRecPacket pubRecPacket)
+ else if (currentPacket is MqttPubRecPacket pubRecPacket)
{
await HandleIncomingPubRecPacket(pubRecPacket).ConfigureAwait(false);
}
- else if (packet is MqttPubRelPacket pubRelPacket)
+ else if (currentPacket is MqttPubRelPacket pubRelPacket)
{
HandleIncomingPubRelPacket(pubRelPacket);
}
- else if (packet is MqttSubscribePacket subscribePacket)
+ else if (currentPacket is MqttSubscribePacket subscribePacket)
{
await HandleIncomingSubscribePacket(subscribePacket, cancellationToken).ConfigureAwait(false);
}
- else if (packet is MqttUnsubscribePacket unsubscribePacket)
+ else if (currentPacket is MqttUnsubscribePacket unsubscribePacket)
{
await HandleIncomingUnsubscribePacket(unsubscribePacket, cancellationToken).ConfigureAwait(false);
}
- else if (packet is MqttPingReqPacket)
+ else if (currentPacket is MqttPingReqPacket)
{
- // See: The Server MUST send a PINGRESP packet in response to a PINGREQ packet [MQTT-3.12.4-1].
- Session.EnqueueHealthPacket(new MqttPacketBusItem(MqttPingRespPacket.Instance));
+ HandleIncomingPingReqPacket();
}
- else if (packet is MqttPingRespPacket)
+ else if (currentPacket is MqttPingRespPacket)
{
throw new MqttProtocolViolationException("A PINGRESP Packet is sent by the Server to the Client in response to a PINGREQ Packet only.");
}
- else if (packet is MqttDisconnectPacket)
+ else if (currentPacket is MqttDisconnectPacket)
{
IsCleanDisconnect = true;
return;
}
else
{
- if (!_packetDispatcher.TryDispatch(packet))
- {
- throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time.");
- }
+ throw new MqttProtocolViolationException("Packet not allowed");
}
}
}
@@ -475,11 +473,25 @@ namespace MQTTnet.Server
{
if (exception is MqttCommunicationException)
{
- _logger.Warning(exception, "Client '{0}': Communication exception while receiving client packets.", Id);
+ _logger.Warning(exception, "Client '{0}': Communication exception while receiving packets", Id);
+ return;
+ }
+
+ var logLevel = MqttNetLogLevel.Error;
+
+ if (!IsRunning)
+ {
+ // There was an exception but the connection is already closed. So there is no chance to send a response to the client etc.
+ logLevel = MqttNetLogLevel.Warning;
+ }
+
+ if (currentPacket == null)
+ {
+ _logger.Publish(logLevel, exception, "Client '{0}': Error while receiving packets", Id);
}
else
{
- _logger.Error(exception, "Client '{0}': Error while receiving client packets.", Id);
+ _logger.Publish(logLevel, exception, "Client '{0}': Error while processing {1} packet", Id, currentPacket.GetRfcName());
}
}
}
@@ -531,15 +543,15 @@ namespace MQTTnet.Server
{
if (exception is MqttCommunicationTimedOutException)
{
- _logger.Warning(exception, "Client '{0}': Sending publish packet failed: Timeout.", Id);
+ _logger.Warning(exception, "Client '{0}': Sending PUBLISH packet failed due to timeout", Id);
}
else if (exception is MqttCommunicationException)
{
- _logger.Warning(exception, "Client '{0}': Sending publish packet failed: Communication exception.", Id);
+ _logger.Warning(exception, "Client '{0}': Sending PUBLISH packet failed due to communication exception", Id);
}
else
{
- _logger.Error(exception, "Client '{0}': Sending publish packet failed.", Id);
+ _logger.Error(exception, "Client '{0}': Sending PUBLISH packet failed", Id);
}
if (packetBusItem?.Packet is MqttPublishPacket publishPacket)
@@ -557,15 +569,7 @@ namespace MQTTnet.Server
void StopInternal()
{
- try
- {
- _cancellationToken?.Cancel();
- }
- catch (ObjectDisposedException)
- {
- // This can happen when connections are created and dropped very quickly.
- // It is not an issue if the cancellation token cannot be canceled multiple times.
- }
+ _cancellationToken?.TryCancel();
}
async Task TrySendDisconnectPacket(MqttDisconnectReasonCode reasonCode)
@@ -574,8 +578,8 @@ namespace MQTTnet.Server
{
// This also indicates that it was tried at least!
_disconnectPacketSent = true;
-
- var disconnectPacket = _packetFactories.Disconnect.Create(reasonCode);
+
+ var disconnectPacket = MqttPacketFactories.Disconnect.Create(reasonCode);
using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
{
@@ -584,7 +588,7 @@ namespace MQTTnet.Server
}
catch (Exception exception)
{
- _logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1}).", Id, reasonCode);
+ _logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, reasonCode);
}
}
}
diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
index 9fb9ca18..75dc3f9f 100644
--- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
@@ -28,7 +28,6 @@ namespace MQTTnet.Server
readonly MqttServerEventContainer _eventContainer;
readonly MqttNetSourceLogger _logger;
readonly MqttServerOptions _options;
- readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
readonly MqttRetainedMessagesManager _retainedMessagesManager;
readonly IMqttNetLogger _rootLogger;
@@ -166,7 +165,7 @@ namespace MQTTnet.Server
continue;
}
- var newPublishPacket = _packetFactories.Publish.Create(applicationMessage);
+ var newPublishPacket = MqttPacketFactories.Publish.Create(applicationMessage);
newPublishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
newPublishPacket.SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers;
@@ -288,7 +287,7 @@ namespace MQTTnet.Server
}
var validatingConnectionEventArgs = await ValidateConnection(connectPacket, channelAdapter).ConfigureAwait(false);
- var connAckPacket = _packetFactories.ConnAck.Create(validatingConnectionEventArgs);
+ var connAckPacket = MqttPacketFactories.ConnAck.Create(validatingConnectionEventArgs);
if (validatingConnectionEventArgs.ReasonCode != MqttConnectReasonCode.Success)
{
@@ -430,7 +429,7 @@ namespace MQTTnet.Server
{
foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
{
- var publishPacket = _packetFactories.Publish.Create(retainedMessageMatch);
+ var publishPacket = MqttPacketFactories.Publish.Create(retainedMessageMatch);
clientSession.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
}