diff --git a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs index 87317a73..a62ecded 100644 --- a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs @@ -1,7 +1,6 @@ using System; using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.DependencyInjection; -using MQTTnet.AspNetCore; using MQTTnet.Core.Server; namespace MQTTnet.AspnetCore diff --git a/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs index 5cd5d555..abb30e37 100644 --- a/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs +++ b/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs @@ -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 { diff --git a/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs index b0297b21..1951368c 100644 --- a/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs @@ -1,6 +1,5 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; -using MQTTnet.AspNetCore; using MQTTnet.Core.Adapter; using MQTTnet.Core.Server; diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs index 7c051bfd..aac8a3c2 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs @@ -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; - + /// /// called on client sockets are created in connect /// @@ -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 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); diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs index 081c74a2..c120775b 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs @@ -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; } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/MqttFactory.cs b/Frameworks/MQTTnet.NetStandard/MqttFactory.cs index 2c18ed96..b725f034 100644 --- a/Frameworks/MQTTnet.NetStandard/MqttFactory.cs +++ b/Frameworks/MQTTnet.NetStandard/MqttFactory.cs @@ -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(); + } + public IMqttCommunicationAdapter CreateClientMqttCommunicationAdapter(IMqttClientOptions options) { var logger = _serviceProvider.GetRequiredService>(); - 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 }; diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs index 03206f1f..8fb45ab5 100644 --- a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs @@ -89,7 +89,7 @@ namespace MQTTnet.Implementations RawReceiveStream = ReceiveStream; } - private static Certificate LoadCertificate(MqttClientOptions options) + private static Certificate LoadCertificate(MqttClientTcpOptions options) { if (options.TlsOptions.Certificates == null || !options.TlsOptions.Certificates.Any()) { diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs index 48433165..6145802b 100644 --- a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs +++ b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs @@ -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; } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs b/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs index a3aeca90..68fecd7f 100644 --- a/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs +++ b/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs @@ -42,7 +42,7 @@ namespace MQTTnet public IMqttCommunicationAdapter CreateClientMqttCommunicationAdapter(IMqttClientOptions options) { var logger = _serviceProvider.GetRequiredService>(); - 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)); diff --git a/MQTTnet.Core/Client/IApplicationMessagePublisher.cs b/MQTTnet.Core/Client/IApplicationMessagePublisher.cs new file mode 100644 index 00000000..c77fd3f6 --- /dev/null +++ b/MQTTnet.Core/Client/IApplicationMessagePublisher.cs @@ -0,0 +1,10 @@ +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace MQTTnet.Core.Client +{ + public interface IApplicationMessagePublisher + { + Task PublishAsync(IEnumerable applicationMessages); + } +} diff --git a/MQTTnet.Core/Client/IMqttClient.cs b/MQTTnet.Core/Client/IMqttClient.cs index 9caefbda..6a39e211 100644 --- a/MQTTnet.Core/Client/IMqttClient.cs +++ b/MQTTnet.Core/Client/IMqttClient.cs @@ -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> SubscribeAsync(IEnumerable topicFilters); Task UnsubscribeAsync(IEnumerable topicFilters); - - Task PublishAsync(IEnumerable applicationMessages); } } \ No newline at end of file diff --git a/MQTTnet.Core/Client/IMqttClientChannelOptions.cs b/MQTTnet.Core/Client/IMqttClientChannelOptions.cs new file mode 100644 index 00000000..004430f9 --- /dev/null +++ b/MQTTnet.Core/Client/IMqttClientChannelOptions.cs @@ -0,0 +1,7 @@ +namespace MQTTnet.Core.Client +{ + public interface IMqttClientChannelOptions + { + MqttClientTlsOptions TlsOptions { get; } + } +} diff --git a/MQTTnet.Core/Client/IMqttClientOptions.cs b/MQTTnet.Core/Client/IMqttClientOptions.cs index f805de0b..d2bcf9d3 100644 --- a/MQTTnet.Core/Client/IMqttClientOptions.cs +++ b/MQTTnet.Core/Client/IMqttClientOptions.cs @@ -6,13 +6,14 @@ namespace MQTTnet.Core.Client public interface IMqttClientOptions { string ClientId { get; } + IMqttClientCredentials Credentials { get; } bool CleanSession { get; } MqttApplicationMessage WillMessage { get; } - IMqttClientCredentials Credentials { get; } - - TimeSpan DefaultCommunicationTimeout { get; } + + TimeSpan CommunicationTimeout { get; } TimeSpan KeepAlivePeriod { get; } MqttProtocolVersion ProtocolVersion { get; } - MqttClientTlsOptions TlsOptions { get; } + + IMqttClientChannelOptions ChannelOptions { get; } } } \ No newline at end of file diff --git a/MQTTnet.Core/Client/MqttClient.cs b/MQTTnet.Core/Client/MqttClient.cs index e2094044..e33b2581 100644 --- a/MQTTnet.Core/Client/MqttClient.cs +++ b/MQTTnet.Core/Client/MqttClient.cs @@ -42,6 +42,7 @@ namespace MQTTnet.Core.Client public async Task 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: @@ -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) @@ -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 SendAndReceiveAsync(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); } diff --git a/MQTTnet.Core/Client/BaseMqttClientOptions.cs b/MQTTnet.Core/Client/MqttClientOptions.cs similarity index 77% rename from MQTTnet.Core/Client/BaseMqttClientOptions.cs rename to MQTTnet.Core/Client/MqttClientOptions.cs index ea7ea7e1..04cf8bdb 100644 --- a/MQTTnet.Core/Client/BaseMqttClientOptions.cs +++ b/MQTTnet.Core/Client/MqttClientOptions.cs @@ -5,8 +5,6 @@ namespace MQTTnet.Core.Client { public class MqttClientOptions : IMqttClientOptions { - public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); - public MqttApplicationMessage WillMessage { get; set; } public string ClientId { get; set; } = Guid.NewGuid().ToString("N"); @@ -17,8 +15,10 @@ namespace MQTTnet.Core.Client 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; } } } diff --git a/MQTTnet.Core/Client/MqttClientOptionsBuilder.cs b/MQTTnet.Core/Client/MqttClientOptionsBuilder.cs new file mode 100644 index 00000000..1eb2a347 --- /dev/null +++ b/MQTTnet.Core/Client/MqttClientOptionsBuilder.cs @@ -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; + } + } +} diff --git a/MQTTnet.Core/Client/MqttClientTcpOptions.cs b/MQTTnet.Core/Client/MqttClientTcpOptions.cs index beaa506a..e5cf8d6c 100644 --- a/MQTTnet.Core/Client/MqttClientTcpOptions.cs +++ b/MQTTnet.Core/Client/MqttClientTcpOptions.cs @@ -1,9 +1,11 @@ namespace MQTTnet.Core.Client { - public class MqttClientTcpOptions : MqttClientOptions + public class MqttClientTcpOptions : IMqttClientChannelOptions { public string Server { get; set; } public int? Port { get; set; } + + public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); } } diff --git a/MQTTnet.Core/Client/MqttClientTlsOptions.cs b/MQTTnet.Core/Client/MqttClientTlsOptions.cs index 6fc4ab93..33cd691f 100644 --- a/MQTTnet.Core/Client/MqttClientTlsOptions.cs +++ b/MQTTnet.Core/Client/MqttClientTlsOptions.cs @@ -2,7 +2,7 @@ namespace MQTTnet.Core.Client { - public sealed class MqttClientTlsOptions + public class MqttClientTlsOptions { public bool UseTls { get; set; } diff --git a/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs b/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs index 1a3fe03a..25e4ae81 100644 --- a/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs +++ b/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs @@ -3,7 +3,7 @@ using System.Net; namespace MQTTnet.Core.Client { - public class MqttClientWebSocketOptions : MqttClientOptions + public class MqttClientWebSocketOptions : IMqttClientChannelOptions { public string Uri { get; set; } @@ -12,5 +12,7 @@ namespace MQTTnet.Core.Client public ICollection SubProtocols { get; set; } public CookieContainer CookieContainer { get; set; } + + public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); } } diff --git a/MQTTnet.Core/ManagedClient/IManagedMqttClient.cs b/MQTTnet.Core/ManagedClient/IManagedMqttClient.cs new file mode 100644 index 00000000..8d2ac8cc --- /dev/null +++ b/MQTTnet.Core/ManagedClient/IManagedMqttClient.cs @@ -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 Connected; + event EventHandler Disconnected; + + Task StartAsync(IManagedMqttClientOptions options); + Task StopAsync(); + + Task SubscribeAsync(IEnumerable topicFilters); + Task UnsubscribeAsync(IEnumerable topicFilters); + } +} \ No newline at end of file diff --git a/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs b/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs index 0018a58d..61c0067a 100644 --- a/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs +++ b/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs @@ -11,7 +11,7 @@ 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 _messageQueue = new BlockingCollection(); @@ -91,7 +91,7 @@ namespace MQTTnet.Core.ManagedClient return Task.FromResult(0); } - public Task EnqueueAsync(IEnumerable applicationMessages) + public Task PublishAsync(IEnumerable applicationMessages) { if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); diff --git a/Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs b/Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs index 0655a096..067d08fd 100644 --- a/Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs +++ b/Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs @@ -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; diff --git a/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs b/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs new file mode 100644 index 00000000..cccdff66 --- /dev/null +++ b/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs @@ -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); + } + } + } +} diff --git a/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs b/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs index 587b847f..a02e72c4 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs @@ -28,11 +28,14 @@ namespace MQTTnet.TestApp.NetCore var options = new ManagedMqttClientOptions { - ClientOptions = new MqttClientTcpOptions + ClientOptions = new MqttClientOptions { - Server = "broker.hivemq.com", ClientId = "MQTTnetManagedClientTest", - Credentials = new RandomPassword() + Credentials = new RandomPassword(), + ChannelOptions = new MqttClientTcpOptions + { + Server = "broker.hivemq.com" + } }, AutoReconnectDelay = TimeSpan.FromSeconds(1), diff --git a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs index 746aa74e..6ef76682 100644 --- a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs @@ -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() - .AddConsole(minLevel: LogLevel.Warning, includeScopes: true); - + services.GetService().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(); diff --git a/Tests/MQTTnet.TestApp.NetCore/Program.cs b/Tests/MQTTnet.TestApp.NetCore/Program.cs index f0b28c80..2e1d8c90 100644 --- a/Tests/MQTTnet.TestApp.NetCore/Program.cs +++ b/Tests/MQTTnet.TestApp.NetCore/Program.cs @@ -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() - .AddConsole(); - - var options = new MqttClientWebSocketOptions - { - Uri = "localhost", - ClientId = "XYZ", - CleanSession = true - }; - - var client = services.GetRequiredService(); - 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 - { - 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(); + 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(); } } } diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs index 68dd42a5..3663ba4c 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs @@ -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,24 +40,35 @@ namespace MQTTnet.TestApp.UniversalWindows private async void Connect(object sender, RoutedEventArgs e) { - MqttClientOptions 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(); } @@ -70,12 +79,6 @@ namespace MQTTnet.TestApp.UniversalWindows 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; - try { if (_mqttClient != null) @@ -205,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(); @@ -282,7 +306,7 @@ namespace MQTTnet.TestApp.UniversalWindows } // ---------------------------------- - + // For UWP apps: MqttTcpChannel.CustomIgnorableServerCertificateErrorsResolver = o => {