mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-18 10:26:11 +08:00
Refactor shared library. Extend error messages
This commit is contained in:
+1
-1
@@ -197,7 +197,7 @@ ClientBin/
|
||||
*.dbmdl
|
||||
*.dbproj.schemaview
|
||||
*.jfm
|
||||
*.pfx
|
||||
# *.pfx
|
||||
*.publishsettings
|
||||
orleans.codegen.cs
|
||||
|
||||
|
||||
@@ -36,7 +36,7 @@ namespace MQTTnet.AspNetCore
|
||||
if (webSocket == null) throw new ArgumentNullException(nameof(webSocket));
|
||||
|
||||
var channel = new MqttWebSocketServerChannel(webSocket);
|
||||
var clientAdapter = _mqttCommunicationAdapterFactory.CreateServerMqttCommunicationAdapter(channel);
|
||||
var clientAdapter = _mqttCommunicationAdapterFactory.CreateServerCommunicationAdapter(channel);
|
||||
|
||||
var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(clientAdapter);
|
||||
ClientAccepted?.Invoke(this, eventArgs);
|
||||
|
||||
+10
-9
@@ -1,5 +1,4 @@
|
||||
#if NET451 || NETSTANDARD1_3
|
||||
#else
|
||||
#if WINDOWS_UWP
|
||||
using System;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Adapter;
|
||||
@@ -12,13 +11,13 @@ namespace MQTTnet.Implementations
|
||||
public class MqttServerAdapter : IMqttServerAdapter, IDisposable
|
||||
{
|
||||
private readonly ILogger<MqttServerAdapter> _logger;
|
||||
private readonly IMqttCommunicationAdapterFactory _mqttCommunicationAdapterFactory;
|
||||
private readonly IMqttCommunicationAdapterFactory _communicationAdapterFactory;
|
||||
private StreamSocketListener _defaultEndpointSocket;
|
||||
|
||||
public MqttServerAdapter(ILogger<MqttServerAdapter> logger, IMqttCommunicationAdapterFactory mqttCommunicationAdapterFactory)
|
||||
public MqttServerAdapter(ILogger<MqttServerAdapter> logger, IMqttCommunicationAdapterFactory communicationAdapterFactory)
|
||||
{
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
_mqttCommunicationAdapterFactory = mqttCommunicationAdapterFactory ?? throw new ArgumentNullException(nameof(mqttCommunicationAdapterFactory));
|
||||
_communicationAdapterFactory = communicationAdapterFactory ?? throw new ArgumentNullException(nameof(communicationAdapterFactory));
|
||||
}
|
||||
|
||||
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
|
||||
@@ -32,7 +31,6 @@ namespace MQTTnet.Implementations
|
||||
if (options.DefaultEndpointOptions.IsEnabled)
|
||||
{
|
||||
_defaultEndpointSocket = new StreamSocketListener();
|
||||
_defaultEndpointSocket.Control.NoDelay = true;
|
||||
await _defaultEndpointSocket.BindServiceNameAsync(options.GetDefaultEndpointPort().ToString(), SocketProtectionLevel.PlainSocket);
|
||||
_defaultEndpointSocket.ConnectionReceived += AcceptDefaultEndpointConnectionsAsync;
|
||||
}
|
||||
@@ -45,6 +43,11 @@ namespace MQTTnet.Implementations
|
||||
|
||||
public Task StopAsync()
|
||||
{
|
||||
if (_defaultEndpointSocket != null)
|
||||
{
|
||||
_defaultEndpointSocket.ConnectionReceived -= AcceptDefaultEndpointConnectionsAsync;
|
||||
}
|
||||
|
||||
_defaultEndpointSocket?.Dispose();
|
||||
_defaultEndpointSocket = null;
|
||||
|
||||
@@ -60,9 +63,7 @@ namespace MQTTnet.Implementations
|
||||
{
|
||||
try
|
||||
{
|
||||
args.Socket.Control.NoDelay = true;
|
||||
|
||||
var clientAdapter = _mqttCommunicationAdapterFactory.CreateServerMqttCommunicationAdapter(new MqttTcpChannel(args.Socket));
|
||||
var clientAdapter = _communicationAdapterFactory.CreateServerCommunicationAdapter(new MqttTcpChannel(args.Socket));
|
||||
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
|
||||
}
|
||||
catch (Exception exception)
|
||||
+5
-6
@@ -1,5 +1,4 @@
|
||||
#if NET451 || NETSTANDARD1_3
|
||||
|
||||
using System;
|
||||
using System.Net;
|
||||
using System.Net.Security;
|
||||
@@ -17,17 +16,17 @@ namespace MQTTnet.Implementations
|
||||
public class MqttServerAdapter : IMqttServerAdapter, IDisposable
|
||||
{
|
||||
private readonly ILogger<MqttServerAdapter> _logger;
|
||||
private readonly IMqttCommunicationAdapterFactory _mqttCommunicationAdapterFactory;
|
||||
private readonly IMqttCommunicationAdapterFactory _communicationAdapterFactory;
|
||||
|
||||
private CancellationTokenSource _cancellationTokenSource;
|
||||
private Socket _defaultEndpointSocket;
|
||||
private Socket _tlsEndpointSocket;
|
||||
private X509Certificate2 _tlsCertificate;
|
||||
|
||||
public MqttServerAdapter(ILogger<MqttServerAdapter> logger, IMqttCommunicationAdapterFactory mqttCommunicationAdapterFactory)
|
||||
public MqttServerAdapter(ILogger<MqttServerAdapter> logger, IMqttCommunicationAdapterFactory communicationAdapterFactory)
|
||||
{
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
_mqttCommunicationAdapterFactory = mqttCommunicationAdapterFactory ?? throw new ArgumentNullException(nameof(mqttCommunicationAdapterFactory));
|
||||
_communicationAdapterFactory = communicationAdapterFactory ?? throw new ArgumentNullException(nameof(communicationAdapterFactory));
|
||||
}
|
||||
|
||||
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
|
||||
@@ -104,7 +103,7 @@ namespace MQTTnet.Implementations
|
||||
#else
|
||||
var clientSocket = await _defaultEndpointSocket.AcceptAsync().ConfigureAwait(false);
|
||||
#endif
|
||||
var clientAdapter = _mqttCommunicationAdapterFactory.CreateServerMqttCommunicationAdapter(new MqttTcpChannel(clientSocket, null));
|
||||
var clientAdapter = _communicationAdapterFactory.CreateServerCommunicationAdapter(new MqttTcpChannel(clientSocket, null));
|
||||
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
|
||||
}
|
||||
catch (Exception exception)
|
||||
@@ -132,7 +131,7 @@ namespace MQTTnet.Implementations
|
||||
var sslStream = new SslStream(new NetworkStream(clientSocket));
|
||||
await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false);
|
||||
|
||||
var clientAdapter = _mqttCommunicationAdapterFactory.CreateServerMqttCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream));
|
||||
var clientAdapter = _communicationAdapterFactory.CreateServerCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream));
|
||||
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
|
||||
}
|
||||
catch (Exception exception)
|
||||
+1
-2
@@ -1,5 +1,4 @@
|
||||
#if NET451 || NETSTANDARD1_3
|
||||
#else
|
||||
#if WINDOWS_UWP
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
@@ -63,7 +63,7 @@ namespace MQTTnet.Implementations
|
||||
}
|
||||
}
|
||||
|
||||
await _webSocket.ConnectAsync(new Uri(uri), CancellationToken.None);
|
||||
await _webSocket.ConnectAsync(new Uri(uri), CancellationToken.None).ConfigureAwait(false);
|
||||
RawReceiveStream = new WebSocketStream(_webSocket);
|
||||
}
|
||||
|
||||
|
||||
@@ -45,20 +45,20 @@ namespace MQTTnet
|
||||
return _serviceProvider.GetRequiredService<ILoggerFactory>();
|
||||
}
|
||||
|
||||
public IMqttCommunicationAdapter CreateClientMqttCommunicationAdapter(IMqttClientOptions options)
|
||||
public IMqttCommunicationAdapter CreateClientCommunicationAdapter(IMqttClientOptions options)
|
||||
{
|
||||
var logger = _serviceProvider.GetRequiredService<ILogger<MqttChannelCommunicationAdapter>>();
|
||||
return new MqttChannelCommunicationAdapter(CreateMqttCommunicationChannel(options.ChannelOptions), CreateSerializer(options.ProtocolVersion), logger);
|
||||
return new MqttChannelCommunicationAdapter(CreateCommunicationChannel(options.ChannelOptions), CreateSerializer(options.ProtocolVersion), logger);
|
||||
}
|
||||
|
||||
public IMqttCommunicationAdapter CreateServerMqttCommunicationAdapter(IMqttCommunicationChannel channel)
|
||||
public IMqttCommunicationAdapter CreateServerCommunicationAdapter(IMqttCommunicationChannel channel)
|
||||
{
|
||||
var serializer = _serviceProvider.GetRequiredService<IMqttPacketSerializer>();
|
||||
var logger = _serviceProvider.GetRequiredService<ILogger<MqttChannelCommunicationAdapter>>();
|
||||
return new MqttChannelCommunicationAdapter(channel, serializer, logger);
|
||||
}
|
||||
|
||||
public IMqttCommunicationChannel CreateMqttCommunicationChannel(IMqttClientChannelOptions options)
|
||||
public IMqttCommunicationChannel CreateCommunicationChannel(IMqttClientChannelOptions options)
|
||||
{
|
||||
if (options == null) throw new ArgumentNullException(nameof(options));
|
||||
|
||||
|
||||
@@ -5,8 +5,8 @@ namespace MQTTnet.Core.Adapter
|
||||
{
|
||||
public interface IMqttCommunicationAdapterFactory
|
||||
{
|
||||
IMqttCommunicationAdapter CreateClientMqttCommunicationAdapter(IMqttClientOptions options);
|
||||
IMqttCommunicationAdapter CreateClientCommunicationAdapter(IMqttClientOptions options);
|
||||
|
||||
IMqttCommunicationAdapter CreateServerMqttCommunicationAdapter(IMqttCommunicationChannel channel);
|
||||
IMqttCommunicationAdapter CreateServerCommunicationAdapter(IMqttCommunicationChannel channel);
|
||||
}
|
||||
}
|
||||
@@ -53,7 +53,7 @@ namespace MQTTnet.Core.Client
|
||||
_latestPacketIdentifier = 0;
|
||||
_packetDispatcher.Reset();
|
||||
|
||||
_adapter = _communicationAdapterFactory.CreateClientMqttCommunicationAdapter(options);
|
||||
_adapter = _communicationAdapterFactory.CreateClientCommunicationAdapter(options);
|
||||
|
||||
_scopeHandle = _logger.BeginScope(options.LogId ?? options.ClientId);
|
||||
_logger.LogTrace("Trying to connect with server.");
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.Text;
|
||||
using System.Threading;
|
||||
@@ -13,7 +14,7 @@ namespace MQTTnet.Core.Serializer
|
||||
public sealed class MqttPacketReader : BinaryReader
|
||||
{
|
||||
private readonly ReceivedMqttPacket _receivedMqttPacket;
|
||||
|
||||
|
||||
public MqttPacketReader(ReceivedMqttPacket receivedMqttPacket)
|
||||
: base(receivedMqttPacket.Body, Encoding.UTF8, true)
|
||||
{
|
||||
@@ -70,6 +71,8 @@ namespace MQTTnet.Core.Serializer
|
||||
var multiplier = 1;
|
||||
var value = 0;
|
||||
byte encodedByte;
|
||||
|
||||
var readBytes = new List<int>();
|
||||
do
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested)
|
||||
@@ -77,14 +80,23 @@ namespace MQTTnet.Core.Serializer
|
||||
throw new TaskCanceledException();
|
||||
}
|
||||
|
||||
encodedByte = (byte)stream.ReadByte();
|
||||
value += (encodedByte & 127) * multiplier;
|
||||
var buffer = stream.ReadByte();
|
||||
readBytes.Add(buffer);
|
||||
|
||||
////if (buffer == -1)
|
||||
////{
|
||||
//// break;
|
||||
////}
|
||||
|
||||
encodedByte = (byte)buffer;
|
||||
value += (byte)(encodedByte & 127) * multiplier;
|
||||
multiplier *= 128;
|
||||
if (multiplier > 128 * 128 * 128)
|
||||
{
|
||||
throw new MqttProtocolViolationException("Remaining length is invalid.");
|
||||
throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", readBytes)}).");
|
||||
}
|
||||
} while ((encodedByte & 128) != 0);
|
||||
|
||||
return value;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,12 +13,12 @@ namespace MQTTnet.Core.Tests
|
||||
_adapter = adapter;
|
||||
}
|
||||
|
||||
public IMqttCommunicationAdapter CreateClientMqttCommunicationAdapter(IMqttClientOptions options)
|
||||
public IMqttCommunicationAdapter CreateClientCommunicationAdapter(IMqttClientOptions options)
|
||||
{
|
||||
return _adapter;
|
||||
}
|
||||
|
||||
public IMqttCommunicationAdapter CreateServerMqttCommunicationAdapter(IMqttCommunicationChannel channel)
|
||||
public IMqttCommunicationAdapter CreateServerCommunicationAdapter(IMqttCommunicationChannel channel)
|
||||
{
|
||||
return _adapter;
|
||||
}
|
||||
|
||||
@@ -18,7 +18,7 @@
|
||||
<ProjectTypeGuids>{A5A43C5B-DE2A-4C0C-9213-0A381AF9435A};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}</ProjectTypeGuids>
|
||||
<WindowsXamlEnableOverview>true</WindowsXamlEnableOverview>
|
||||
<PackageCertificateKeyFile>MQTTnet.TestApp.UniversalWindows_TemporaryKey.pfx</PackageCertificateKeyFile>
|
||||
<PackageCertificateThumbprint>13E377A693C923EE9E88EE84B32A4B9881657362</PackageCertificateThumbprint>
|
||||
<PackageCertificateThumbprint>05F70CAF1426E296BE2F7396F0B654070D72B930</PackageCertificateThumbprint>
|
||||
<RuntimeIdentifiers>win10;win10-arm;win10-arm-aot;win10-x86;win10-x86-aot;win10-x64;win10-x64-aot</RuntimeIdentifiers>
|
||||
</PropertyGroup>
|
||||
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Debug|x86'">
|
||||
|
||||
BIN
Binary file not shown.
@@ -27,7 +27,9 @@
|
||||
<TextBox x:Name="Password"></TextBox>
|
||||
<TextBlock>ClientId:</TextBlock>
|
||||
<TextBox x:Name="ClientId"></TextBox>
|
||||
|
||||
<TextBlock>Clean session:</TextBlock>
|
||||
<CheckBox x:Name="CleanSession" IsChecked="True"></CheckBox>
|
||||
|
||||
<StackPanel Orientation="Horizontal">
|
||||
<RadioButton x:Name="UseTcp" IsChecked="True" GroupName="connection">TCP</RadioButton>
|
||||
<RadioButton x:Name="UseWs" GroupName="connection">WS</RadioButton>
|
||||
|
||||
@@ -81,6 +81,8 @@ namespace MQTTnet.TestApp.UniversalWindows
|
||||
Password = Password.Text
|
||||
};
|
||||
|
||||
options.CleanSession = CleanSession.IsChecked == true;
|
||||
|
||||
try
|
||||
{
|
||||
if (_mqttClient != null)
|
||||
@@ -226,6 +228,7 @@ namespace MQTTnet.TestApp.UniversalWindows
|
||||
.WithTcpServer("broker.hivemq.com")
|
||||
.WithCredentials("bud", "%spencer%")
|
||||
.WithTls()
|
||||
.WithCleanSession()
|
||||
.Build();
|
||||
|
||||
await client.ConnectAsync(options);
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
<?xml version="1.0" encoding="utf-8"?>
|
||||
<Package xmlns="http://schemas.microsoft.com/appx/manifest/foundation/windows10" xmlns:mp="http://schemas.microsoft.com/appx/2014/phone/manifest" xmlns:uap="http://schemas.microsoft.com/appx/manifest/uap/windows10" xmlns:uap3="http://schemas.microsoft.com/appx/manifest/uap/windows10/3" IgnorableNamespaces="uap mp uap3">
|
||||
<Identity Name="4fa21172-9128-4e84-8a6d-74b9acde4d58" Publisher="CN=Christian" Version="1.0.0.0" />
|
||||
<Identity Name="4fa21172-9128-4e84-8a6d-74b9acde4d58" Publisher="CN=test" Version="1.0.0.0" />
|
||||
<mp:PhoneIdentity PhoneProductId="4fa21172-9128-4e84-8a6d-74b9acde4d58" PhonePublisherId="00000000-0000-0000-0000-000000000000" />
|
||||
<Properties>
|
||||
<DisplayName>MQTTnet.TestApp.UniversalWindows</DisplayName>
|
||||
<PublisherDisplayName>chris</PublisherDisplayName>
|
||||
<PublisherDisplayName>test</PublisherDisplayName>
|
||||
<Logo>Assets\StoreLogo.png</Logo>
|
||||
</Properties>
|
||||
<Dependencies>
|
||||
|
||||
Reference in New Issue
Block a user