Refactor options and add options builder

This commit is contained in:
Christian Kratky
2017-10-25 22:33:07 +02:00
parent a9b394401b
commit a18c533603
27 changed files with 448 additions and 233 deletions
@@ -0,0 +1,10 @@
using System.Collections.Generic;
using System.Threading.Tasks;
namespace MQTTnet.Core.Client
{
public interface IApplicationMessagePublisher
{
Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages);
}
}
+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);
}
}
@@ -0,0 +1,7 @@
namespace MQTTnet.Core.Client
{
public interface IMqttClientChannelOptions
{
MqttClientTlsOptions TlsOptions { get; }
}
}
+5 -4
View File
@@ -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; }
}
}
+9 -7
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:
@@ -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<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);
}
@@ -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; }
}
}
@@ -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;
}
}
}
+3 -1
View File
@@ -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();
}
}
+1 -1
View File
@@ -2,7 +2,7 @@
namespace MQTTnet.Core.Client
{
public sealed class MqttClientTlsOptions
public class MqttClientTlsOptions
{
public bool UseTls { get; set; }
@@ -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<string> SubProtocols { get; set; }
public CookieContainer CookieContainer { get; set; }
public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
}
}