diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs
index 55a38250..a4d71b39 100644
--- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs
+++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs
@@ -283,15 +283,19 @@ namespace MQTTnet.Adapter
private static bool IsWrappedException(Exception exception)
{
- return exception is TaskCanceledException ||
- exception is OperationCanceledException ||
+ return exception is OperationCanceledException ||
exception is MqttCommunicationTimedOutException ||
exception is MqttCommunicationException;
}
private static void WrapException(Exception exception)
{
- if (exception is IOException && exception.InnerException is SocketException socketException)
+ if (exception is IOException && exception.InnerException is SocketException innerException)
+ {
+ exception = innerException;
+ }
+
+ if (exception is SocketException socketException)
{
if (socketException.SocketErrorCode == SocketError.ConnectionAborted ||
socketException.SocketErrorCode == SocketError.OperationAborted)
diff --git a/Source/MQTTnet/Adapter/MqttServerAdapterClientAcceptedEventArgs.cs b/Source/MQTTnet/Adapter/MqttServerAdapterClientAcceptedEventArgs.cs
index 42852493..0c7db4a5 100644
--- a/Source/MQTTnet/Adapter/MqttServerAdapterClientAcceptedEventArgs.cs
+++ b/Source/MQTTnet/Adapter/MqttServerAdapterClientAcceptedEventArgs.cs
@@ -5,12 +5,12 @@ namespace MQTTnet.Adapter
{
public class MqttServerAdapterClientAcceptedEventArgs : EventArgs
{
- public MqttServerAdapterClientAcceptedEventArgs(IMqttChannelAdapter client)
+ public MqttServerAdapterClientAcceptedEventArgs(IMqttChannelAdapter channelAdapter)
{
- Client = client ?? throw new ArgumentNullException(nameof(client));
+ ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
}
- public IMqttChannelAdapter Client { get; }
+ public IMqttChannelAdapter ChannelAdapter { get; }
public Task SessionTask { get; set; }
}
diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs
index d397985d..ab52885e 100644
--- a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs
+++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs
@@ -1,5 +1,4 @@
using System;
-using System.Collections.Generic;
using MQTTnet.Adapter;
using MQTTnet.Packets;
diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs
index 33d9f18a..138d2eb4 100644
--- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs
+++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs
@@ -6,6 +6,7 @@ using System.Security.Cryptography.X509Certificates;
using System.Threading.Tasks;
using System.IO;
using System.Linq;
+using System.Runtime.ExceptionServices;
using System.Threading;
using MQTTnet.Channel;
using MQTTnet.Client.Options;
@@ -17,12 +18,8 @@ namespace MQTTnet.Implementations
private readonly IMqttClientOptions _clientOptions;
private readonly MqttClientTcpOptions _options;
- private Socket _socket;
private Stream _stream;
- ///
- /// called on client sockets are created in connect
- ///
public MqttTcpChannel(IMqttClientOptions clientOptions)
{
_clientOptions = clientOptions ?? throw new ArgumentNullException(nameof(clientOptions));
@@ -31,48 +28,60 @@ namespace MQTTnet.Implementations
IsSecureConnection = clientOptions.ChannelOptions?.TlsOptions?.UseTls == true;
}
- ///
- /// called on server, sockets are passed in
- /// connect will not be called
- ///
- public MqttTcpChannel(Socket socket, SslStream sslStream)
+ public MqttTcpChannel(Stream stream)
{
- _socket = socket ?? throw new ArgumentNullException(nameof(socket));
+ _stream = stream ?? throw new ArgumentNullException(nameof(stream));
- IsSecureConnection = sslStream != null;
-
- CreateStream(sslStream);
+ IsSecureConnection = stream is SslStream;
}
- public string Endpoint => _socket?.RemoteEndPoint?.ToString();
+ public string Endpoint { get; private set; }
public bool IsSecureConnection { get; }
public async Task ConnectAsync(CancellationToken cancellationToken)
{
- if (_socket == null)
+ Socket socket;
+
+ if (_options.AddressFamily == AddressFamily.Unspecified)
{
- _socket = new Socket(SocketType.Stream, ProtocolType.Tcp) { NoDelay = true };
+ socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+ }
+ else
+ {
+ socket = new Socket(_options.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
}
+ socket.ReceiveBufferSize = _options.BufferSize;
+ socket.SendBufferSize = _options.BufferSize;
+ socket.DualMode = _options.DualMode;
+ socket.NoDelay = _options.NoDelay;
+
// Workaround for: workaround for https://github.com/dotnet/corefx/issues/24430
- using (cancellationToken.Register(() => _socket.Dispose()))
+ using (cancellationToken.Register(() => socket.Dispose()))
{
#if NET452 || NET461
- await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, _options.Server, _options.GetPort(), null).ConfigureAwait(false);
+ await Task.Factory.FromAsync(socket.BeginConnect, socket.EndConnect, _options.Server, _options.GetPort(), null).ConfigureAwait(false);
#else
- await _socket.ConnectAsync(_options.Server, _options.GetPort()).ConfigureAwait(false);
+ await socket.ConnectAsync(_options.Server, _options.GetPort()).ConfigureAwait(false);
#endif
}
- SslStream sslStream = null;
+ var networkStream = new NetworkStream(socket, true);
+
if (_options.TlsOptions.UseTls)
{
- sslStream = new SslStream(new NetworkStream(_socket, true), false, InternalUserCertificateValidationCallback);
+ var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
+
+ _stream = sslStream;
+ }
+ else
+ {
+ _stream = networkStream;
}
- CreateStream(sslStream);
+ Endpoint = socket.RemoteEndPoint?.ToString();
}
public Task DisconnectAsync(CancellationToken cancellationToken)
@@ -83,37 +92,58 @@ namespace MQTTnet.Implementations
public async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
- // Workaround for: https://github.com/dotnet/corefx/issues/24430
- using (cancellationToken.Register(Dispose))
+ try
{
- if (cancellationToken.IsCancellationRequested)
+ // Workaround for: https://github.com/dotnet/corefx/issues/24430
+ using (cancellationToken.Register(Dispose))
{
- return 0;
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return 0;
+ }
+
+ return await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+ }
+ }
+ catch (IOException exception)
+ {
+ if (exception.InnerException is SocketException socketException)
+ {
+ ExceptionDispatchInfo.Capture(socketException).Throw();
}
- return await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+ throw;
}
}
public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
- // Workaround for: https://github.com/dotnet/corefx/issues/24430
- using (cancellationToken.Register(Dispose))
+ try
{
- if (cancellationToken.IsCancellationRequested)
+ // Workaround for: https://github.com/dotnet/corefx/issues/24430
+ using (cancellationToken.Register(Dispose))
{
- return;
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return;
+ }
+
+ await _stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+ }
+ }
+ catch (IOException exception)
+ {
+ if (exception.InnerException is SocketException socketException)
+ {
+ ExceptionDispatchInfo.Capture(socketException).Throw();
}
- await _stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
- //await _stream.FlushAsync(cancellationToken);
+ throw;
}
}
public void Dispose()
{
- _socket = null;
-
// When the stream is disposed it will also close the socket and this will also dispose it.
// So there is no need to dispose the socket again.
// https://stackoverflow.com/questions/3601521/should-i-manually-dispose-the-socket-after-closing-it
@@ -127,6 +157,8 @@ namespace MQTTnet.Implementations
catch (NullReferenceException)
{
}
+
+ _stream = null;
}
private bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
@@ -175,18 +207,6 @@ namespace MQTTnet.Implementations
return certificates;
}
-
- private void CreateStream(Stream stream)
- {
- if (stream != null)
- {
- _stream = stream;
- }
- else
- {
- _stream = new NetworkStream(_socket, true);
- }
- }
}
}
#endif
diff --git a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs
index f1207b2d..8d771904 100644
--- a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs
+++ b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs
@@ -1,5 +1,6 @@
#if !WINDOWS_UWP
using System;
+using System.IO;
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
@@ -35,14 +36,14 @@ namespace MQTTnet.Implementations
_tlsCertificate = tlsCertificate;
_cancellationToken = cancellationToken;
_logger = logger.CreateChildLogger(nameof(MqttTcpServerListener));
-
+
if (_options is MqttServerTlsTcpEndpointOptions tlsOptions)
{
_tlsOptions = tlsOptions;
}
}
- public Action ClientAcceptedHandler;
+ public Action ClientAcceptedHandler { get; set; }
public void Start()
{
@@ -58,53 +59,7 @@ namespace MQTTnet.Implementations
_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;
-
- SslStream sslStream = null;
-
- if (_tlsCertificate != null)
- {
- sslStream = new SslStream(new NetworkStream(clientSocket), false);
- await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, _tlsOptions.SslProtocol, false).ConfigureAwait(false);
- }
-
- _logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.",
- clientSocket.RemoteEndPoint,
- _socket.LocalEndPoint,
- _addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
-
- var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketFormatterAdapter(), _logger);
- ClientAcceptedHandler?.Invoke(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);
- }
- }
+ Task.Run(() => AcceptClientConnectionsAsync(_cancellationToken), _cancellationToken);
}
public void Dispose()
@@ -115,6 +70,78 @@ namespace MQTTnet.Implementations
_tlsCertificate?.Dispose();
#endif
}
+
+ private async Task AcceptClientConnectionsAsync(CancellationToken cancellationToken)
+ {
+ 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
+ await HandleClientConnection(clientSocket).ConfigureAwait(false);
+ }
+ catch (Exception exception)
+ {
+ _logger.Error(exception, $"Error while accepting connection at TCP listener {_socket.LocalEndPoint} TLS={_tlsCertificate != null}.");
+ await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
+ }
+ }
+ }
+
+ private async Task HandleClientConnection(Socket clientSocket)
+ {
+ Stream stream = null;
+
+ try
+ {
+ clientSocket.NoDelay = _options.NoDelay;
+
+ stream = new NetworkStream(clientSocket, true);
+
+ if (_tlsCertificate != null)
+ {
+ var sslStream = new SslStream(stream, false);
+ await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, _tlsOptions.SslProtocol, false).ConfigureAwait(false);
+ stream = sslStream;
+ }
+
+ _logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.",
+ clientSocket.RemoteEndPoint,
+ _socket.LocalEndPoint,
+ _addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
+
+ var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(stream), new MqttPacketFormatterAdapter(), _logger);
+ ClientAcceptedHandler?.Invoke(new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
+ }
+ catch (Exception exception)
+ {
+ if (exception is ObjectDisposedException)
+ {
+ // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
+ return;
+ }
+
+ if (exception is SocketException socketException && socketException.SocketErrorCode == SocketError.OperationAborted)
+ {
+ return;
+ }
+
+ try
+ {
+ // Dispose already allocated resources.
+ stream?.Dispose();
+ clientSocket?.Dispose();
+ }
+ catch (Exception disposeException)
+ {
+ throw new AggregateException(exception, disposeException);
+ }
+ }
+ }
}
}
#endif
\ No newline at end of file
diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs
index 7ce7adf1..df9028c2 100644
--- a/Source/MQTTnet/Server/MqttClientConnection.cs
+++ b/Source/MQTTnet/Server/MqttClientConnection.cs
@@ -127,7 +127,7 @@ namespace MQTTnet.Server
Session.EnqueueApplicationMessage(applicationMessage, senderClientId, isRetainedApplicationMessage);
- _logger.Verbose("Enqueued application message (ClientId: {0}).", ClientId);
+ _logger.Verbose("Queued application message (ClientId: {0}).", ClientId);
}
private async Task RunInternalAsync()
@@ -362,24 +362,6 @@ namespace MQTTnet.Server
};
return SendAsync(pubRecPacket);
-
- ////Task.Run(async () =>
- ////{
- //// using (var pubRelPacketAwaiter = _packetDispatcher.AddPacketAwaiter(publishPacket.PacketIdentifier))
- //// {
- //// await pubRelPacketAwaiter.WaitOneAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
-
- //// var pubCompPacket = new MqttPubCompPacket
- //// {
- //// PacketIdentifier = publishPacket.PacketIdentifier,
- //// ReasonCode = MqttPubCompReasonCode.Success
- //// };
-
- //// await SendAsync(pubCompPacket).ConfigureAwait(false);
- //// }
- ////});
-
- //await Task.FromResult(0);
}
private async Task SendPendingPacketsAsync(CancellationToken cancellationToken)
diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs
index 6c55d43b..06d61048 100644
--- a/Source/MQTTnet/Server/MqttServer.cs
+++ b/Source/MQTTnet/Server/MqttServer.cs
@@ -186,7 +186,7 @@ namespace MQTTnet.Server
private void OnClientAccepted(MqttServerAdapterClientAcceptedEventArgs eventArgs)
{
- eventArgs.SessionTask = _clientSessionsManager.HandleConnectionAsync(eventArgs.Client);
+ eventArgs.SessionTask = _clientSessionsManager.HandleConnectionAsync(eventArgs.ChannelAdapter);
}
}
}
diff --git a/Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs b/Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs
index c18a7313..b8c54c4b 100644
--- a/Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs
+++ b/Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs
@@ -21,7 +21,7 @@ namespace MQTTnet.Benchmarks
{
var factory = new MqttFactory();
var tcpServer = new MqttTcpServerAdapter(new MqttNetLogger().CreateChildLogger());
- tcpServer.ClientAcceptedHandler += args => _serverChannel = (IMqttChannel)args.Client.GetType().GetField("_channel", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance).GetValue(args.Client);
+ tcpServer.ClientAcceptedHandler += args => _serverChannel = (IMqttChannel)args.ChannelAdapter.GetType().GetField("_channel", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance).GetValue(args.ChannelAdapter);
_mqttServer = factory.CreateMqttServer(new[] { tcpServer }, new MqttNetLogger());