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

Refactor server socket management for Linux support.

Browse Source
This commit is contained in:
Christian Kratky
2018-06-09 15:31:47 +02:00
parent 9e318fc40d
commit e62316105f
24 changed files with 229 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpServerAdapter.Uwp.cs
+2 -2
View File
@@ -38,8 +38,8 @@ namespace MQTTnet.Implementations
_defaultEndpointSocket.Control.KeepAlive = true;
_defaultEndpointSocket.Control.QualityOfService = SocketQualityOfService.LowLatency;
_defaultEndpointSocket.ConnectionReceived += AcceptDefaultEndpointConnectionsAsync;
await _defaultEndpointSocket.BindServiceNameAsync(options.GetDefaultEndpointPort().ToString(), SocketProtectionLevel.PlainSocket);
await _defaultEndpointSocket.BindServiceNameAsync(options.DefaultEndpointOptions.Port.ToString(), SocketProtectionLevel.PlainSocket);
}
Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpServerAdapter.cs
+49 -99
View File
@@ -1,28 +1,23 @@
#if NET452 || NET461 || NETSTANDARD1_3 || NETSTANDARD2_0
using System;
using System.Net;
using System.Net.Security;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Serializer;
using MQTTnet.Server;
namespace MQTTnet.Implementations
{
public class MqttTcpServerAdapter : IMqttServerAdapter
{
private readonly List<MqttTcpServerListener> _listeners = new List<MqttTcpServerListener>();
private readonly IMqttNetChildLogger _logger;
private CancellationTokenSource _cancellationTokenSource;
private Socket _defaultEndpointSocket;
private Socket _tlsEndpointSocket;
private X509Certificate2 _tlsCertificate;
public MqttTcpServerAdapter(IMqttNetChildLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
@@ -40,13 +35,7 @@ namespace MQTTnet.Implementations
if (options.DefaultEndpointOptions.IsEnabled)
{
_defaultEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp) { NoDelay = true };
_defaultEndpointSocket.Bind(new IPEndPoint(options.DefaultEndpointOptions.BoundIPAddress, options.GetDefaultEndpointPort()));
_defaultEndpointSocket.Listen(options.ConnectionBacklog);
Task.Run(() => AcceptDefaultEndpointConnectionsAsync(_cancellationTokenSource.Token),
_cancellationTokenSource.Token);
RegisterListeners(options.DefaultEndpointOptions);
}
if (options.TlsEndpointOptions.IsEnabled)
@@ -56,19 +45,13 @@ namespace MQTTnet.Implementations
throw new ArgumentException("TLS certificate is not set.");
}
_tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate);
if (!_tlsCertificate.HasPrivateKey)
var tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate);
if (!tlsCertificate.HasPrivateKey)
{
throw new InvalidOperationException("The certificate for TLS encryption must contain the private key.");
}
_tlsEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
_tlsEndpointSocket.Bind(new IPEndPoint(options.TlsEndpointOptions.BoundIPAddress, options.GetTlsEndpointPort()));
_tlsEndpointSocket.Listen(options.ConnectionBacklog);
Task.Run(
() => AcceptTlsEndpointConnectionsAsync(_cancellationTokenSource.Token),
_cancellationTokenSource.Token);
RegisterListeners(options.TlsEndpointOptions);
}
return Task.FromResult(0);
@@ -76,93 +59,60 @@ namespace MQTTnet.Implementations
public Task StopAsync()
{
_cancellationTokenSource?.Cancel(false);
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
_defaultEndpointSocket?.Dispose();
_defaultEndpointSocket = null;
_tlsCertificate = null;
_tlsEndpointSocket?.Dispose();
_tlsEndpointSocket = null;
Dispose();
return Task.FromResult(0);
}
public void Dispose()
{
StopAsync().GetAwaiter().GetResult();
_cancellationTokenSource?.Cancel(false);
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
foreach (var listener in _listeners)
{
listener.Dispose();
}
_listeners.Clear();
}
private async Task AcceptDefaultEndpointConnectionsAsync(CancellationToken cancellationToken)
private void RegisterListeners(MqttServerTcpEndpointBaseOptions options)
{
while (!cancellationToken.IsCancellationRequested)
var tlsOptions = options as MqttServerTlsTcpEndpointOptions;
X509Certificate2 tlsCertificate = null;
if (tlsOptions != null)
{
try
{
//todo: else branch can be used with min dependency NET46
#if NET452 || NET461
var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null).ConfigureAwait(false);
#else
var clientSocket = await _defaultEndpointSocket.AcceptAsync().ConfigureAwait(false);
#endif
clientSocket.NoDelay = true;
var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer(), _logger);
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
}
catch (ObjectDisposedException)
{
// It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
}
catch (Exception exception)
{
if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
{
return;
}
_logger.Error(exception, "Error while accepting connection at default endpoint.");
await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
}
tlsCertificate = new X509Certificate2(tlsOptions.Certificate);
}
var listenerV4 = new MqttTcpServerListener(
AddressFamily.InterNetwork,
options,
tlsCertificate,
_cancellationTokenSource.Token,
_logger);
listenerV4.ClientAccepted += OnClientAccepted;
listenerV4.Start();
_listeners.Add(listenerV4);
var listenerV6 = new MqttTcpServerListener(
AddressFamily.InterNetworkV6,
options,
tlsCertificate,
_cancellationTokenSource.Token,
_logger);
listenerV6.ClientAccepted += OnClientAccepted;
listenerV6.Start();
_listeners.Add(listenerV6);
}
private async Task AcceptTlsEndpointConnectionsAsync(CancellationToken cancellationToken)
private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs e)
{
while (!cancellationToken.IsCancellationRequested)
{
try
{
#if NET452 || NET461
var clientSocket = await Task.Factory.FromAsync(_tlsEndpointSocket.BeginAccept, _tlsEndpointSocket.EndAccept, null).ConfigureAwait(false);
#else
var clientSocket = await _tlsEndpointSocket.AcceptAsync().ConfigureAwait(false);
#endif
var sslStream = new SslStream(new NetworkStream(clientSocket));
await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false);
var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer(), _logger);
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
}
catch (ObjectDisposedException)
{
// It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
}
catch (Exception exception)
{
if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
{
return;
}
_logger.Error(exception, "Error while accepting connection at TLS endpoint.");
await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
}
}
ClientAccepted?.Invoke(this, e);
}
}
}
Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpServerListener.cs
+109
View File
@@ -0,0 +1,109 @@
#if NET452 || NET461 || NETSTANDARD1_3 || NETSTANDARD2_0
using System;
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Serializer;
using MQTTnet.Server;
namespace MQTTnet.Implementations
{
public class MqttTcpServerListener : IDisposable
{
private readonly IMqttNetChildLogger _logger;
private readonly CancellationToken _cancellationToken;
private readonly AddressFamily _addressFamily;
private readonly MqttServerTcpEndpointBaseOptions _options;
private readonly X509Certificate2 _tlsCertificate;
private Socket _socket;
public MqttTcpServerListener(
AddressFamily addressFamily,
MqttServerTcpEndpointBaseOptions options,
X509Certificate2 tlsCertificate,
CancellationToken cancellationToken,
IMqttNetChildLogger logger)
{
_addressFamily = addressFamily;
_options = options;
_tlsCertificate = tlsCertificate;
_cancellationToken = cancellationToken;
_logger = logger.CreateChildLogger(nameof(MqttTcpServerListener));
}
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
public void Start()
{
var boundIp = _options.BoundInterNetworkAddress;
if (_addressFamily == AddressFamily.InterNetworkV6)
{
boundIp = _options.BoundInterNetworkV6Address;
}
_socket = new Socket(_addressFamily, SocketType.Stream, ProtocolType.Tcp);
_socket.Bind(new IPEndPoint(boundIp, _options.Port));
_logger.Info($"Starting TCP listener for {_socket.LocalEndPoint} TLS={_tlsCertificate != null}.");
_socket.Listen(_options.ConnectionBacklog);
Task.Run(AcceptClientConnectionsAsync, _cancellationToken);
}
private async Task AcceptClientConnectionsAsync()
{
while (!_cancellationToken.IsCancellationRequested)
{
try
{
#if NET452 || NET461
var clientSocket = await Task.Factory.FromAsync(_socket.BeginAccept, _socket.EndAccept, null).ConfigureAwait(false);
#else
var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
#endif
clientSocket.NoDelay = true;
if (_tlsCertificate != null)
{
var sslStream = new SslStream(new NetworkStream(clientSocket), false);
await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false);
}
var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer(), _logger);
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
}
catch (ObjectDisposedException)
{
// It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
}
catch (Exception exception)
{
if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
{
return;
}
_logger.Error(exception, $"Error while accepting connection at TCP listener {_socket.LocalEndPoint} TLS={_tlsCertificate != null}.");
await Task.Delay(TimeSpan.FromSeconds(1), _cancellationToken).ConfigureAwait(false);
}
}
}
public void Dispose()
{
_socket?.Dispose();
#if NETSTANDARD1_3 || NETSTANDARD2_0 || NET461
_tlsCertificate?.Dispose();
#endif
}
}
}
#endif
Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj → Frameworks/MQTTnet.NetStandard/MQTTnet.csproj
View File
Frameworks/MQTTnet.NetStandard/Server/IMqttServerOptions.cs
+2 -4
View File
@@ -4,8 +4,6 @@ namespace MQTTnet.Server
{
public interface IMqttServerOptions
{
int ConnectionBacklog { get; }
bool EnablePersistentSessions { get; }
int MaxPendingMessagesPerClient { get; }
@@ -18,8 +16,8 @@ namespace MQTTnet.Server
Action<MqttApplicationMessageInterceptorContext> ApplicationMessageInterceptor { get; }
Action<MqttClientMessageQueueInterceptorContext> ClientMessageQueueInterceptor { get; set; }
MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; }
MqttServerTlsEndpointOptions TlsEndpointOptions { get; }
MqttServerTcpEndpointOptions DefaultEndpointOptions { get; }
MqttServerTlsTcpEndpointOptions TlsEndpointOptions { get; }
IMqttServerStorage Storage { get; }
}
Frameworks/MQTTnet.NetStandard/Server/MqttServerDefaultEndpointOptions.cs
-13
View File
@@ -1,13 +0,0 @@
using System.Net;
namespace MQTTnet.Server
{
public class MqttServerDefaultEndpointOptions
{
public bool IsEnabled { get; set; } = true;
public int? Port { get; set; }
public IPAddress BoundIPAddress { get; set; } = IPAddress.Any;
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttServerOptions.cs
+2 -4
View File
@@ -4,14 +4,12 @@ namespace MQTTnet.Server
{
public class MqttServerOptions : IMqttServerOptions
{
public MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; } = new MqttServerDefaultEndpointOptions();
public MqttServerTcpEndpointOptions DefaultEndpointOptions { get; } = new MqttServerTcpEndpointOptions();
public MqttServerTlsEndpointOptions TlsEndpointOptions { get; } = new MqttServerTlsEndpointOptions();
public MqttServerTlsTcpEndpointOptions TlsEndpointOptions { get; } = new MqttServerTlsTcpEndpointOptions();
public bool EnablePersistentSessions { get; set; }
public int ConnectionBacklog { get; set; } = 10;
public int MaxPendingMessagesPerClient { get; set; } = 250;
public MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; set; } = MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage;
Frameworks/MQTTnet.NetStandard/Server/MqttServerOptionsBuilder.cs
+13 -6
View File
@@ -9,7 +9,8 @@ namespace MQTTnet.Server
public MqttServerOptionsBuilder WithConnectionBacklog(int value)
{
_options.ConnectionBacklog = value;
_options.DefaultEndpointOptions.ConnectionBacklog = value;
_options.TlsEndpointOptions.ConnectionBacklog = value;
return this;
}
@@ -31,7 +32,7 @@ namespace MQTTnet.Server
return this;
}
public MqttServerOptionsBuilder WithDefaultEndpointPort(int? value)
public MqttServerOptionsBuilder WithDefaultEndpointPort(int value)
{
_options.DefaultEndpointOptions.Port = value;
return this;
@@ -39,10 +40,16 @@ namespace MQTTnet.Server
public MqttServerOptionsBuilder WithDefaultEndpointBoundIPAddress(IPAddress value)
{
_options.DefaultEndpointOptions.BoundIPAddress = value ?? IPAddress.Any;
_options.DefaultEndpointOptions.BoundInterNetworkAddress = value ?? IPAddress.Any;
return this;
}
public MqttServerOptionsBuilder WithDefaultEndpointBoundIPV6Address(IPAddress value)
{
_options.DefaultEndpointOptions.BoundInterNetworkV6Address = value ?? IPAddress.Any;
return this;
}
public MqttServerOptionsBuilder WithoutDefaultEndpoint()
{
_options.DefaultEndpointOptions.IsEnabled = false;
@@ -55,7 +62,7 @@ namespace MQTTnet.Server
return this;
}
public MqttServerOptionsBuilder WithEncryptedEndpointPort(int? value)
public MqttServerOptionsBuilder WithEncryptedEndpointPort(int value)
{
_options.TlsEndpointOptions.Port = value;
return this;
@@ -63,7 +70,7 @@ namespace MQTTnet.Server
public MqttServerOptionsBuilder WithEncryptedEndpointBoundIPAddress(IPAddress value)
{
_options.TlsEndpointOptions.BoundIPAddress = value;
_options.TlsEndpointOptions.BoundInterNetworkAddress = value;
return this;
}
Frameworks/MQTTnet.NetStandard/Server/MqttServerOptionsExtensions.cs
-31
View File
@@ -1,31 +0,0 @@
using System;
namespace MQTTnet.Server
{
public static class MqttServerOptionsExtensions
{
public static int GetTlsEndpointPort(this IMqttServerOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (!options.TlsEndpointOptions.Port.HasValue)
{
return 8883;
}
return options.TlsEndpointOptions.Port.Value;
}
public static int GetDefaultEndpointPort(this IMqttServerOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (!options.DefaultEndpointOptions.Port.HasValue)
{
return 1883;
}
return options.DefaultEndpointOptions.Port.Value;
}
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttServerTcpEndpointBaseOptions.cs
+17
View File
@@ -0,0 +1,17 @@
using System.Net;
namespace MQTTnet.Server
{
public abstract class MqttServerTcpEndpointBaseOptions
{
public bool IsEnabled { get; set; }
public int Port { get; set; }
public int ConnectionBacklog { get; set; } = 10;
public IPAddress BoundInterNetworkAddress { get; set; } = IPAddress.Any;
public IPAddress BoundInterNetworkV6Address { get; set; } = IPAddress.IPv6Any;
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttServerTcpEndpointOptions.cs
+11
View File
@@ -0,0 +1,11 @@
namespace MQTTnet.Server
{
public class MqttServerTcpEndpointOptions : MqttServerTcpEndpointBaseOptions
{
public MqttServerTcpEndpointOptions()
{
IsEnabled = true;
Port = 1883;
}
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttServerTlsEndpointOptions.cs
-15
View File
@@ -1,15 +0,0 @@
using System.Net;
namespace MQTTnet.Server
{
public class MqttServerTlsEndpointOptions
{
public bool IsEnabled { get; set; }
public int? Port { get; set; }
public byte[] Certificate { get; set; }
public IPAddress BoundIPAddress { get; set; } = IPAddress.Any;
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttServerTlsTcpEndpointOptions.cs
+12
View File
@@ -0,0 +1,12 @@
namespace MQTTnet.Server
{
public class MqttServerTlsTcpEndpointOptions : MqttServerTcpEndpointBaseOptions
{
public MqttServerTlsTcpEndpointOptions()
{
Port = 8883;
}
public byte[] Certificate { get; set; }
}
}