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

Merge pull request #70 from JanEggers/msExtensions

Browse Source 
initial version of ms.extensions.logging and di
This commit is contained in:
Christian
2017-10-23 18:44:17 +02:00
committed by GitHub
parent 84a647e712 56a86b6127
commit 3529724bdc
51 changed files with 927 additions and 538 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Frameworks/MQTTnet.NetStandard/Implementations/MqttCommunicationAdapterFactory.cs
-30
View File
@@ -1,30 +0,0 @@
using System;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Client;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Serializer;
namespace MQTTnet.Implementations
{
public class MqttCommunicationAdapterFactory : IMqttCommunicationAdapterFactory
{
public IMqttCommunicationAdapter CreateMqttCommunicationAdapter(IMqttClientOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (options is MqttClientTcpOptions tcpOptions)
{
var trace = new MqttNetTrace();
return new MqttChannelCommunicationAdapter(new MqttTcpChannel(tcpOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion }, trace);
}
if (options is MqttClientWebSocketOptions webSocketOptions)
{
var trace = new MqttNetTrace();
return new MqttChannelCommunicationAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion }, trace);
}
throw new NotSupportedException();
}
}
}
Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs
+13 -10
View File
@@ -7,23 +7,26 @@ 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;
using Microsoft.Extensions.Logging;
using MQTTnet.Core.Client;
namespace MQTTnet.Implementations
{
public class MqttServerAdapter : IMqttServerAdapter, IDisposable
{
private readonly MqttNetTrace _trace;
private readonly ILogger<MqttServerAdapter> _logger;
private readonly IMqttCommunicationAdapterFactory _mqttCommunicationAdapterFactory;
private CancellationTokenSource _cancellationTokenSource;
private Socket _defaultEndpointSocket;
private Socket _tlsEndpointSocket;
private X509Certificate2 _tlsCertificate;
public MqttServerAdapter(MqttNetTrace trace)
public MqttServerAdapter(ILogger<MqttServerAdapter> logger, IMqttCommunicationAdapterFactory mqttCommunicationAdapterFactory)
{
_trace = trace ?? throw new ArgumentNullException(nameof(trace));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_mqttCommunicationAdapterFactory = mqttCommunicationAdapterFactory ?? throw new ArgumentNullException(nameof(mqttCommunicationAdapterFactory));
}
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
@@ -95,17 +98,17 @@ namespace MQTTnet.Implementations
try
{
//todo: else branch can be used with min dependency NET46
#if NET45
#if NET451
var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null).ConfigureAwait(false);
#else
var clientSocket = await _defaultEndpointSocket.AcceptAsync().ConfigureAwait(false);
#endif
var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer(), _trace);
var clientAdapter = _mqttCommunicationAdapterFactory.CreateServerMqttCommunicationAdapter(new MqttTcpChannel(clientSocket, null));
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
}
catch (Exception exception)
{
_trace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at default endpoint.");
_logger.LogError(new EventId(), exception, "Error while accepting connection at default endpoint.");
//excessive CPU consumed if in endless loop of socket errors
await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
@@ -119,7 +122,7 @@ namespace MQTTnet.Implementations
{
try
{
#if NET45
#if NET451
var clientSocket = await Task.Factory.FromAsync(_tlsEndpointSocket.BeginAccept, _tlsEndpointSocket.EndAccept, null).ConfigureAwait(false);
#else
var clientSocket = await _tlsEndpointSocket.AcceptAsync().ConfigureAwait(false);
@@ -128,12 +131,12 @@ namespace MQTTnet.Implementations
var sslStream = new SslStream(new NetworkStream(clientSocket));
await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false);
var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer(), _trace);
var clientAdapter = _mqttCommunicationAdapterFactory.CreateServerMqttCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream));
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
}
catch (Exception exception)
{
_trace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at TLS endpoint.");
_logger.LogError(new EventId(), exception, "Error while accepting connection at TLS endpoint.");
//excessive CPU consumed if in endless loop of socket errors
await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs
+1 -1
View File
@@ -59,7 +59,7 @@ namespace MQTTnet.Implementations
}
//todo: else brach can be used with min dependency NET46
#if NET45
#if NET451
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);