You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

164 rivejä
6.7 KiB

  1. #if NET452 || NET461 || NETSTANDARD1_3 || NETSTANDARD2_0
  2. using System;
  3. using System.Net;
  4. using System.Net.Security;
  5. using System.Net.Sockets;
  6. using System.Security.Authentication;
  7. using System.Security.Cryptography.X509Certificates;
  8. using System.Threading;
  9. using System.Threading.Tasks;
  10. using MQTTnet.Adapter;
  11. using MQTTnet.Diagnostics;
  12. using MQTTnet.Serializer;
  13. using MQTTnet.Server;
  14. namespace MQTTnet.Implementations
  15. {
  16. public class MqttTcpServerAdapter : IMqttServerAdapter
  17. {
  18. private readonly IMqttNetLogger _logger;
  19. private CancellationTokenSource _cancellationTokenSource;
  20. private Socket _defaultEndpointSocket;
  21. private Socket _tlsEndpointSocket;
  22. private X509Certificate2 _tlsCertificate;
  23. public MqttTcpServerAdapter(IMqttNetLogger logger)
  24. {
  25. _logger = logger ?? throw new ArgumentNullException(nameof(logger));
  26. }
  27. public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
  28. public Task StartAsync(IMqttServerOptions options)
  29. {
  30. if (_cancellationTokenSource != null) throw new InvalidOperationException("Server is already started.");
  31. _cancellationTokenSource = new CancellationTokenSource();
  32. if (options.DefaultEndpointOptions.IsEnabled)
  33. {
  34. _defaultEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp) { NoDelay = true };
  35. _defaultEndpointSocket.Bind(new IPEndPoint(options.DefaultEndpointOptions.BoundIPAddress, options.GetDefaultEndpointPort()));
  36. _defaultEndpointSocket.Listen(options.ConnectionBacklog);
  37. Task.Run(() => AcceptDefaultEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
  38. }
  39. if (options.TlsEndpointOptions.IsEnabled)
  40. {
  41. if (options.TlsEndpointOptions.Certificate == null)
  42. {
  43. throw new ArgumentException("TLS certificate is not set.");
  44. }
  45. _tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate);
  46. if (!_tlsCertificate.HasPrivateKey)
  47. {
  48. throw new InvalidOperationException("The certificate for TLS encryption must contain the private key.");
  49. }
  50. _tlsEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
  51. _tlsEndpointSocket.Bind(new IPEndPoint(options.TlsEndpointOptions.BoundIPAddress, options.GetTlsEndpointPort()));
  52. _tlsEndpointSocket.Listen(options.ConnectionBacklog);
  53. Task.Run(() => AcceptTlsEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
  54. }
  55. return Task.FromResult(0);
  56. }
  57. public Task StopAsync()
  58. {
  59. _cancellationTokenSource?.Cancel(false);
  60. _cancellationTokenSource?.Dispose();
  61. _cancellationTokenSource = null;
  62. _defaultEndpointSocket?.Dispose();
  63. _defaultEndpointSocket = null;
  64. _tlsCertificate = null;
  65. _tlsEndpointSocket?.Dispose();
  66. _tlsEndpointSocket = null;
  67. return Task.FromResult(0);
  68. }
  69. public void Dispose()
  70. {
  71. StopAsync();
  72. }
  73. private async Task AcceptDefaultEndpointConnectionsAsync(CancellationToken cancellationToken)
  74. {
  75. while (!cancellationToken.IsCancellationRequested)
  76. {
  77. try
  78. {
  79. //todo: else branch can be used with min dependency NET46
  80. #if NET452 || NET461
  81. var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null).ConfigureAwait(false);
  82. #else
  83. var clientSocket = await _defaultEndpointSocket.AcceptAsync().ConfigureAwait(false);
  84. #endif
  85. clientSocket.NoDelay = true;
  86. var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer(), _logger);
  87. ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
  88. }
  89. catch (ObjectDisposedException)
  90. {
  91. // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
  92. }
  93. catch (Exception exception)
  94. {
  95. if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
  96. {
  97. return;
  98. }
  99. _logger.Error<MqttTcpServerAdapter>(exception, "Error while accepting connection at default endpoint.");
  100. await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
  101. }
  102. }
  103. }
  104. private async Task AcceptTlsEndpointConnectionsAsync(CancellationToken cancellationToken)
  105. {
  106. while (!cancellationToken.IsCancellationRequested)
  107. {
  108. try
  109. {
  110. #if NET452 || NET461
  111. var clientSocket = await Task.Factory.FromAsync(_tlsEndpointSocket.BeginAccept, _tlsEndpointSocket.EndAccept, null).ConfigureAwait(false);
  112. #else
  113. var clientSocket = await _tlsEndpointSocket.AcceptAsync().ConfigureAwait(false);
  114. #endif
  115. var sslStream = new SslStream(new NetworkStream(clientSocket));
  116. await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false);
  117. var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer(), _logger);
  118. ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
  119. }
  120. catch (ObjectDisposedException)
  121. {
  122. // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
  123. }
  124. catch (Exception exception)
  125. {
  126. if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
  127. {
  128. return;
  129. }
  130. _logger.Error<MqttTcpServerAdapter>(exception, "Error while accepting connection at TLS endpoint.");
  131. await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
  132. }
  133. }
  134. }
  135. }
  136. }
  137. #endif