YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-19 02:53:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix breaking changes.

Browse Source
This commit is contained in:
Christian Kratky
2017-06-03 22:48:22 +02:00
parent 1317fa3e2a
commit 929cdb971a
92 changed files with 1363 additions and 1309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Frameworks/MQTTnet.NetCoreApp/Implementations/MqttServerAdapter.cs
+117
View File
@@ -0,0 +1,117 @@
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.Core.Adapter;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Serializer;
using MQTTnet.Core.Server;
namespace MQTTnet.Implementations
{
public class MqttServerAdapter : IMqttServerAdapter, IDisposable
{
private CancellationTokenSource _cancellationTokenSource;
private Socket _defaultEndpointSocket;
private Socket _sslEndpointSocket;
private X509Certificate2 _sslCertificate;
private bool _isRunning;
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
public void Start(MqttServerOptions options)
{
if (_isRunning) throw new InvalidOperationException("Server is already started.");
_isRunning = true;
_cancellationTokenSource = new CancellationTokenSource();
if (options.DefaultEndpointOptions.IsEnabled)
{
_defaultEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
_defaultEndpointSocket.Bind(new IPEndPoint(IPAddress.Any, options.GetDefaultEndpointPort()));
_defaultEndpointSocket.Listen(options.ConnectionBacklog);
Task.Run(() => AcceptDefaultEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
}
if (options.SslEndpointOptions.IsEnabled)
{
if (options.SslEndpointOptions.Certificate == null)
{
throw new ArgumentException("SSL certificate is not set.");
}
_sslCertificate = new X509Certificate2(options.SslEndpointOptions.Certificate);
_sslEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
_sslEndpointSocket.Bind(new IPEndPoint(IPAddress.Any, options.GetSslEndpointPort()));
_sslEndpointSocket.Listen(options.ConnectionBacklog);
Task.Run(() => AcceptSslEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
}
}
public void Stop()
{
_isRunning = false;
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
_defaultEndpointSocket?.Dispose();
_defaultEndpointSocket = null;
_sslEndpointSocket?.Dispose();
_sslEndpointSocket = null;
}
public void Dispose()
{
Stop();
}
private async Task AcceptDefaultEndpointConnectionsAsync(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
try
{
var clientSocket = await _defaultEndpointSocket.AcceptAsync();
var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new DefaultMqttV311PacketSerializer());
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
}
catch (Exception exception)
{
MqttTrace.Error(nameof(MqttServerAdapter), exception, "Error while acceping connection at default endpoint.");
}
}
}
private async Task AcceptSslEndpointConnectionsAsync(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
try
{
var clientSocket = await _defaultEndpointSocket.AcceptAsync();
var sslStream = new SslStream(new NetworkStream(clientSocket));
await sslStream.AuthenticateAsServerAsync(_sslCertificate, false, SslProtocols.Tls12, false);
var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new DefaultMqttV311PacketSerializer());
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
}
catch (Exception exception)
{
MqttTrace.Error(nameof(MqttServerAdapter), exception, "Error while acceping connection at SSL endpoint.");
}
}
}
}
}
Frameworks/MQTTnet.NetCoreApp/Implementations/MqttTcpChannel.cs
+122
View File
@@ -0,0 +1,122 @@
using System;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Threading.Tasks;
using MQTTnet.Core.Channel;
using MQTTnet.Core.Client;
using MQTTnet.Core.Exceptions;
namespace MQTTnet.Implementations
{
public class MqttTcpChannel : IMqttCommunicationChannel, IDisposable
{
private readonly Socket _socket;
private SslStream _sslStream;
public MqttTcpChannel()
{
_socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
}
public MqttTcpChannel(Socket socket, SslStream sslStream)
{
_socket = socket ?? throw new ArgumentNullException(nameof(socket));
_sslStream = sslStream;
}
public async Task ConnectAsync(MqttClientOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
try
{
await _socket.ConnectAsync(options.Server, options.GetPort());
if (options.SslOptions.UseSsl)
{
_sslStream = new SslStream(new NetworkStream(_socket, true));
await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.SslOptions.CheckCertificateRevocation);
}
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
public Task DisconnectAsync()
{
try
{
_sslStream.Dispose();
_socket.Dispose();
return Task.FromResult(0);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
public Task WriteAsync(byte[] buffer)
{
if (buffer == null) throw new ArgumentNullException(nameof(buffer));
try
{
if (_sslStream != null)
{
return _sslStream.WriteAsync(buffer, 0, buffer.Length);
}
return _socket.SendAsync(new ArraySegment<byte>(buffer), SocketFlags.None);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
public Task ReadAsync(byte[] buffer)
{
if (buffer == null) throw new ArgumentNullException(nameof(buffer));
try
{
if (_sslStream != null)
{
return _sslStream.ReadAsync(buffer, 0, buffer.Length);
}
return _socket.ReceiveAsync(new ArraySegment<byte>(buffer), SocketFlags.None);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
public void Dispose()
{
_socket?.Dispose();
_sslStream?.Dispose();
}
private static X509CertificateCollection LoadCertificates(MqttClientOptions options)
{
var certificates = new X509CertificateCollection();
if (options.SslOptions.Certificates == null)
{
return certificates;
}
foreach (var certificate in options.SslOptions.Certificates)
{
certificates.Add(new X509Certificate(certificate));
}
return certificates;
}
}
}
Frameworks/MQTTnet.NetCoreApp/MqttClientFactory.cs
+2 -4
View File
@@ -1,8 +1,8 @@
using System;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Channel;
using MQTTnet.Core.Client;
using MQTTnet.Core.Serializer;
using MQTTnet.Implementations;
namespace MQTTnet
{
@@ -12,9 +12,7 @@ namespace MQTTnet
{
if (options == null) throw new ArgumentNullException(nameof(options));
return new MqttClient(options,
new MqttChannelCommunicationAdapter(options.UseSSL ? new MqttClientSslChannel() : (IMqttCommunicationChannel) new MqttTcpChannel(),
new DefaultMqttV311PacketSerializer()));
return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new DefaultMqttV311PacketSerializer()));
}
}
}
Frameworks/MQTTnet.NetCoreApp/MqttClientSslChannel.cs
-122
View File
@@ -1,122 +0,0 @@
using System;
using System.IO;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Authentication;
using System.Threading.Tasks;
using MQTTnet.Core.Channel;
using MQTTnet.Core.Client;
using MQTTnet.Core.Exceptions;
namespace MQTTnet
{
/// <summary>
/// Describes an SSL channel to an MQTT server.
/// </summary>
public class MqttClientSslChannel : IMqttCommunicationChannel, IDisposable
{
private readonly Socket _socket;
private SslStream _sslStream;
/// <summary>
/// Creates a new <see cref="MqttClientSslChannel"/>.
/// </summary>
public MqttClientSslChannel()
{
_socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
}
/// <summary>
/// Creates a new <see cref="MqttClientSslChannel"/> with a predefined <paramref name="socket"/>.
/// </summary>
/// <param name="socket"></param>
public MqttClientSslChannel(Socket socket)
{
_socket = socket ?? throw new ArgumentNullException(nameof(socket));
}
/// <summary>
/// Asynchronously connects to the host described in the <see cref="MqttClientOptions"/>.
/// </summary>
/// <param name="options">The <see cref="MqttClientOptions"/> describing the connection.</param>
public async Task ConnectAsync(MqttClientOptions options)
{
try
{
await _socket.ConnectAsync(options.Server, options.Port);
NetworkStream ns = new NetworkStream(_socket, true);
_sslStream = new SslStream(ns);
await _sslStream.AuthenticateAsClientAsync(options.Server, null, SslProtocols.Tls12, false);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
/// <summary>
/// Asynchronously disconnects the client from the server.
/// </summary>
public Task DisconnectAsync()
{
try
{
_socket.Dispose();
return Task.FromResult(0);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
/// <summary>
/// Asynchronously writes a sequence of bytes to the socket.
/// </summary>
/// <param name="buffer">The buffer to write data from.</param>
public Task WriteAsync(byte[] buffer)
{
if (buffer == null)
throw new ArgumentNullException(nameof(buffer));
try
{
return _sslStream.WriteAsync(buffer, 0, buffer.Length);
}
catch (Exception ex)
when (ex is SocketException || ex is IOException)
{
throw new MqttCommunicationException(ex);
}
}
/// <summary>
/// Asynchronously reads a sequence of bytes from the socket.
/// </summary>
/// <param name="buffer">The buffer to write the data into.</param>
public Task ReadAsync(byte[] buffer)
{
try
{
return _sslStream.ReadAsync(buffer, 0, buffer.Length);
}
catch (Exception ex)
when (ex is SocketException || ex is IOException)
{
throw new MqttCommunicationException(ex);
}
}
/// <summary>
/// Releases all resources used by the <see cref="MqttClientSslChannel"/>.
/// </summary>
public void Dispose()
{
_sslStream?.Dispose();
_socket?.Dispose();
}
}
}
Frameworks/MQTTnet.NetCoreApp/MqttServerAdapter.cs
-56
View File
@@ -1,56 +0,0 @@
using System;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Serializer;
using MQTTnet.Core.Server;
namespace MQTTnet
{
public sealed class MqttServerAdapter : IMqttServerAdapter, IDisposable
{
private CancellationTokenSource _cancellationTokenSource;
private Socket _socket;
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
public void Start(MqttServerOptions options)
{
if (_socket != null) throw new InvalidOperationException("Server is already started.");
_cancellationTokenSource = new CancellationTokenSource();
_socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
_socket.Bind(new IPEndPoint(IPAddress.Any, options.Port));
_socket.Listen(options.ConnectionBacklog);
Task.Run(async () => await AcceptConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
}
public void Stop()
{
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
_socket?.Dispose();
_socket = null;
}
public void Dispose()
{
Stop();
}
private async Task AcceptConnectionsAsync(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
var clientSocket = await _socket.AcceptAsync();
var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket), new DefaultMqttV311PacketSerializer());
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
}
}
}
}
Frameworks/MQTTnet.NetCoreApp/MqttServerFactory.cs
+4 -2
View File
@@ -1,6 +1,8 @@
using System;
using System.Collections.Generic;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Server;
using MQTTnet.Implementations;
namespace MQTTnet
{
@@ -9,8 +11,8 @@ namespace MQTTnet
public MqttServer CreateMqttServer(MqttServerOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
return new MqttServer(options, options.UseSSL ? (IMqttServerAdapter)new MqttSslServerAdapter() : new MqttServerAdapter());
return new MqttServer(options, new List<IMqttServerAdapter> { new MqttServerAdapter() });
}
}
}
Frameworks/MQTTnet.NetCoreApp/MqttServerSslChannel.cs
-123
View File
@@ -1,123 +0,0 @@
using System;
using System.IO;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Threading.Tasks;
using MQTTnet.Core.Channel;
using MQTTnet.Core.Client;
using MQTTnet.Core.Exceptions;
namespace MQTTnet
{
/// <summary>
/// Describes an SSL channel to a client.
/// </summary>
public class MqttServerSslChannel : IMqttCommunicationChannel, IDisposable
{
private readonly Socket _socket;
private readonly SslStream _sslStream;
private readonly X509Certificate2 _cert;
/// <summary>
/// Creates a new <see cref="MqttClientSslChannel"/> with a predefined <paramref name="socket"/>.
/// </summary>
/// <param name="socket">The client socket.</param>
/// <param name="cert">The X509 certificate used to authenticate as a server.</param>
public MqttServerSslChannel(Socket socket, X509Certificate2 cert)
{
_socket = socket ?? throw new ArgumentNullException(nameof(socket));
_cert = cert ?? throw new ArgumentNullException(nameof(cert));
if (!_socket.Connected)
return;
NetworkStream ns = new NetworkStream(_socket, true);
_sslStream = new SslStream(ns);
}
public Task Authenticate()
{
return _sslStream.AuthenticateAsServerAsync(_cert, false, SslProtocols.Tls12, false);
}
/// <summary>
/// Asynchronously connects to the client described in the <see cref="MqttClientOptions"/>.
/// </summary>
/// <param name="options">The <see cref="MqttClientOptions"/> describing the connection.</param>
public Task ConnectAsync(MqttClientOptions options)
{
try
{
return _socket.ConnectAsync(options.Server, options.Port);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
/// <summary>
/// Asynchronously disconnects the client from the server.
/// </summary>
public Task DisconnectAsync()
{
try
{
_socket.Dispose();
return Task.FromResult(0);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
/// <summary>
/// Asynchronously writes a sequence of bytes to the socket.
/// </summary>
/// <param name="buffer">The buffer to write data from.</param>
public Task WriteAsync(byte[] buffer)
{
if (buffer == null)
throw new ArgumentNullException(nameof(buffer));
try
{
return _sslStream.WriteAsync(buffer, 0, buffer.Length);
}
catch (Exception ex)
when (ex is SocketException || ex is IOException)
{
throw new MqttCommunicationException(ex);
}
}
/// <summary>
/// Asynchronously reads a sequence of bytes from the socket.
/// </summary>
/// <param name="buffer">The buffer to write the data into.</param>
public Task ReadAsync(byte[] buffer)
{
try
{
return _sslStream.ReadAsync(buffer, 0, buffer.Length);
}
catch (Exception ex)
when (ex is SocketException || ex is IOException)
{
throw new MqttCommunicationException(ex);
}
}
/// <summary>
/// Releases all resources used by the <see cref="MqttClientSslChannel"/>.
/// </summary>
public void Dispose()
{
_sslStream?.Dispose();
_socket?.Dispose();
}
}
}
Frameworks/MQTTnet.NetCoreApp/MqttSslServerAdapter.cs
-64
View File
@@ -1,64 +0,0 @@
using System;
using System.Net;
using System.Net.Sockets;
using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Serializer;
using MQTTnet.Core.Server;
namespace MQTTnet
{
public class MqttSslServerAdapter : IMqttServerAdapter, IDisposable
{
private CancellationTokenSource _cancellationTokenSource;
private Socket _socket;
private X509Certificate2 _x590Certificate2;
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
public void Start(MqttServerOptions options)
{
if (_socket != null) throw new InvalidOperationException("Server is already started.");
_cancellationTokenSource = new CancellationTokenSource();
_socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
_socket.Bind(new IPEndPoint(IPAddress.Any, options.Port));
_socket.Listen(options.ConnectionBacklog);
Task.Run(async () => await AcceptConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
_x590Certificate2 = new X509Certificate2(options.CertificatePath);
}
public void Stop()
{
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
_socket?.Dispose();
_socket = null;
}
public void Dispose()
{
Stop();
}
private async Task AcceptConnectionsAsync(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
var clientSocket = await _socket.AcceptAsync();
MqttServerSslChannel mssc = new MqttServerSslChannel(clientSocket, _x590Certificate2);
await mssc.Authenticate();
var clientAdapter = new MqttChannelCommunicationAdapter(mssc, new DefaultMqttV311PacketSerializer());
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
}
}
}
}
Frameworks/MQTTnet.NetCoreApp/MqttTcpChannel.cs
-81
View File
@@ -1,81 +0,0 @@
using System;
using System.Net.Sockets;
using System.Threading.Tasks;
using MQTTnet.Core.Channel;
using MQTTnet.Core.Client;
using MQTTnet.Core.Exceptions;
namespace MQTTnet
{
public class MqttTcpChannel : IMqttCommunicationChannel, IDisposable
{
private readonly Socket _socket;
public MqttTcpChannel()
{
_socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
}
public MqttTcpChannel(Socket socket)
{
_socket = socket ?? throw new ArgumentNullException(nameof(socket));
}
public Task ConnectAsync(MqttClientOptions options)
{
try
{
return _socket.ConnectAsync(options.Server, options.Port);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
public Task DisconnectAsync()
{
try
{
_socket.Dispose();
return Task.FromResult(0);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
public Task WriteAsync(byte[] buffer)
{
if (buffer == null) throw new ArgumentNullException(nameof(buffer));
try
{
return _socket.SendAsync(new ArraySegment<byte>(buffer), SocketFlags.None);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
public Task ReadAsync(byte[] buffer)
{
try
{
var buffer2 = new ArraySegment<byte>(buffer);
return _socket.ReceiveAsync(buffer2, SocketFlags.None);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}
public void Dispose()
{
_socket?.Dispose();
}
}
}