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

Improve logging and exception handling when stopping the server.

Browse Source
This commit is contained in:
Christian Kratky
2019-06-23 15:58:21 +02:00
parent dd96cf98a8
commit 7a767c7d2f
3 changed files with 70 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs
+13 -9
View File
@@ -28,6 +28,8 @@ namespace MQTTnet.Implementations
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
public bool TreatSocketOpeningErrorAsWarning { get; set; }
public Task StartAsync(IMqttServerOptions options)
{
if (_cancellationTokenSource != null) throw new InvalidOperationException("Server is already started.");
@@ -36,7 +38,7 @@ namespace MQTTnet.Implementations
if (options.DefaultEndpointOptions.IsEnabled)
{
RegisterListeners(options.DefaultEndpointOptions, null);
RegisterListeners(options.DefaultEndpointOptions, null, _cancellationTokenSource.Token);
}
if (options.TlsEndpointOptions.IsEnabled)
@@ -52,7 +54,7 @@ namespace MQTTnet.Implementations
throw new InvalidOperationException("The certificate for TLS encryption must contain the private key.");
}
RegisterListeners(options.TlsEndpointOptions, tlsCertificate);
RegisterListeners(options.TlsEndpointOptions, tlsCertificate, _cancellationTokenSource.Token);
}
return Task.FromResult(0);
@@ -78,7 +80,7 @@ namespace MQTTnet.Implementations
_listeners.Clear();
}
private void RegisterListeners(MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate)
private void RegisterListeners(MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate, CancellationToken cancellationToken)
{
if (!options.BoundInterNetworkAddress.Equals(IPAddress.None))
{
@@ -86,14 +88,15 @@ namespace MQTTnet.Implementations
AddressFamily.InterNetwork,
options,
tlsCertificate,
_cancellationTokenSource.Token,
_logger)
{
ClientHandler = OnClientAcceptedAsync
};
listenerV4.Start();
_listeners.Add(listenerV4);
if (listenerV4.Start(TreatSocketOpeningErrorAsWarning, cancellationToken))
{
_listeners.Add(listenerV4);
}
}
if (!options.BoundInterNetworkV6Address.Equals(IPAddress.None))
@@ -102,14 +105,15 @@ namespace MQTTnet.Implementations
AddressFamily.InterNetworkV6,
options,
tlsCertificate,
_cancellationTokenSource.Token,
_logger)
{
ClientHandler = OnClientAcceptedAsync
};
listenerV6.Start();
_listeners.Add(listenerV6);
if (listenerV6.Start(TreatSocketOpeningErrorAsWarning, cancellationToken))
{
_listeners.Add(listenerV6);
}
}
}
Source/MQTTnet/Implementations/MqttTcpServerListener.cs
+53 -20
View File
@@ -10,6 +10,7 @@ using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Server;
namespace MQTTnet.Implementations
@@ -17,24 +18,23 @@ 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 MqttServerTlsTcpEndpointOptions _tlsOptions;
private readonly X509Certificate2 _tlsCertificate;
private Socket _socket;
private IPEndPoint _localEndPoint;
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));
if (_options is MqttServerTlsTcpEndpointOptions tlsOptions)
@@ -45,21 +45,38 @@ namespace MQTTnet.Implementations
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
public void Start()
public bool Start(bool treatErrorsAsWarning, CancellationToken cancellationToken)
{
var boundIp = _options.BoundInterNetworkAddress;
if (_addressFamily == AddressFamily.InterNetworkV6)
try
{
boundIp = _options.BoundInterNetworkV6Address;
var boundIp = _options.BoundInterNetworkAddress;
if (_addressFamily == AddressFamily.InterNetworkV6)
{
boundIp = _options.BoundInterNetworkV6Address;
}
_localEndPoint = new IPEndPoint(boundIp, _options.Port);
_logger.Info($"Starting TCP listener for {_localEndPoint} TLS={_tlsCertificate != null}.");
_socket = new Socket(_addressFamily, SocketType.Stream, ProtocolType.Tcp);
_socket.Bind(_localEndPoint);
_socket.Listen(_options.ConnectionBacklog);
Task.Run(() => AcceptClientConnectionsAsync(cancellationToken), cancellationToken).Forget(_logger);
return true;
}
catch (Exception exception)
{
if (!treatErrorsAsWarning)
{
throw;
}
_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), _cancellationToken);
_logger.Warning(exception,"Error while creating listener socket for local end point '{0}'.", _localEndPoint);
return false;
}
}
public void Dispose()
@@ -82,13 +99,29 @@ namespace MQTTnet.Implementations
#else
var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
#endif
#pragma warning disable 4014
Task.Run(() => TryHandleClientConnectionAsync(clientSocket), cancellationToken);
#pragma warning restore 4014
if (clientSocket == null)
{
continue;
}
Task.Run(() => TryHandleClientConnectionAsync(clientSocket), cancellationToken).Forget(_logger);
}
catch (OperationCanceledException)
{
}
catch (Exception exception)
{
_logger.Error(exception, $"Error while accepting connection at TCP listener {_socket.LocalEndPoint} TLS={_tlsCertificate != null}.");
if (exception is SocketException socketException)
{
if (socketException.SocketErrorCode == SocketError.ConnectionAborted ||
socketException.SocketErrorCode == SocketError.OperationAborted)
{
continue;
}
}
_logger.Error(exception, $"Error while accepting connection at TCP listener {_localEndPoint} TLS={_tlsCertificate != null}.");
await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
}
}
@@ -105,7 +138,7 @@ namespace MQTTnet.Implementations
_logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.",
remoteEndPoint,
_socket.LocalEndPoint,
_localEndPoint,
_addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
clientSocket.NoDelay = _options.NoDelay;
@@ -163,7 +196,7 @@ namespace MQTTnet.Implementations
_logger.Verbose("Client '{0}' disconnected at TCP listener '{1}, {2}'.",
remoteEndPoint,
_socket.LocalEndPoint,
_localEndPoint,
_addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
}
catch (Exception disposeException)