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

Merge pull request #2 from chkr1011/develop

Browse Source 
From develop
This commit is contained in:
Gerardo
2017-10-26 12:09:02 +02:00
committed by GitHub
parent 29918fb454 f85721fbf5
commit e6d811ea7c
49 changed files with 586 additions and 381 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Build/MQTTnet.nuspec
+5
View File
@@ -13,12 +13,17 @@
<releaseNotes>* [Core] Merged the .NET Framwork and netstandard projects (Thanks to @JanEggers)
* [Core] Migrated the trace to a non-static approach (Breaking Change!)
* [Core] Added a builder for application messages using a fluent API
* [Core] Introduced CI (Thanks to @JanEggers)
* [Core] Added interfaces for publishing and receiving which applies to every client and server (Thanks to @ChristianRiedl)
* [Client] Added a first version of a managed client which will manage the connection, subscription etc. automatically (Thanks to @JTrotta)
* [Client] The session state response from the server is now returned in the _ConnectAsync_ method and also part of the _Connected_ event args
* [Client] Added a _TopicFilterBuilder_ using a fluent API (Namespace Changes!)
* [Client] Added several new options for the WebSocket channel (Thanks to @ChristianRiedl)
* [Client] Refactored the options and added a builder using a fluent API (Breaking Change!)
* [Client] Added more options for WebSocket connections like RequestHeaders, SubProtocol etc.
* [Server] Added support for WebSockets via ASP.NET Core 2.0 (Thanks to @ChristianRiedl)
* [Server] Added support for a custom application message interceptor
* [Server] Fixed an issue with dropped connections on UWP (Thanks to @haeberle)
</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2017</copyright>
<tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation</tags>
Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs
-1
View File
@@ -1,7 +1,6 @@
using System;
using Microsoft.AspNetCore.Builder;
using Microsoft.Extensions.DependencyInjection;
using MQTTnet.AspNetCore;
using MQTTnet.Core.Server;
namespace MQTTnet.AspnetCore
Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs
+1 -1
View File
@@ -8,7 +8,7 @@ using MQTTnet.Core.Channel;
using MQTTnet.Core.Server;
using MQTTnet.Implementations;
namespace MQTTnet.AspNetCore
namespace MQTTnet.AspnetCore
{
public class MqttWebSocketServerAdapter : IMqttServerAdapter, IDisposable
{
Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs
-1
View File
@@ -1,6 +1,5 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using MQTTnet.AspNetCore;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Server;
Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs
+1 -2
View File
@@ -7,10 +7,8 @@ using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Serializer;
using MQTTnet.Core.Server;
using Microsoft.Extensions.Logging;
using MQTTnet.Core.Client;
namespace MQTTnet.Implementations
{
@@ -18,6 +16,7 @@ namespace MQTTnet.Implementations
{
private readonly ILogger<MqttServerAdapter> _logger;
private readonly IMqttCommunicationAdapterFactory _mqttCommunicationAdapterFactory;
private CancellationTokenSource _cancellationTokenSource;
private Socket _defaultEndpointSocket;
private Socket _tlsEndpointSocket;
Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs
+6 -6
View File
@@ -15,7 +15,7 @@ namespace MQTTnet.Implementations
{
private readonly MqttClientTcpOptions _options;
//todo: this can be used with min dependency NetStandard1.6
//todo: this can be used with min dependency NetStandard1.6
#if NET45
// ReSharper disable once MemberCanBePrivate.Global
// ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global
@@ -24,7 +24,7 @@ namespace MQTTnet.Implementations
private Socket _socket;
private SslStream _sslStream;
/// <summary>
/// called on client sockets are created in connect
/// </summary>
@@ -48,7 +48,7 @@ namespace MQTTnet.Implementations
public Stream SendStream { get; private set; }
public Stream ReceiveStream { get; private set; }
public Stream RawReceiveStream { get; private set; }
public static Func<X509Certificate, X509Chain, SslPolicyErrors, MqttClientTcpOptions, bool> CustomCertificateValidationCallback { get; set; }
public async Task ConnectAsync()
@@ -58,7 +58,7 @@ namespace MQTTnet.Implementations
_socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
}
//todo: else brach can be used with min dependency NET46
//todo: else brach can be used with min dependency NET46
#if NET451
await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, _options.Server, _options.GetPort(), null).ConfigureAwait(false);
#else
@@ -120,7 +120,7 @@ namespace MQTTnet.Implementations
return _options.TlsOptions.AllowUntrustedCertificates;
}
private static X509CertificateCollection LoadCertificates(IMqttClientOptions options)
private static X509CertificateCollection LoadCertificates(MqttClientTcpOptions options)
{
var certificates = new X509CertificateCollection();
if (options.TlsOptions.Certificates == null)
@@ -144,7 +144,7 @@ namespace MQTTnet.Implementations
//cannot use this as default buffering prevents from receiving the first connect message
//need two streams otherwise read and write have to be synchronized
//todo: if branch can be used with min dependency NetStandard1.6
//todo: if branch can be used with min dependency NetStandard1.6
#if NET45
SendStream = new BufferedStream(RawReceiveStream, BufferSize);
ReceiveStream = new BufferedStream(RawReceiveStream, BufferSize);
Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs
+5 -2
View File
@@ -32,7 +32,6 @@ namespace MQTTnet.Implementations
}
_webSocket = new ClientWebSocket();
_webSocket.Options.KeepAliveInterval = _options.KeepAlivePeriod;
if (_options.RequestHeaders != null)
{
@@ -77,12 +76,16 @@ namespace MQTTnet.Implementations
return;
}
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false);
if (_webSocket.State == WebSocketState.Open || _webSocket.State == WebSocketState.Connecting)
{
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false);
}
}
public void Dispose()
{
_webSocket?.Dispose();
_webSocket = null;
}
}
}
Frameworks/MQTTnet.NetStandard/MqttFactory.cs
+12 -8
View File
@@ -7,7 +7,6 @@ using Microsoft.Extensions.DependencyInjection;
using MQTTnet.Implementations;
using MQTTnet.Core.ManagedClient;
using MQTTnet.Core.Server;
using System.Linq;
using MQTTnet.Core.Channel;
namespace MQTTnet
@@ -18,7 +17,7 @@ namespace MQTTnet
private static IServiceProvider BuildServiceProvider()
{
var serviceProvider = new ServiceCollection()
var serviceProvider = new ServiceCollection()
.AddMqttClient()
.AddMqttServer()
.AddLogging()
@@ -40,10 +39,15 @@ namespace MQTTnet
_serviceProvider = serviceProvider;
}
public ILoggerFactory GetLoggerFactory()
{
return _serviceProvider.GetRequiredService<ILoggerFactory>();
}
public IMqttCommunicationAdapter CreateClientMqttCommunicationAdapter(IMqttClientOptions options)
{
var logger = _serviceProvider.GetRequiredService<ILogger<MqttChannelCommunicationAdapter>>();
return new MqttChannelCommunicationAdapter(CreateMqttCommunicationChannel(options), CreateSerializer(options.ProtocolVersion), logger);
return new MqttChannelCommunicationAdapter(CreateMqttCommunicationChannel(options.ChannelOptions), CreateSerializer(options.ProtocolVersion), logger);
}
public IMqttCommunicationAdapter CreateServerMqttCommunicationAdapter(IMqttCommunicationChannel channel)
@@ -53,7 +57,7 @@ namespace MQTTnet
return new MqttChannelCommunicationAdapter(channel, serializer, logger);
}
public IMqttCommunicationChannel CreateMqttCommunicationChannel(IMqttClientOptions options)
public IMqttCommunicationChannel CreateMqttCommunicationChannel(IMqttClientChannelOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
@@ -66,7 +70,7 @@ namespace MQTTnet
default:
throw new NotSupportedException();
}
}
}
public MqttTcpChannel CreateTcpChannel(MqttClientTcpOptions tcpOptions)
{
@@ -80,7 +84,7 @@ namespace MQTTnet
public MqttPacketSerializer CreateSerializer(MqttProtocolVersion protocolVersion)
{
return new MqttPacketSerializer()
return new MqttPacketSerializer
{
ProtocolVersion = protocolVersion
};
@@ -96,9 +100,9 @@ namespace MQTTnet
return _serviceProvider.GetRequiredService<IMqttClient>();
}
public ManagedMqttClient CreateManagedMqttClient()
public IManagedMqttClient CreateManagedMqttClient()
{
return _serviceProvider.GetRequiredService<ManagedMqttClient>();
return _serviceProvider.GetRequiredService<IManagedMqttClient>();
}
public IMqttServer CreateMqttServer()
Frameworks/MQTTnet.NetStandard/ServiceCollectionExtensions.cs
+2 -1
View File
@@ -52,8 +52,9 @@ namespace MQTTnet
services.AddTransient<IMqttClient, MqttClient>();
services.AddTransient<MqttClient>();
services.AddTransient<IMqttPacketSerializer, MqttPacketSerializer>();
services.AddTransient<IManagedMqttClient, ManagedMqttClient>();
services.AddTransient<ManagedMqttClient>();
services.AddTransient<IMqttPacketSerializer, MqttPacketSerializer>();
services.AddTransient<MqttPacketDispatcher>();
return services;
Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs
+1 -1
View File
@@ -89,7 +89,7 @@ namespace MQTTnet.Implementations
RawReceiveStream = ReceiveStream;
}
private static Certificate LoadCertificate(BaseMqttClientOptions options)
private static Certificate LoadCertificate(MqttClientTcpOptions options)
{
if (options.TlsOptions.Certificates == null || !options.TlsOptions.Certificates.Any())
{
Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs
+6 -3
View File
@@ -32,8 +32,7 @@ namespace MQTTnet.Implementations
}
_webSocket = new ClientWebSocket();
_webSocket.Options.KeepAliveInterval = _options.KeepAlivePeriod;
if (_options.RequestHeaders != null)
{
foreach (var requestHeader in _options.RequestHeaders)
@@ -78,12 +77,16 @@ namespace MQTTnet.Implementations
return;
}
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false);
if (_webSocket.State == WebSocketState.Open || _webSocket.State == WebSocketState.Connecting)
{
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false);
}
}
public void Dispose()
{
_webSocket?.Dispose();
_webSocket = null;
}
}
}
Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs
+4 -4
View File
@@ -42,7 +42,7 @@ namespace MQTTnet
public IMqttCommunicationAdapter CreateClientMqttCommunicationAdapter(IMqttClientOptions options)
{
var logger = _serviceProvider.GetRequiredService<ILogger<MqttChannelCommunicationAdapter>>();
return new MqttChannelCommunicationAdapter(CreateMqttCommunicationChannel(options), CreateSerializer(options.ProtocolVersion), logger);
return new MqttChannelCommunicationAdapter(CreateMqttCommunicationChannel(options.ChannelOptions), CreateSerializer(options.ProtocolVersion), logger);
}
public IMqttCommunicationAdapter CreateServerMqttCommunicationAdapter(IMqttCommunicationChannel channel)
@@ -52,7 +52,7 @@ namespace MQTTnet
return new MqttChannelCommunicationAdapter(channel, serializer, logger);
}
public IMqttCommunicationChannel CreateMqttCommunicationChannel(IMqttClientOptions options)
public IMqttCommunicationChannel CreateMqttCommunicationChannel(IMqttClientChannelOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
@@ -79,7 +79,7 @@ namespace MQTTnet
public MqttPacketSerializer CreateSerializer(MqttProtocolVersion protocolVersion)
{
return new MqttPacketSerializer()
return new MqttPacketSerializer
{
ProtocolVersion = protocolVersion
};
@@ -95,7 +95,7 @@ namespace MQTTnet
return _serviceProvider.GetRequiredService<IMqttClient>();
}
public ManagedMqttClient CreateManagedMqttClient()
public IManagedMqttClient CreateManagedMqttClient()
{
return _serviceProvider.GetRequiredService<ManagedMqttClient>();
}
Frameworks/MQTTnet.UniversalWindows/ServiceCollectionExtensions.cs
+2 -1
View File
@@ -46,8 +46,9 @@ namespace MQTTnet
services.AddTransient<IMqttClient, MqttClient>();
services.AddTransient<MqttClient>();
services.AddTransient<IMqttPacketSerializer, MqttPacketSerializer>();
services.AddTransient<IManagedMqttClient, ManagedMqttClient>();
services.AddTransient<ManagedMqttClient>();
services.AddTransient<IMqttPacketSerializer, MqttPacketSerializer>();
services.AddTransient<MqttPacketDispatcher>();
return services;
MQTTnet.Core/Client/IApplicationMessagePublisher.cs
+10
View File
@@ -0,0 +1,10 @@
using System.Collections.Generic;
using System.Threading.Tasks;
namespace MQTTnet.Core.Client
{
public interface IApplicationMessagePublisher
{
Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages);
}
}
MQTTnet.Core/Client/IMqttClient.cs
+1 -3
View File
@@ -5,7 +5,7 @@ using MQTTnet.Core.Packets;
namespace MQTTnet.Core.Client
{
public interface IMqttClient : IApplicationMessageReceiver
public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher
{
bool IsConnected { get; }
@@ -17,7 +17,5 @@ namespace MQTTnet.Core.Client
Task<IList<MqttSubscribeResult>> SubscribeAsync(IEnumerable<TopicFilter> topicFilters);
Task UnsubscribeAsync(IEnumerable<string> topicFilters);
Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages);
}
}
MQTTnet.Core/Client/IMqttClientChannelOptions.cs
+7
View File
@@ -0,0 +1,7 @@
namespace MQTTnet.Core.Client
{
public interface IMqttClientChannelOptions
{
MqttClientTlsOptions TlsOptions { get; }
}
}
MQTTnet.Core/Client/IMqttClientCredentials.cs
+8
View File
@@ -0,0 +1,8 @@
namespace MQTTnet.Core.Client
{
public interface IMqttClientCredentials
{
string Password { get; }
string Username { get; }
}
}
MQTTnet.Core/Client/IMqttClientFactory.cs
+1 -1
View File
@@ -6,6 +6,6 @@ namespace MQTTnet.Core.Client
{
IMqttClient CreateMqttClient();
ManagedMqttClient CreateManagedMqttClient();
IManagedMqttClient CreateManagedMqttClient();
}
}
MQTTnet.Core/Client/IMqttClientOptions.cs
+8 -7
View File
@@ -5,14 +5,15 @@ namespace MQTTnet.Core.Client
{
public interface IMqttClientOptions
{
bool CleanSession { get; }
string ClientId { get; }
TimeSpan DefaultCommunicationTimeout { get; }
TimeSpan KeepAlivePeriod { get; }
string Password { get; }
MqttProtocolVersion ProtocolVersion { get; }
MqttClientTlsOptions TlsOptions { get; }
string UserName { get; }
IMqttClientCredentials Credentials { get; }
bool CleanSession { get; }
MqttApplicationMessage WillMessage { get; }
TimeSpan CommunicationTimeout { get; }
TimeSpan KeepAlivePeriod { get; }
MqttProtocolVersion ProtocolVersion { get; }
IMqttClientChannelOptions ChannelOptions { get; }
}
}
MQTTnet.Core/Client/MqttApplicationMessageReceivedEventArgs.cs
+4 -1
View File
@@ -4,11 +4,14 @@ namespace MQTTnet.Core.Client
{
public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs
{
public MqttApplicationMessageReceivedEventArgs(MqttApplicationMessage applicationMessage)
public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage)
{
ClientId = clientId;
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
}
public string ClientId { get; }
public MqttApplicationMessage ApplicationMessage { get; }
}
}
MQTTnet.Core/Client/MqttClient.cs
+12 -10
View File
@@ -42,6 +42,7 @@ namespace MQTTnet.Core.Client
public async Task<MqttClientConnectResult> ConnectAsync(IMqttClientOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (options.ChannelOptions == null) throw new ArgumentException("ChannelOptions are not set.");
ThrowIfConnected("It is not allowed to connect with a server after the connection is established.");
@@ -56,7 +57,7 @@ namespace MQTTnet.Core.Client
_scopeHandle = _logger.BeginScope(options.ClientId);
_logger.LogTrace("Trying to connect with server.");
await _adapter.ConnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
await _adapter.ConnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
_logger.LogTrace("Connection with server established.");
await SetupIncomingPacketProcessingAsync();
@@ -73,8 +74,9 @@ namespace MQTTnet.Core.Client
Connected?.Invoke(this, new MqttClientConnectedEventArgs(connectResponse.IsSessionPresent));
return new MqttClientConnectResult(connectResponse.IsSessionPresent);
}
catch (Exception)
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Error while connecting with server.");
await DisconnectInternalAsync().ConfigureAwait(false);
throw;
}
@@ -149,7 +151,7 @@ namespace MQTTnet.Core.Client
case MqttQualityOfServiceLevel.AtMostOnce:
{
// No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, qosPackets);
await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosPackets);
break;
}
case MqttQualityOfServiceLevel.AtLeastOnce:
@@ -186,8 +188,8 @@ namespace MQTTnet.Core.Client
var connectPacket = new MqttConnectPacket
{
ClientId = _options.ClientId,
Username = _options.UserName,
Password = _options.Password,
Username = _options.Credentials?.Username,
Password = _options.Credentials?.Password,
CleanSession = _options.CleanSession,
KeepAlivePeriod = (ushort)_options.KeepAlivePeriod.TotalSeconds,
WillMessage = willApplicationMessage
@@ -247,7 +249,7 @@ namespace MQTTnet.Core.Client
try
{
await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
await _adapter.DisconnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
_logger.LogInformation("Disconnected from adapter.");
}
catch (Exception exception)
@@ -303,7 +305,7 @@ namespace MQTTnet.Core.Client
try
{
var applicationMessage = publishPacket.ToApplicationMessage();
ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(applicationMessage));
ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage));
}
catch (Exception exception)
{
@@ -354,13 +356,13 @@ namespace MQTTnet.Core.Client
private Task SendAsync(MqttBasePacket packet)
{
return _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, packet);
return _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, packet);
}
private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket
{
var packetAwaiter = _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.DefaultCommunicationTimeout);
await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, requestPacket).ConfigureAwait(false);
var packetAwaiter = _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.CommunicationTimeout);
await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, requestPacket).ConfigureAwait(false);
return (TResponsePacket)await packetAwaiter.ConfigureAwait(false);
}
MQTTnet.Core/Client/MqttClientCredentials.cs
+9
View File
@@ -0,0 +1,9 @@
namespace MQTTnet.Core.Client
{
public class MqttClientCredentials : IMqttClientCredentials
{
public string Username { get; set; }
public string Password { get; set; }
}
}
MQTTnet.Core/Client/MqttClientExtensions.cs
+1 -2
View File
@@ -2,13 +2,12 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using MQTTnet.Core.Packets;
namespace MQTTnet.Core.Client
{
public static class MqttClientExtensions
{
public static Task PublishAsync(this IMqttClient client, params MqttApplicationMessage[] applicationMessages)
public static Task PublishAsync(this IApplicationMessagePublisher client, params MqttApplicationMessage[] applicationMessages)
{
if (client == null) throw new ArgumentNullException(nameof(client));
if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
MQTTnet.Core/Client/BaseMqttClientOptions.cs → MQTTnet.Core/Client/MqttClientOptions.cs
+6 -8
View File
@@ -3,24 +3,22 @@ using MQTTnet.Core.Serializer;
namespace MQTTnet.Core.Client
{
public abstract class BaseMqttClientOptions : IMqttClientOptions
public class MqttClientOptions : IMqttClientOptions
{
public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
public MqttApplicationMessage WillMessage { get; set; }
public string UserName { get; set; }
public string Password { get; set; }
public string ClientId { get; set; } = Guid.NewGuid().ToString("N");
public bool CleanSession { get; set; } = true;
public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials();
public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(5);
public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10);
public TimeSpan CommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10);
public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
public IMqttClientChannelOptions ChannelOptions { get; set; }
}
}
MQTTnet.Core/Client/MqttClientOptionsBuilder.cs
+116
View File
@@ -0,0 +1,116 @@
using System;
using MQTTnet.Core.Serializer;
namespace MQTTnet.Core.Client
{
public class MqttClientOptionsBuilder
{
private readonly MqttClientOptions _options = new MqttClientOptions();
private MqttClientTcpOptions _tcpOptions;
private MqttClientWebSocketOptions _webSocketOptions;
private MqttClientTlsOptions _tlsOptions;
public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion protocolVersion)
{
_options.ProtocolVersion = protocolVersion;
return this;
}
public MqttClientOptionsBuilder WithCommunicationTimeout(TimeSpan communicationTimeout)
{
_options.CommunicationTimeout = communicationTimeout;
return this;
}
public MqttClientOptionsBuilder WithCleanSession(bool value = true)
{
_options.CleanSession = value;
return this;
}
public MqttClientOptionsBuilder WithKeepAlivePeriod(TimeSpan keepAlivePeriod)
{
_options.KeepAlivePeriod = keepAlivePeriod;
return this;
}
public MqttClientOptionsBuilder WithClientId(string clientId)
{
_options.ClientId = clientId;
return this;
}
public MqttClientOptionsBuilder WithWillMessage(MqttApplicationMessage applicationMessage)
{
_options.WillMessage = applicationMessage;
return this;
}
public MqttClientOptionsBuilder WithCredentials(string username, string password = null)
{
_options.Credentials = new MqttClientCredentials
{
Username = username,
Password = password
};
return this;
}
public MqttClientOptionsBuilder WithTcpServer(string server, int? port = null)
{
_tcpOptions = new MqttClientTcpOptions
{
Server = server,
Port = port
};
return this;
}
public MqttClientOptionsBuilder WithWebSocketServer(string uri)
{
_webSocketOptions = new MqttClientWebSocketOptions
{
Uri = uri
};
return this;
}
public MqttClientOptionsBuilder WithTls()
{
_tlsOptions = new MqttClientTlsOptions
{
UseTls = true
};
return this;
}
public IMqttClientOptions Build()
{
if (_tlsOptions != null)
{
if (_tcpOptions == null && _webSocketOptions == null)
{
throw new InvalidOperationException("A channel (TCP or WebSocket) must be set if TLS is configured.");
}
if (_tcpOptions != null)
{
_options.ChannelOptions = _tcpOptions;
}
else
{
_options.ChannelOptions = _webSocketOptions;
}
}
_options.ChannelOptions = (IMqttClientChannelOptions)_tcpOptions ?? _webSocketOptions;
return _options;
}
}
}
MQTTnet.Core/Client/MqttClientTcpOptions.cs
+3 -1
View File
@@ -1,9 +1,11 @@
namespace MQTTnet.Core.Client
{
public class MqttClientTcpOptions : BaseMqttClientOptions
public class MqttClientTcpOptions : IMqttClientChannelOptions
{
public string Server { get; set; }
public int? Port { get; set; }
public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
}
}
MQTTnet.Core/Client/MqttClientTlsOptions.cs
+1 -1
View File
@@ -2,7 +2,7 @@
namespace MQTTnet.Core.Client
{
public sealed class MqttClientTlsOptions
public class MqttClientTlsOptions
{
public bool UseTls { get; set; }
MQTTnet.Core/Client/MqttClientWebSocketOptions.cs
+3 -1
View File
@@ -3,7 +3,7 @@ using System.Net;
namespace MQTTnet.Core.Client
{
public class MqttClientWebSocketOptions : BaseMqttClientOptions
public class MqttClientWebSocketOptions : IMqttClientChannelOptions
{
public string Uri { get; set; }
@@ -12,5 +12,7 @@ namespace MQTTnet.Core.Client
public ICollection<string> SubProtocols { get; set; }
public CookieContainer CookieContainer { get; set; }
public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
}
}
MQTTnet.Core/ManagedClient/IManagedMqttClient.cs
+21
View File
@@ -0,0 +1,21 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using MQTTnet.Core.Client;
namespace MQTTnet.Core.ManagedClient
{
public interface IManagedMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher
{
bool IsConnected { get; }
event EventHandler<MqttClientConnectedEventArgs> Connected;
event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
Task StartAsync(IManagedMqttClientOptions options);
Task StopAsync();
Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters);
Task UnsubscribeAsync(IEnumerable<TopicFilter> topicFilters);
}
}
MQTTnet.Core/ManagedClient/IManagedMqttClientOptions.cs
-2
View File
@@ -10,7 +10,5 @@ namespace MQTTnet.Core.ManagedClient
TimeSpan AutoReconnectDelay { get; }
IManagedMqttClientStorage Storage { get; }
Func<IManagedMqttClientOptions, string> PasswordProvider { get; }
}
}
MQTTnet.Core/ManagedClient/ManagedMqttClient.cs
+2 -4
View File
@@ -6,13 +6,12 @@ using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Client;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
using Microsoft.Extensions.Logging;
namespace MQTTnet.Core.ManagedClient
{
public class ManagedMqttClient : IApplicationMessageReceiver
public class ManagedMqttClient : IManagedMqttClient
{
private readonly ManagedMqttClientStorageManager _storageManager = new ManagedMqttClientStorageManager();
private readonly BlockingCollection<MqttApplicationMessage> _messageQueue = new BlockingCollection<MqttApplicationMessage>();
@@ -92,7 +91,7 @@ namespace MQTTnet.Core.ManagedClient
return Task.FromResult(0);
}
public Task EnqueueAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
public Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
{
if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
@@ -277,7 +276,6 @@ namespace MQTTnet.Core.ManagedClient
try
{
_options.PasswordProvider?.Invoke(_options);
await _mqttClient.ConnectAsync(_options.ClientOptions).ConfigureAwait(false);
return ReconnectionResult.Reconnected;
}
MQTTnet.Core/ManagedClient/ManagedMqttClientOptions.cs
-3
View File
@@ -10,8 +10,5 @@ namespace MQTTnet.Core.ManagedClient
public TimeSpan AutoReconnectDelay { get; set; } = TimeSpan.FromSeconds(5);
public IManagedMqttClientStorage Storage { get; set; }
public Func<IManagedMqttClientOptions, string> PasswordProvider { get; set; }
}
}
MQTTnet.Core/Server/IMqttServer.cs
+2 -4
View File
@@ -1,18 +1,16 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using MQTTnet.Core.Client;
namespace MQTTnet.Core.Server
{
public interface IMqttServer
public interface IMqttServer : IApplicationMessageReceiver, IApplicationMessagePublisher
{
event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
IList<ConnectedMqttClient> GetConnectedClients();
void Publish(MqttApplicationMessage applicationMessage);
Task StartAsync();
Task StopAsync();
MQTTnet.Core/Server/MqttApplicationMessageReceivedEventArgs.cs
-17
View File
@@ -1,17 +0,0 @@
using System;
namespace MQTTnet.Core.Server
{
public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs
{
public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage)
{
ClientId = clientId;
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
}
public string ClientId { get; }
public MqttApplicationMessage ApplicationMessage { get; }
}
}
MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs
+2 -2
View File
@@ -28,7 +28,7 @@ namespace MQTTnet.Core.Server
{
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
Task.Run(() => SendPendingPublishPacketsAsync(adapter, cancellationToken), cancellationToken);
Task.Factory.StartNew(async () => await SendPendingPublishPacketsAsync(adapter, cancellationToken), cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default).ConfigureAwait(false);
}
public void Enqueue(MqttPublishPacket publishPacket)
@@ -74,7 +74,7 @@ namespace MQTTnet.Core.Server
{
_logger.LogWarning(new EventId(), exception, "Sending publish packet failed due to communication exception.");
}
if (exception is OperationCanceledException)
else if (exception is OperationCanceledException)
{
}
else
MQTTnet.Core/Server/MqttClientSession.cs
+18 -14
View File
@@ -28,11 +28,11 @@ namespace MQTTnet.Core.Server
public MqttClientSession(string clientId, MqttClientSessionsManager mqttClientSessionsManager, ILogger<MqttClientSession> logger, ILogger<MqttClientPendingMessagesQueue> msgQueueLogger)
{
ClientId = clientId;
_options = mqttClientSessionsManager.Options;
_mqttClientSessionsManager = mqttClientSessionsManager ?? throw new ArgumentNullException(nameof(mqttClientSessionsManager));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
ClientId = clientId;
_options = mqttClientSessionsManager.Options;
_pendingMessagesQueue = new MqttClientPendingMessagesQueue(mqttClientSessionsManager.Options, this, msgQueueLogger);
}
@@ -46,10 +46,9 @@ namespace MQTTnet.Core.Server
{
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
_willMessage = willMessage;
try
{
_willMessage = willMessage;
_adapter = adapter;
_cancellationTokenSource = new CancellationTokenSource();
@@ -71,18 +70,23 @@ namespace MQTTnet.Core.Server
public void Stop()
{
if (_willMessage != null)
try
{
_mqttClientSessionsManager.DispatchApplicationMessage(this, _willMessage);
_cancellationTokenSource?.Cancel(false);
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
_adapter = null;
_logger.LogInformation("Client '{0}': Disconnected.", ClientId);
}
finally
{
if (_willMessage != null)
{
_mqttClientSessionsManager.DispatchApplicationMessage(this, _willMessage);
}
}
_cancellationTokenSource?.Cancel(false);
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
_adapter = null;
_logger.LogInformation("Client '{0}': Disconnected.", ClientId);
}
public void EnqueuePublishPacket(MqttPublishPacket publishPacket)
MQTTnet.Core/Server/MqttClientSessionsManager.cs
+3 -1
View File
@@ -11,6 +11,7 @@ using MQTTnet.Core.Protocol;
using MQTTnet.Core.Serializer;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using MQTTnet.Core.Client;
namespace MQTTnet.Core.Server
{
@@ -28,9 +29,9 @@ namespace MQTTnet.Core.Server
_mqttClientSesssionFactory = mqttClientSesssionFactory ?? throw new ArgumentNullException(nameof(mqttClientSesssionFactory));
}
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public MqttClientRetainedMessagesManager RetainedMessagesManager { get; }
public MqttServerOptions Options { get; }
@@ -166,6 +167,7 @@ namespace MQTTnet.Core.Server
_clientSessions.Remove(connectPacket.ClientId);
clientSession.Dispose();
clientSession = null;
_logger.LogTrace("Disposed existing session of client '{0}'.", connectPacket.ClientId);
}
else
MQTTnet.Core/Server/MqttServer.cs
+16 -4
View File
@@ -6,6 +6,7 @@ using MQTTnet.Core.Adapter;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System.Linq;
using MQTTnet.Core.Client;
namespace MQTTnet.Core.Server
{
@@ -28,6 +29,7 @@ namespace MQTTnet.Core.Server
{
throw new ArgumentNullException(nameof(adapters));
}
_adapters = adapters.ToList();
_clientSessionsManager.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
@@ -44,12 +46,22 @@ namespace MQTTnet.Core.Server
public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public void Publish(MqttApplicationMessage applicationMessage)
public Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
_options.ApplicationMessageInterceptor?.Invoke(applicationMessage);
_clientSessionsManager.DispatchApplicationMessage(null, applicationMessage);
if (_cancellationTokenSource == null)
{
throw new InvalidOperationException("The server is not started.");
}
foreach (var applicationMessage in applicationMessages)
{
_options.ApplicationMessageInterceptor?.Invoke(applicationMessage);
_clientSessionsManager.DispatchApplicationMessage(null, applicationMessage);
}
return Task.FromResult(0);
}
public async Task StartAsync()
MQTTnet.Core/Server/MqttServerOptions.cs
+1 -1
View File
@@ -4,7 +4,7 @@ using MQTTnet.Core.Protocol;
namespace MQTTnet.Core.Server
{
public sealed class MqttServerOptions
public class MqttServerOptions
{
public MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; } = new MqttServerDefaultEndpointOptions();
README.md
+1 -1
View File
@@ -18,7 +18,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
* Performance optimized (processing ~27.000 messages / second)*
* Interfaces included for mocking and testing
* Access to internal trace messages
* Unit tested (67+ tests)
* Unit tested (70+ tests)
\* Tested on local machine with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetFramework_.
Tests/MQTTnet.Core.Tests/MqttServerTests.cs
+27 -53
View File
@@ -5,7 +5,6 @@ using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Client;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
using MQTTnet.Core.Server;
using Microsoft.Extensions.DependencyInjection;
@@ -52,12 +51,7 @@ namespace MQTTnet.Core.Tests
public async Task MqttServer_WillMessage()
{
var serverAdapter = new TestMqttServerAdapter();
var services = new ServiceCollection()
.AddMqttServer()
.AddSingleton<IMqttServerAdapter>(serverAdapter)
.BuildServiceProvider();
var s = services.GetRequiredService<IMqttServer>();
var s = new MqttFactory().CreateMqttServer();
await s.StartAsync();
var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build();
@@ -81,12 +75,7 @@ namespace MQTTnet.Core.Tests
public async Task MqttServer_Unsubscribe()
{
var serverAdapter = new TestMqttServerAdapter();
var services = new ServiceCollection()
.AddMqttServer()
.AddSingleton<IMqttServerAdapter>(serverAdapter)
.BuildServiceProvider();
var s = services.GetRequiredService<IMqttServer>();
var s = new MqttFactory().CreateMqttServer();
await s.StartAsync();
var c1 = await serverAdapter.ConnectTestClient(s, "c1");
@@ -122,12 +111,7 @@ namespace MQTTnet.Core.Tests
public async Task MqttServer_Publish()
{
var serverAdapter = new TestMqttServerAdapter();
var services = new ServiceCollection()
.AddMqttServer()
.AddSingleton<IMqttServerAdapter>(serverAdapter)
.BuildServiceProvider();
var s = services.GetRequiredService<IMqttServer>();
var s = new MqttFactory().CreateMqttServer();
await s.StartAsync();
var c1 = await serverAdapter.ConnectTestClient(s, "c1");
@@ -138,7 +122,7 @@ namespace MQTTnet.Core.Tests
var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build();
await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce));
s.Publish(message);
s.PublishAsync(message).Wait();
await Task.Delay(500);
await s.StopAsync();
@@ -150,22 +134,17 @@ namespace MQTTnet.Core.Tests
public async Task MqttServer_NoRetainedMessage()
{
var serverAdapter = new TestMqttServerAdapter();
var services = new ServiceCollection()
.AddMqttServer()
.AddSingleton<IMqttServerAdapter>(serverAdapter)
.BuildServiceProvider();
var s = services.GetRequiredService<IMqttServer>();
var s = new MqttFactory().CreateMqttServer();
await s.StartAsync();
var c1 = await serverAdapter.ConnectTestClient(s, "c1");
await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, false));
await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).Build());
await c1.DisconnectAsync();
var c2 = await serverAdapter.ConnectTestClient(s, "c2");
var receivedMessagesCount = 0;
c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce));
await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce));
await Task.Delay(500);
@@ -178,22 +157,17 @@ namespace MQTTnet.Core.Tests
public async Task MqttServer_RetainedMessage()
{
var serverAdapter = new TestMqttServerAdapter();
var services = new ServiceCollection()
.AddMqttServer()
.AddSingleton<IMqttServerAdapter>(serverAdapter)
.BuildServiceProvider();
var s = services.GetRequiredService<IMqttServer>();
var s = new MqttFactory().CreateMqttServer();
await s.StartAsync();
var c1 = await serverAdapter.ConnectTestClient(s, "c1");
await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, true));
await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
await c1.DisconnectAsync();
var c2 = await serverAdapter.ConnectTestClient(s, "c2");
var receivedMessagesCount = 0;
c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce));
await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce));
await Task.Delay(500);
@@ -207,22 +181,23 @@ namespace MQTTnet.Core.Tests
{
var serverAdapter = new TestMqttServerAdapter();
var services = new ServiceCollection()
.AddMqttServer()
.AddLogging()
.AddMqttServer() // TODO: Is there maybe an easier way for the library user to set the options?
.AddSingleton<IMqttServerAdapter>(serverAdapter)
.BuildServiceProvider();
var s = services.GetRequiredService<IMqttServer>();
var s = new MqttFactory(services).CreateMqttServer();
await s.StartAsync();
var c1 = await serverAdapter.ConnectTestClient(s, "c1");
await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, true));
await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[0], MqttQualityOfServiceLevel.AtLeastOnce, true));
await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build());
await c1.DisconnectAsync();
var c2 = await serverAdapter.ConnectTestClient(s, "c2");
var receivedMessagesCount = 0;
c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce));
await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce));
await Task.Delay(500);
@@ -238,15 +213,16 @@ namespace MQTTnet.Core.Tests
var serverAdapter = new TestMqttServerAdapter();
var services = new ServiceCollection()
.AddMqttServer()
.AddLogging()
.AddMqttServer(options => options.Storage = storage) // TODO: Is there maybe an easier way for the library user to set the options?
.AddSingleton<IMqttServerAdapter>(serverAdapter)
.BuildServiceProvider();
var s = services.GetRequiredService<IMqttServer>();
var s = new MqttFactory(services).CreateMqttServer(); // TODO: Like here?
await s.StartAsync();
var c1 = await serverAdapter.ConnectTestClient(s, "c1");
await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, true));
await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
await c1.DisconnectAsync();
await s.StopAsync();
@@ -257,7 +233,7 @@ namespace MQTTnet.Core.Tests
var c2 = await serverAdapter.ConnectTestClient(s, "c2");
var receivedMessagesCount = 0;
c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce));
await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce));
await Task.Delay(500);
@@ -269,18 +245,16 @@ namespace MQTTnet.Core.Tests
[TestMethod]
public async Task MqttServer_InterceptMessage()
{
var options = new MqttServerOptions
MqttApplicationMessage Interceptor(MqttApplicationMessage message)
{
ApplicationMessageInterceptor = message =>
{
message.Payload = Encoding.ASCII.GetBytes("extended");
return message;
}
};
message.Payload = Encoding.ASCII.GetBytes("extended");
return message;
}
var serverAdapter = new TestMqttServerAdapter();
var services = new ServiceCollection()
.AddMqttServer()
.AddLogging()
.AddMqttServer(options => options.ApplicationMessageInterceptor = Interceptor)
.AddSingleton<IMqttServerAdapter>(serverAdapter)
.BuildServiceProvider();
Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs
+11 -1
View File
@@ -21,7 +21,17 @@ namespace MQTTnet.Core.Tests
var connected = WaitForClientToConnect(server, clientId);
FireClientAcceptedEvent(adapterB);
await client.ConnectAsync(new MqttClientTcpOptions { ClientId = clientId, WillMessage = willMessage });
var options = new MqttClientOptions
{
ClientId = clientId,
WillMessage = willMessage,
ChannelOptions = new MqttClientTcpOptions()
};
options.ChannelOptions = new MqttClientTcpOptions();
await client.ConnectAsync(options);
await connected;
return client;
Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs
+2 -1
View File
@@ -8,6 +8,7 @@ using Microsoft.Extensions.FileProviders;
using Microsoft.Extensions.Logging;
using MQTTnet.AspnetCore;
using MQTTnet.Core;
using MQTTnet.Core.Client;
namespace MQTTnet.TestApp.AspNetCore2
{
@@ -29,7 +30,7 @@ namespace MQTTnet.TestApp.AspNetCore2
while (true)
{
server.Publish(msg.Build());
server.PublishAsync(msg.Build()).Wait();
await Task.Delay(TimeSpan.FromSeconds(2));
msg.WithPayload("Mqtt is still awesome at " + DateTime.Now);
}
Tests/MQTTnet.TestApp.NetCore/ClientTest.cs
+99
View File
@@ -0,0 +1,99 @@
using System;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using MQTTnet.Core;
using MQTTnet.Core.Client;
namespace MQTTnet.TestApp.NetCore
{
public static class ClientTest
{
public static async Task RunAsync()
{
try
{
var options = new MqttClientOptions
{
ClientId = "XYZ",
CleanSession = true,
ChannelOptions = new MqttClientTcpOptions
{
Server = "localhost"
},
////ChannelOptions = new MqttClientWebSocketOptions
////{
//// Uri = "localhost"
////}
};
var factory = new MqttFactory();
factory.GetLoggerFactory().AddConsole();
var client = factory.CreateMqttClient();
client.ApplicationMessageReceived += (s, e) =>
{
Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###");
Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}");
Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}");
Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}");
Console.WriteLine();
};
client.Connected += async (s, e) =>
{
Console.WriteLine("### CONNECTED WITH SERVER ###");
await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("#").Build());
Console.WriteLine("### SUBSCRIBED ###");
};
client.Disconnected += async (s, e) =>
{
Console.WriteLine("### DISCONNECTED FROM SERVER ###");
await Task.Delay(TimeSpan.FromSeconds(5));
try
{
await client.ConnectAsync(options);
}
catch
{
Console.WriteLine("### RECONNECTING FAILED ###");
}
};
try
{
await client.ConnectAsync(options);
}
catch (Exception exception)
{
Console.WriteLine("### CONNECTING FAILED ###" + Environment.NewLine + exception);
}
Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###");
while (true)
{
Console.ReadLine();
var applicationMessage = new MqttApplicationMessageBuilder()
.WithTopic("A/B/C")
.WithPayload("Hello World")
.WithAtLeastOnceQoS()
.Build();
await client.PublishAsync(applicationMessage);
}
}
catch (Exception exception)
{
Console.WriteLine(exception);
}
}
}
}
Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs
+19 -21
View File
@@ -3,7 +3,6 @@ using System.Threading.Tasks;
using MQTTnet.Core;
using MQTTnet.Core.Client;
using MQTTnet.Core.ManagedClient;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
@@ -25,32 +24,24 @@ namespace MQTTnet.TestApp.NetCore
services.GetService<ILoggerFactory>()
.AddConsole();
ClientRetainedMessageHandler ms = new ClientRetainedMessageHandler();
Func<ManagedMqttClientOptions, string> func = delegate (ManagedMqttClientOptions managedMqttClientOptions)
{
return "password";
};
var ms = new ClientRetainedMessageHandler();
var options = new ManagedMqttClientOptions
{
ClientOptions = new MqttClientTcpOptions
ClientOptions = new MqttClientOptions
{
Server = "broker.hivemq.com",
ClientId = "MQTTnetManagedClientTest",
Password = "pippo",
Credentials = new RandomPassword(),
ChannelOptions = new MqttClientTcpOptions
{
Server = "broker.hivemq.com"
}
},
AutoReconnectDelay = TimeSpan.FromSeconds(1),
Storage = ms,
PasswordProvider = o =>
{
//o.ClientOptions.Password = GetPassword();
return o.ClientOptions.Password;
}
Storage = ms
};
try
{
var managedClient = services.GetRequiredService<ManagedMqttClient>();
@@ -78,11 +69,18 @@ namespace MQTTnet.TestApp.NetCore
}
public static string GetPassword()
public class RandomPassword : IMqttClientCredentials
{
return "password";
}
public string Password
{
get
{
return Guid.NewGuid().ToString(); // The random password.
}
}
public string Username => "the_static_user";
}
public class ClientRetainedMessageHandler : IManagedMqttClientStorage
{
Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs
+8 -9
View File
@@ -2,7 +2,6 @@
using Microsoft.Extensions.Logging;
using MQTTnet.Core;
using MQTTnet.Core.Client;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
using MQTTnet.Core.Server;
using System;
@@ -20,7 +19,8 @@ namespace MQTTnet.TestApp.NetCore
public static async Task RunAsync()
{
var services = new ServiceCollection()
.AddMqttServer(options => {
.AddMqttServer(options =>
{
options.ConnectionValidator = p =>
{
@@ -41,11 +41,10 @@ namespace MQTTnet.TestApp.NetCore
.AddLogging()
.BuildServiceProvider();
services.GetService<ILoggerFactory>()
.AddConsole(minLevel: LogLevel.Warning, includeScopes: true);
services.GetService<ILoggerFactory>().AddConsole(LogLevel.Warning, true);
Console.WriteLine("Press 'c' for concurrent sends. Otherwise in one batch.");
var concurrent = Console.ReadKey(intercept: true).KeyChar == 'c';
var concurrent = Console.ReadKey(true).KeyChar == 'c';
var server = Task.Factory.StartNew(() => RunServerAsync(services), TaskCreationOptions.LongRunning);
var client = Task.Factory.StartNew(() => RunClientAsync(2000, TimeSpan.FromMilliseconds(10), services, concurrent), TaskCreationOptions.LongRunning);
@@ -62,12 +61,12 @@ namespace MQTTnet.TestApp.NetCore
{
try
{
var options = new MqttClientTcpOptions
var options = new MqttClientOptions
{
Server = "localhost",
ChannelOptions = new MqttClientTcpOptions { Server = "localhost" },
ClientId = "Client1",
CleanSession = true,
DefaultCommunicationTimeout = TimeSpan.FromMinutes(10)
CommunicationTimeout = TimeSpan.FromMinutes(10)
};
var client = serviceProvider.GetRequiredService<IMqttClient>();
Tests/MQTTnet.TestApp.NetCore/Program.cs
+17 -104
View File
@@ -1,17 +1,12 @@
using MQTTnet.Core;
using MQTTnet.Core.Client;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
using MQTTnet.Core.Server;
using System;
using System.Collections.Generic;
using System.IO;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Newtonsoft.Json;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
namespace MQTTnet.TestApp.NetCore
@@ -25,11 +20,11 @@ namespace MQTTnet.TestApp.NetCore
Console.WriteLine("2 = Start server");
Console.WriteLine("3 = Start performance test");
Console.WriteLine("4 = Start managed client");
var pressedKey = Console.ReadKey(true);
if (pressedKey.KeyChar == '1')
{
Task.Run(RunClientAsync);
Task.Run(ClientTest.RunAsync);
}
else if (pressedKey.KeyChar == '2')
{
@@ -46,105 +41,22 @@ namespace MQTTnet.TestApp.NetCore
Thread.Sleep(Timeout.Infinite);
}
private static async Task RunClientAsync()
{
try
{
var services = new ServiceCollection()
.AddMqttServer()
.AddLogging()
.BuildServiceProvider();
services.GetService<ILoggerFactory>()
.AddConsole();
var options = new MqttClientWebSocketOptions
{
Uri = "localhost",
ClientId = "XYZ",
CleanSession = true
};
var client = services.GetRequiredService<IMqttClient>();
client.ApplicationMessageReceived += (s, e) =>
{
Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###");
Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}");
Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}");
Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}");
Console.WriteLine();
};
client.Connected += async (s, e) =>
{
Console.WriteLine("### CONNECTED WITH SERVER ###");
await client.SubscribeAsync(new List<TopicFilter>
{
new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce)
});
Console.WriteLine("### SUBSCRIBED ###");
};
client.Disconnected += async (s, e) =>
{
Console.WriteLine("### DISCONNECTED FROM SERVER ###");
await Task.Delay(TimeSpan.FromSeconds(5));
try
{
await client.ConnectAsync(options);
}
catch
{
Console.WriteLine("### RECONNECTING FAILED ###");
}
};
try
{
await client.ConnectAsync(options);
}
catch (Exception exception)
{
Console.WriteLine("### CONNECTING FAILED ###" + Environment.NewLine + exception);
}
Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###");
while (true)
{
Console.ReadLine();
var applicationMessage = new MqttApplicationMessage
{
Topic = "A/B/C",
Payload = Encoding.UTF8.GetBytes("Hello World"),
QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce
};
await client.PublishAsync(applicationMessage);
}
}
catch (Exception exception)
{
Console.WriteLine(exception);
}
}
// This code is used at the Wiki on GitHub!
// ReSharper disable once UnusedMember.Local
private static async void WikiCode()
{
{
var serviceProvider = new ServiceCollection()
.AddMqttServer()
.AddLogging()
.BuildServiceProvider();
var client = new MqttFactory().CreateMqttClient();
var client = serviceProvider.GetRequiredService<IMqttClient>();
var options = new MqttClientOptionsBuilder()
.WithClientId("Client1")
.WithTcpServer("broker.hivemq.com")
.WithCredentials("bud", "%spencer%")
.WithTls()
.Build();
await client.ConnectAsync(options);
var message = new MqttApplicationMessageBuilder()
.WithTopic("MyTopic")
@@ -157,9 +69,10 @@ namespace MQTTnet.TestApp.NetCore
}
{
var message = new MqttApplicationMessageBuilder()
.WithTopic("/MQTTnet/is/awesome")
.Build();
var factory = new MqttFactory();
factory.GetLoggerFactory().AddConsole();
var client = factory.CreateMqttClient();
}
}
}
Tests/MQTTnet.TestApp.NetCore/ServerTest.cs
+2 -1
View File
@@ -49,6 +49,7 @@ namespace MQTTnet.TestApp.NetCore
});
var serviceProvider = services.BuildServiceProvider();
serviceProvider.GetRequiredService<ILoggerFactory>().AddConsole();
//var certificate = new X509Certificate(@"C:\certs\test\test.cer", "");
//options.TlsEndpointOptions.Certificate = certificate.Export(X509ContentType.Cert);
@@ -56,7 +57,7 @@ namespace MQTTnet.TestApp.NetCore
//options.DefaultEndpointOptions.IsEnabled = true;
//options.TlsEndpointOptions.IsEnabled = false;
var mqttServer = serviceProvider.GetRequiredService<IMqttServer>();
var mqttServer = new MqttFactory(serviceProvider).CreateMqttServer();
mqttServer.ClientDisconnected += (s, e) =>
{
Console.Write("Client disconnected event fired.");
Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs
+100 -72
View File
@@ -8,9 +8,7 @@ using Microsoft.Extensions.DependencyInjection;
using MQTTnet.Core;
using MQTTnet.Core.Client;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
using MQTTnet.Core.Server;
using MQTTnet.Implementations;
namespace MQTTnet.TestApp.UniversalWindows
@@ -42,35 +40,44 @@ namespace MQTTnet.TestApp.UniversalWindows
private async void Connect(object sender, RoutedEventArgs e)
{
BaseMqttClientOptions options = null;
var tlsOptions = new MqttClientTlsOptions
{
UseTls = UseTls.IsChecked == true,
IgnoreCertificateChainErrors = true,
IgnoreCertificateRevocationErrors = true,
AllowUntrustedCertificates = true
};
var options = new MqttClientOptions { ClientId = ClientId.Text };
if (UseTcp.IsChecked == true)
{
options = new MqttClientTcpOptions
options.ChannelOptions = new MqttClientTcpOptions
{
Server = Server.Text
Server = Server.Text,
TlsOptions = tlsOptions
};
}
if (UseWs.IsChecked == true)
{
options = new MqttClientWebSocketOptions
options.ChannelOptions = new MqttClientWebSocketOptions
{
Uri = Server.Text
Uri = Server.Text,
TlsOptions = tlsOptions
};
}
if (options == null)
if (options.ChannelOptions == null)
{
throw new InvalidOperationException();
}
options.UserName = User.Text;
options.Password = Password.Text;
options.ClientId = ClientId.Text;
options.TlsOptions.UseTls = UseTls.IsChecked == true;
options.TlsOptions.IgnoreCertificateChainErrors = true;
options.TlsOptions.IgnoreCertificateRevocationErrors = true;
options.TlsOptions.AllowUntrustedCertificates = true;
options.Credentials = new MqttClientCredentials
{
Username = User.Text,
Password = Password.Text
};
try
{
@@ -201,73 +208,94 @@ namespace MQTTnet.TestApp.UniversalWindows
// ReSharper disable once UnusedMember.Local
private async Task WikiCode()
{
var services = new ServiceCollection()
.AddMqttClient()
.AddMqttServer(options =>
{
options.ConnectionValidator = c =>
{
if (c.ClientId.Length < 10)
{
return MqttConnectReturnCode.ConnectionRefusedIdentifierRejected;
}
if (c.Username != "mySecretUser")
{
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
}
if (c.Password != "mySecretPassword")
{
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
}
return MqttConnectReturnCode.ConnectionAccepted;
};
})
.BuildServiceProvider();
var factory = new MqttFactory(services);
var mqttClient = factory.CreateMqttClient();
// ----------------------------------
var tcpOptions = new MqttClientTcpOptions
{
Server = "broker.hivemq.org",
ClientId = "TestClient"
};
// Create a new MQTT client
var services = new ServiceCollection()
.AddMqttClient()
.BuildServiceProvider();
await mqttClient.ConnectAsync(tcpOptions);
var factory = new MqttFactory(services);
var client = factory.CreateMqttClient();
// ----------------------------------
var secureTcpOptions = new MqttClientTcpOptions
{
Server = "broker.hivemq.org",
ClientId = "TestClient",
TlsOptions = new MqttClientTlsOptions
{
UseTls = true,
IgnoreCertificateChainErrors = true,
IgnoreCertificateRevocationErrors = true,
AllowUntrustedCertificates = true
// Create TCP based options using the builder
var options = new MqttClientOptionsBuilder()
.WithClientId("Client1")
.WithTcpServer("broker.hivemq.com")
.WithCredentials("bud", "%spencer%")
.WithTls()
.Build();
await client.ConnectAsync(options);
}
};
// ----------------------------------
{
// Create TCP based options manually
var options = new MqttClientOptions
{
ClientId = "Client1",
Credentials = new MqttClientCredentials
{
Username = "bud",
Password = "%spencer%"
},
ChannelOptions = new MqttClientTcpOptions
{
Server = "broker.hivemq.org",
TlsOptions = new MqttClientTlsOptions
{
UseTls = true
}
},
};
}
var wsOptions = new MqttClientWebSocketOptions
{
Uri = "broker.hivemq.com:8000/mqtt",
ClientId = "TestClient"
};
{
// Subscribe to a topic
await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build());
await mqttClient.ConnectAsync(wsOptions);
// Unsubscribe from a topic
await client.UnsubscribeAsync("my/topic");
// Publish an application message
var applicationMessage = new MqttApplicationMessageBuilder()
.WithTopic("A/B/C")
.WithPayload("Hello World")
.WithAtLeastOnceQoS()
.Build();
await client.PublishAsync(applicationMessage);
}
}
// ----------------------------------
{
var services = new ServiceCollection()
.AddMqttServer(options =>
{
options.ConnectionValidator = c =>
{
if (c.ClientId.Length < 10)
{
return MqttConnectReturnCode.ConnectionRefusedIdentifierRejected;
}
if (c.Username != "mySecretUser")
{
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
}
if (c.Password != "mySecretPassword")
{
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
}
return MqttConnectReturnCode.ConnectionAccepted;
};
})
.BuildServiceProvider();
var factory = new MqttFactory(services);
var mqttServer = factory.CreateMqttServer();
await mqttServer.StartAsync();
@@ -278,7 +306,7 @@ namespace MQTTnet.TestApp.UniversalWindows
}
// ----------------------------------
// For UWP apps:
MqttTcpChannel.CustomIgnorableServerCertificateErrorsResolver = o =>
{