From 36c6d8974eac9ed4f04e342229e7d35b6a836471 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Thu, 6 Apr 2023 17:54:36 +0200 Subject: [PATCH] Allow disabling of packet fragmentation (#1709) * Add compatibility option for AWS * Fix UWP build * Fix comment in MqttClientOptions (#1706) * Refactor naming * Fix wrong option value --------- Co-authored-by: Michi --- .github/workflows/ReleaseNotes.md | 1 + Samples/Client/Client_Connection_Samples.cs | 26 +++++++ .../MqttWebSocketServerAdapter.cs | 2 +- .../ChannelAdapterBenchmark.cs | 2 +- .../MQTTnet.Benchmarks/SerializerBenchmark.cs | 2 +- .../ManagedMqttClientOptions.cs | 2 +- .../WebSocket4NetMqttClientAdapterFactory.cs | 12 ++-- ...MqttPacketSerialization_V3_Binary_Tests.cs | 12 +--- .../MqttPacketSerializationHelper.cs | 2 +- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 42 ++++++----- .../Client/Options/MqttClientOptions.cs | 10 ++- .../Options/MqttClientOptionsBuilder.cs | 70 +++++++++++-------- .../MqttClientAdapterFactory.cs | 7 +- .../MqttTcpServerAdapter.Uwp.cs | 2 +- .../Implementations/MqttTcpServerListener.cs | 2 +- 15 files changed, 123 insertions(+), 71 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 22322c50..e84ee186 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -3,6 +3,7 @@ * [Client] Added support for passing MQTT v5 options (User properties etc.) for disconnects. * [Client] An internal exception is no longer caught silently when calling _DisconnectAsync_ to indicate that the disconnect is not clean (BREAKING CHANGE). * [Client] MQTTv5 features are now checked and an exception is thrown if they are used when using protocol version 3.1.1 and lower. These checks can be disabled in client options. (BREAKING CHANGE!). +* [Client] Added support for disabling packet fragmentation (required for i.e. AWS, #1690, thanks to @logicaloud). * [Server] Exposed MQTT v5 sent properties from the affected client in _ClientDisconnectedAsync_ event. * [Server] Fixed wrong client ID passed to _InterceptingUnsubscriptionEventArgs_ (#1631, thanks to @ghord). * [Server] Exposed socket settings for TCP keep alive in TCP options (#1544). diff --git a/Samples/Client/Client_Connection_Samples.cs b/Samples/Client/Client_Connection_Samples.cs index 81808a4e..d74ccb85 100644 --- a/Samples/Client/Client_Connection_Samples.cs +++ b/Samples/Client/Client_Connection_Samples.cs @@ -70,6 +70,32 @@ public static class Client_Connection_Samples await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None); } } + + public static async Task Connect_With_Amazon_AWS() + { + /* + * This sample creates a simple MQTT client and connects to an Amazon Web Services broker. + * + * The broker requires special settings which are set here. + */ + + var mqttFactory = new MqttFactory(); + + using (var mqttClient = mqttFactory.CreateMqttClient()) + { + var mqttClientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("amazon.web.services.broker") + // Disabling packet fragmentation is very important! + .WithoutPacketFragmentation() + .Build(); + + await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); + + Console.WriteLine("The MQTT client is connected."); + + await mqttClient.DisconnectAsync(); + } + } public static async Task Connect_Client_Timeout() { diff --git a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs index a83eb026..0594af78 100644 --- a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs +++ b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs @@ -48,7 +48,7 @@ namespace MQTTnet.AspNetCore var formatter = new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535)); var channel = new MqttWebSocketChannel(webSocket, endpoint, isSecureConnection, clientCertificate); - using (var channelAdapter = new MqttChannelAdapter(channel, formatter, null, _logger)) + using (var channelAdapter = new MqttChannelAdapter(channel, formatter, _logger)) { await clientHandler(channelAdapter).ConfigureAwait(false); } diff --git a/Source/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs b/Source/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs index be2e0eb4..3b6a7d85 100644 --- a/Source/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs @@ -73,7 +73,7 @@ namespace MQTTnet.Benchmarks var channel = new MemoryMqttChannel(_stream); - _channelAdapter = new MqttChannelAdapter(channel, serializer, null, new MqttNetEventLogger()); + _channelAdapter = new MqttChannelAdapter(channel, serializer, new MqttNetEventLogger()); } static byte[] Join(params ArraySegment[] chunks) diff --git a/Source/MQTTnet.Benchmarks/SerializerBenchmark.cs b/Source/MQTTnet.Benchmarks/SerializerBenchmark.cs index eca6771a..de6ed3dc 100644 --- a/Source/MQTTnet.Benchmarks/SerializerBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/SerializerBenchmark.cs @@ -54,7 +54,7 @@ namespace MQTTnet.Benchmarks public void Deserialize_10000_Messages() { var channel = new BenchmarkMqttChannel(_serializedPacket); - var reader = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535)), null, new MqttNetEventLogger()); + var reader = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535)), new MqttNetEventLogger()); for (var i = 0; i < 10000; i++) { diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs index dcd0c119..81bf3b93 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs @@ -24,7 +24,7 @@ namespace MQTTnet.Extensions.ManagedClient /// /// Defines the maximum amount of topic filters which will be sent in a SUBSCRIBE/UNSUBSCRIBE packet. - /// Amazon AWS limits this number to 8. The default is int.MaxValue. + /// Amazon Web Services (AWS) limits this number to 8. The default is int.MaxValue. /// public int MaxTopicFiltersInSubscribeUnsubscribePackets { get; set; } = int.MaxValue; } diff --git a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs index 27359a1a..30138fe0 100644 --- a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs +++ b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs @@ -27,8 +27,10 @@ namespace MQTTnet.Extensions.WebSocket4Net return new MqttChannelAdapter( new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(options.WriterBufferSize, options.WriterBufferSizeMax)), - packetInspector, - logger); + logger) + { + PacketInspector = packetInspector + }; } case MqttClientWebSocketOptions webSocketOptions: @@ -36,8 +38,10 @@ namespace MQTTnet.Extensions.WebSocket4Net return new MqttChannelAdapter( new WebSocket4NetMqttChannel(options, webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(options.WriterBufferSize, options.WriterBufferSizeMax)), - packetInspector, - logger); + logger) + { + PacketInspector = packetInspector + }; } default: diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs index 4cda6c0a..b9260848 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs @@ -566,7 +566,6 @@ namespace MQTTnet.Tests.Formatter using (var adapter = new MqttChannelAdapter( channel, new MqttPacketFormatterAdapter(protocolVersion, new MqttBufferWriter(4096, 65535)), - null, new MqttNetEventLogger())) { var receivedPacket = adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult(); @@ -582,19 +581,12 @@ namespace MQTTnet.Tests.Formatter MqttProtocolVersion DeserializeAndDetectVersion(MqttPacketFormatterAdapter packetFormatterAdapter, byte[] buffer) { var channel = new MemoryMqttChannel(buffer); - var adapter = new MqttChannelAdapter(channel, packetFormatterAdapter, null, new MqttNetEventLogger()); + var adapter = new MqttChannelAdapter(channel, packetFormatterAdapter, new MqttNetEventLogger()); adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult(); return packetFormatterAdapter.ProtocolVersion; } - MqttBufferReader ReaderFactory(byte[] data) - { - var reader = new MqttBufferReader(); - reader.SetBuffer(data, 0, data.Length); - return reader; - } - TPacket Roundtrip(TPacket packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311, MqttBufferWriter bufferWriter = null) where TPacket : MqttPacket { var writer = bufferWriter ?? WriterFactory(); @@ -603,7 +595,7 @@ namespace MQTTnet.Tests.Formatter using (var channel = new MemoryMqttChannel(buffer.Join().ToArray())) { - var adapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(protocolVersion, new MqttBufferWriter(4096, 65535)), null, new MqttNetEventLogger()); + var adapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(protocolVersion, new MqttBufferWriter(4096, 65535)), new MqttNetEventLogger()); return (TPacket)adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult(); } } diff --git a/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs b/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs index 5d93c72a..43d1323d 100644 --- a/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs +++ b/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs @@ -31,7 +31,7 @@ namespace MQTTnet.Tests { var formatterAdapter = new MqttPacketFormatterAdapter(_protocolVersion, new MqttBufferWriter(4096, 65535)); - var adapter = new MqttChannelAdapter(channel, formatterAdapter, null, MqttNetNullLogger.Instance); + var adapter = new MqttChannelAdapter(channel, formatterAdapter, MqttNetNullLogger.Instance); return adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult(); } } diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 47e28b31..92a2e012 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -27,18 +27,15 @@ namespace MQTTnet.Adapter readonly byte[] _fixedHeaderBuffer = new byte[2]; readonly MqttNetSourceLogger _logger; - readonly MqttPacketInspector _packetInspector; - readonly byte[] _singleByteBuffer = new byte[1]; readonly AsyncLock _syncRoot = new AsyncLock(); - Statistics _statistics; // mutable struct, don't make readonly! + Statistics _statistics; // mutable struct, don't make readonly! - public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, MqttPacketInspector packetInspector, IMqttNetLogger logger) + public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, IMqttNetLogger logger) { _channel = channel ?? throw new ArgumentNullException(nameof(channel)); - _packetInspector = packetInspector; PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter)); @@ -64,6 +61,10 @@ namespace MQTTnet.Adapter public MqttPacketFormatterAdapter PacketFormatterAdapter { get; } + public MqttPacketInspector PacketInspector { get; set; } + + public bool AllowPacketFragmentation { get; set; } = true; + public async Task ConnectAsync(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); @@ -126,7 +127,7 @@ namespace MQTTnet.Adapter try { - _packetInspector?.BeginReceivePacket(); + PacketInspector?.BeginReceivePacket(); ReceivedMqttPacket receivedPacket; var receivedPacketTask = ReceiveAsync(cancellationToken); @@ -144,7 +145,7 @@ namespace MQTTnet.Adapter return null; } - _packetInspector?.EndReceivePacket(); + PacketInspector?.EndReceivePacket(); Interlocked.Add(ref _statistics._bytesSent, receivedPacket.TotalLength); @@ -180,7 +181,10 @@ namespace MQTTnet.Adapter return null; } - public void ResetStatistics() => _statistics.Reset(); + public void ResetStatistics() + { + _statistics.Reset(); + } public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken) { @@ -197,18 +201,18 @@ namespace MQTTnet.Adapter try { var packetBuffer = PacketFormatterAdapter.Encode(packet); - _packetInspector?.BeginSendPacket(packetBuffer); + PacketInspector?.BeginSendPacket(packetBuffer); _logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet); - - if (packetBuffer.Payload.Count > 0) + + if (packetBuffer.Payload.Count == 0 || !AllowPacketFragmentation) { - await _channel.WriteAsync(packetBuffer.Packet, false, cancellationToken).ConfigureAwait(false); - await _channel.WriteAsync(packetBuffer.Payload, true, cancellationToken).ConfigureAwait(false); + await _channel.WriteAsync(packetBuffer.Join(), true, cancellationToken).ConfigureAwait(false); } else { - await _channel.WriteAsync(packetBuffer.Packet, true, cancellationToken).ConfigureAwait(false); + await _channel.WriteAsync(packetBuffer.Packet, false, cancellationToken).ConfigureAwait(false); + await _channel.WriteAsync(packetBuffer.Payload, true, cancellationToken).ConfigureAwait(false); } Interlocked.Add(ref _statistics._bytesReceived, packetBuffer.Length); @@ -270,7 +274,7 @@ namespace MQTTnet.Adapter return 0; } - _packetInspector?.FillReceiveBuffer(_singleByteBuffer); + PacketInspector?.FillReceiveBuffer(_singleByteBuffer); encodedByte = _singleByteBuffer[0]; @@ -309,7 +313,7 @@ namespace MQTTnet.Adapter { return ReadFixedHeaderResult.ConnectionClosed; } - + if (cancellationToken.IsCancellationRequested) { return ReadFixedHeaderResult.Canceled; @@ -323,7 +327,7 @@ namespace MQTTnet.Adapter totalBytesRead += bytesRead; } - _packetInspector?.FillReceiveBuffer(buffer); + PacketInspector?.FillReceiveBuffer(buffer); var hasRemainingLength = buffer[1] != 0; if (!hasRemainingLength) @@ -407,7 +411,7 @@ namespace MQTTnet.Adapter bodyOffset += readBytes; } while (bodyOffset < bodyLength); - _packetInspector?.FillReceiveBuffer(body); + PacketInspector?.FillReceiveBuffer(body); var bodySegment = new ArraySegment(body, 0, bodyLength); return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength); @@ -455,7 +459,7 @@ namespace MQTTnet.Adapter throw new MqttCommunicationException(exception); } - private struct Statistics + struct Statistics { public long _bytesReceived; public long _bytesSent; diff --git a/Source/MQTTnet/Client/Options/MqttClientOptions.cs b/Source/MQTTnet/Client/Options/MqttClientOptions.cs index 1101b3e5..f296777f 100644 --- a/Source/MQTTnet/Client/Options/MqttClientOptions.cs +++ b/Source/MQTTnet/Client/Options/MqttClientOptions.cs @@ -24,6 +24,14 @@ namespace MQTTnet.Client /// public string AuthenticationMethod { get; set; } + /// + /// Usually the MQTT packets can be send partially. This is done by using multiple TCP packets + /// or WebSocket frames etc. Unfortunately not all brokers (like Amazon Web Services (AWS)) do support this feature and + /// will close the connection when receiving such packets. If such a service is used this flag must + /// be set to _false_. + /// + public bool AllowPacketFragmentation { get; set; } = true; + public IMqttClientChannelOptions ChannelOptions { get; set; } /// @@ -214,4 +222,4 @@ namespace MQTTnet.Client /// public int WriterBufferSizeMax { get; set; } = 65535; } -} \ No newline at end of file +} diff --git a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs index 86ac9eec..fb17733e 100644 --- a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs @@ -75,10 +75,22 @@ namespace MQTTnet.Client _options.ChannelOptions = (IMqttClientChannelOptions)_tcpOptions ?? _webSocketOptions; MqttClientOptionsValidator.ThrowIfNotSupported(_options); - + return _options; } + /// + /// Usually the MQTT packets can be send partially. This is done by using multiple TCP packets + /// or WebSocket frames etc. Unfortunately not all brokers (like Amazon Web Services (AWS)) do support this feature and + /// will close the connection when receiving such packets. If such a service is used this flag must + /// be set to _true_. + /// + public MqttClientOptionsBuilder WithoutPacketFragmentation() + { + _options.AllowPacketFragmentation = false; + return this; + } + public MqttClientOptionsBuilder WithAuthentication(string method, byte[] data) { _options.AuthenticationMethod = method; @@ -266,7 +278,7 @@ namespace MQTTnet.Client return this; } - + public MqttClientOptionsBuilder WithTcpServer(Action optionsBuilder) { if (optionsBuilder == null) @@ -373,6 +385,18 @@ namespace MQTTnet.Client return this; } + public MqttClientOptionsBuilder WithWillContentType(string willContentType) + { + _options.WillContentType = willContentType; + return this; + } + + public MqttClientOptionsBuilder WithWillCorrelationData(byte[] willCorrelationData) + { + _options.WillCorrelationData = willCorrelationData; + return this; + } + public MqttClientOptionsBuilder WithWillDelayInterval(uint willDelayInterval) { _options.WillDelayInterval = willDelayInterval; @@ -384,27 +408,33 @@ namespace MQTTnet.Client _options.WillPayload = willPayload; return this; } - + public MqttClientOptionsBuilder WithWillPayload(string willPayload) { if (string.IsNullOrEmpty(willPayload)) { return WithWillPayload((byte[])null); } - + _options.WillPayload = Encoding.UTF8.GetBytes(willPayload); return this; } + public MqttClientOptionsBuilder WithWillPayloadFormatIndicator(MqttPayloadFormatIndicator willPayloadFormatIndicator) + { + _options.WillPayloadFormatIndicator = willPayloadFormatIndicator; + return this; + } + public MqttClientOptionsBuilder WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel willQualityOfServiceLevel) { _options.WillQualityOfServiceLevel = willQualityOfServiceLevel; return this; } - public MqttClientOptionsBuilder WithWillTopic(string willTopic) + public MqttClientOptionsBuilder WithWillResponseTopic(string willResponseTopic) { - _options.WillTopic = willTopic; + _options.WillResponseTopic = willResponseTopic; return this; } @@ -413,38 +443,20 @@ namespace MQTTnet.Client _options.WillRetain = willRetain; return this; } - - public MqttClientOptionsBuilder WithWillContentType(string willContentType) + + public MqttClientOptionsBuilder WithWillTopic(string willTopic) { - _options.WillContentType = willContentType; + _options.WillTopic = willTopic; return this; } - - public MqttClientOptionsBuilder WithWillCorrelationData(byte[] willCorrelationData) - { - _options.WillCorrelationData = willCorrelationData; - return this; - } - - public MqttClientOptionsBuilder WithWillResponseTopic(string willResponseTopic) - { - _options.WillResponseTopic = willResponseTopic; - return this; - } - - public MqttClientOptionsBuilder WithWillPayloadFormatIndicator(MqttPayloadFormatIndicator willPayloadFormatIndicator) - { - _options.WillPayloadFormatIndicator = willPayloadFormatIndicator; - return this; - } - + public MqttClientOptionsBuilder WithWillUserProperty(string name, string value) { if (_options.WillUserProperties == null) { _options.WillUserProperties = new List(); } - + _options.WillUserProperties.Add(new MqttUserProperty(name, value)); return this; } diff --git a/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs b/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs index 9d2cbd8b..1749817c 100644 --- a/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs +++ b/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs @@ -40,7 +40,12 @@ namespace MQTTnet.Implementations var bufferWriter = new MqttBufferWriter(options.WriterBufferSize, options.WriterBufferSizeMax); var packetFormatterAdapter = new MqttPacketFormatterAdapter(options.ProtocolVersion, bufferWriter); - return new MqttChannelAdapter(channel, packetFormatterAdapter, packetInspector, logger); + + return new MqttChannelAdapter(channel, packetFormatterAdapter, logger) + { + AllowPacketFragmentation = options.AllowPacketFragmentation, + PacketInspector = packetInspector + }; } } } diff --git a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs index 3efe8409..1a8ee398 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs @@ -95,7 +95,7 @@ namespace MQTTnet.Implementations var packetFormatterAdapter = new MqttPacketFormatterAdapter(bufferWriter); var tcpChannel = new MqttTcpChannel(args.Socket, clientCertificate, _options); - using (var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, null, _rootLogger)) + using (var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, _rootLogger)) { await clientHandler(clientAdapter).ConfigureAwait(false); } diff --git a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs index 60e406d1..d22f4157 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs @@ -229,7 +229,7 @@ namespace MQTTnet.Implementations var bufferWriter = new MqttBufferWriter(_serverOptions.WriterBufferSize, _serverOptions.WriterBufferSizeMax); var packetFormatterAdapter = new MqttPacketFormatterAdapter(bufferWriter); - using (var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, null, _rootLogger)) + using (var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, _rootLogger)) { await clientHandler(clientAdapter).ConfigureAwait(false); }