enabled clients based on tcp to benefit from new api and added Benchmark for EndtoEnd with new API

Frameworks/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs

@@ -0,0 +1,34 @@
+using System;
+using System.Net;
+using MQTTnet.Adapter;
+using MQTTnet.AspNetCore.Client.Tcp;
+using MQTTnet.Client;
+using MQTTnet.Diagnostics;
+using MQTTnet.Serializer;
+
+namespace MQTTnet.AspNetCore.Client
+{
+    public class MqttClientConnectionContextFactory : IMqttClientAdapterFactory
+    {
+        public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+
+            var serializer = new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion };
+
+            switch (options.ChannelOptions)
+            {
+                case MqttClientTcpOptions tcpOptions:
+                    {
+                        var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort());
+                        var tcpConnection = new TcpConnection(endpoint);
+                        return new MqttConnectionContext(serializer, tcpConnection);
+                    }
+                default:
+                    {
+                        throw new NotSupportedException();
+                    }
+            }
+        }
+    }
+}

Frameworks/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Runtime.InteropServices;
 
-namespace MQTTnet.Benchmarks.Tcp
+namespace MQTTnet.AspNetCore.Client.Tcp
 {
     public static class BufferExtensions
     {

Frameworks/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs

@@ -1,6 +1,6 @@
 using System.IO.Pipelines;
 
-namespace MQTTnet.Benchmarks.Tcp
+namespace MQTTnet.AspNetCore.Client.Tcp
 {
     public class DuplexPipe : IDuplexPipe
     {

Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs

@@ -6,7 +6,7 @@ using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 
-namespace MQTTnet.Benchmarks.Tcp
+namespace MQTTnet.AspNetCore.Client.Tcp
 {
     public class SocketAwaitable : ICriticalNotifyCompletion
     {

Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs

@@ -2,7 +2,7 @@
 using System.IO.Pipelines;
 using System.Net.Sockets;
 
-namespace MQTTnet.Benchmarks.Tcp
+namespace MQTTnet.AspNetCore.Client.Tcp
 {
     public class SocketReceiver
     {

Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs

@@ -5,7 +5,7 @@ using System.Diagnostics;
 using System.IO.Pipelines;
 using System.Net.Sockets;
 
-namespace MQTTnet.Benchmarks.Tcp
+namespace MQTTnet.AspNetCore.Client.Tcp
 {
     public class SocketSender
     {

Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs

@@ -4,27 +4,35 @@ using System.IO.Pipelines;
 using System.Net;
 using System.Net.Sockets;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
+using Microsoft.AspNetCore.Http.Features;
 using MQTTnet.Exceptions;
 
-namespace MQTTnet.Benchmarks.Tcp
+namespace MQTTnet.AspNetCore.Client.Tcp
 {
-    public class TcpConnection
+    public class TcpConnection : ConnectionContext
     {
-        private readonly Socket _socket;
         private volatile bool _aborted;
         private readonly EndPoint _endPoint;
+        private SocketSender _sender;
+        private SocketReceiver _receiver;
+
+        private Socket _socket;
         private IDuplexPipe _application;
-        private IDuplexPipe _transport;
+
-        private readonly SocketSender _sender;
+
-        private readonly SocketReceiver _receiver;
+        public bool IsConnected { get; private set; }
+
+        public override string ConnectionId { get; set; }
+
+        public override IFeatureCollection Features { get; }
+
+        public override IDictionary<object, object> Items { get; set; }
+        public override IDuplexPipe Transport { get; set; }
 
         public TcpConnection(EndPoint endPoint)
         {
-            _socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
             _endPoint = endPoint;
-
-            _sender = new SocketSender(_socket, PipeScheduler.ThreadPool);
-            _receiver = new SocketReceiver(_socket, PipeScheduler.ThreadPool);
         }
 
         public TcpConnection(Socket socket)
@@ -38,29 +46,34 @@ namespace MQTTnet.Benchmarks.Tcp
 
         public Task DisposeAsync()
         {
-            _transport?.Output.Complete();
+            IsConnected = false;
-            _transport?.Input.Complete();
+
+            Transport?.Output.Complete();
+            Transport?.Input.Complete();
 
             _socket?.Dispose();
 
             return Task.CompletedTask;
         }
 
-        public async Task<IDuplexPipe> StartAsync()
+        public async Task StartAsync()
         {
-            if (!_socket.Connected)
+            if (_socket == null)
             {
+                _socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+                _sender = new SocketSender(_socket, PipeScheduler.ThreadPool);
+                _receiver = new SocketReceiver(_socket, PipeScheduler.ThreadPool);
                 await _socket.ConnectAsync(_endPoint);
             }
 
             var pair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default);
 
-            _transport = pair.Transport;
+            Transport = pair.Transport;
             _application = pair.Application;
 
             _ = ExecuteAsync();
 
-            return pair.Transport;
+            IsConnected = true;
         }
 
         private async Task ExecuteAsync()

Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -1,5 +1,6 @@
 using Microsoft.AspNetCore.Connections;
 using MQTTnet.Adapter;
+using MQTTnet.AspNetCore.Client.Tcp;
 using MQTTnet.Packets;
 using MQTTnet.Serializer;
 using System;
@@ -27,6 +28,10 @@ namespace MQTTnet.AspNetCore
 
         public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
         {
+            if (Connection is TcpConnection tcp && !tcp.IsConnected)
+            {
+                return tcp.StartAsync();
+            }
             return Task.CompletedTask;
         }
 
@@ -34,6 +39,7 @@ namespace MQTTnet.AspNetCore
         {
             Connection.Transport.Input.Complete();
             Connection.Transport.Output.Complete();
+
             return Task.CompletedTask;
         }
 

Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs

@@ -76,13 +76,10 @@ namespace MQTTnet.AspNetCore
             }
 
             var bodySlice = copy.Slice(0, bodyLength);
-            using (var body = new MemoryStream(bodySlice.GetArray()))
+            packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(bodySlice.GetArray())));
-            {
+            consumed = bodySlice.End;
-                packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, body));
+            observed = bodySlice.End;
-                consumed = bodySlice.End;
+            return true;
-                observed = bodySlice.End;
-                return true;
-            }
         }
     }
 }

Source/MQTTnet/MqttFactory.cs

@@ -22,6 +22,14 @@ namespace MQTTnet
             return new MqttClient(new MqttClientAdapterFactory(), logger);
         }
 
+        public IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory mqttClientAdapterFactory)
+        {
+            if (logger == null) throw new ArgumentNullException(nameof(logger));
+            if (mqttClientAdapterFactory == null) throw new ArgumentNullException(nameof(mqttClientAdapterFactory));
+
+            return new MqttClient(mqttClientAdapterFactory, logger);
+        }
+
         public IMqttServer CreateMqttServer()
         {
             var logger = new MqttNetLogger();

Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj

@@ -10,9 +10,11 @@
   <ItemGroup>
     <PackageReference Include="BenchmarkDotNet" Version="0.10.14" />
     <PackageReference Include="System.IO.Pipelines" Version="4.5.0" />
+    <PackageReference Include="Microsoft.AspNetCore" Version="2.1.0" />
   </ItemGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\..\Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj" />
     <ProjectReference Include="..\..\Source\MQTTnet\MQTTnet.csproj" />
   </ItemGroup>
 

Tests/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs

@@ -0,0 +1,76 @@
+using BenchmarkDotNet.Attributes;
+using MQTTnet.Client;
+
+
+using MQTTnet.AspNetCore;
+
+using Microsoft.AspNetCore;
+using Microsoft.AspNetCore.Hosting;
+using MQTTnet.Server;
+using MQTTnet.Diagnostics;
+using MQTTnet.AspNetCore.Client;
+
+namespace MQTTnet.Benchmarks
+{
+    [MemoryDiagnoser]
+    public class MessageProcessingMqttConnectionContextBenchmark
+    {
+        private IWebHost _host;
+        private IMqttClient _mqttClient;
+        private MqttApplicationMessage _message;
+
+        [GlobalSetup]
+        public void Setup()
+        {
+            _host = WebHost.CreateDefaultBuilder()
+                   .UseKestrel(o => o.ListenAnyIP(1883, l => l.UseMqtt()))
+                   .ConfigureServices(services => {
+                       var mqttServerOptions = new MqttServerOptionsBuilder()
+                            .WithoutDefaultEndpoint()
+                            .Build();
+                        services
+                            .AddHostedMqttServer(mqttServerOptions)
+                            .AddMqttConnectionHandler();
+                   })
+                   .Configure(app => {
+                       app.UseMqttServer(s => {
+
+                       });
+                   })
+                   .Build();
+
+            var factory = new MqttFactory();
+            _mqttClient = factory.CreateMqttClient(new MqttNetLogger(), new MqttClientConnectionContextFactory());
+
+            _host.StartAsync().GetAwaiter().GetResult();
+
+            var clientOptions = new MqttClientOptionsBuilder()
+                .WithTcpServer("localhost").Build();
+
+            _mqttClient.ConnectAsync(clientOptions).GetAwaiter().GetResult();
+
+            _message = new MqttApplicationMessageBuilder()
+                .WithTopic("A")
+                .Build();
+        }
+
+        [GlobalCleanup]
+        public void Cleanup()
+        {
+            _mqttClient.DisconnectAsync().GetAwaiter().GetResult();
+            _mqttClient.Dispose();
+
+            _host.StopAsync().GetAwaiter().GetResult();
+            _host.Dispose();
+        }
+
+        [Benchmark]
+        public void Send_10000_Messages()
+        {
+            for (var i = 0; i < 10000; i++)
+            {
+                _mqttClient.PublishAsync(_message).GetAwaiter().GetResult();
+            }
+        }
+    }
+}

Tests/MQTTnet.Benchmarks/Program.cs

@@ -16,6 +16,7 @@ namespace MQTTnet.Benchmarks
             Console.WriteLine("5 = ChannelAdapterBenchmark");
             Console.WriteLine("6 = MqttTcpChannelBenchmark");
             Console.WriteLine("7 = TcpPipesBenchmark");
+            Console.WriteLine("8 = MessageProcessingMqttConnectionContextBenchmark");
 
             var pressedKey = Console.ReadKey(true);
             switch (pressedKey.KeyChar)
@@ -41,6 +42,9 @@ namespace MQTTnet.Benchmarks
                 case '7':
                     BenchmarkRunner.Run<TcpPipesBenchmark>();
                     break;
+                case '8':
+                    BenchmarkRunner.Run<MessageProcessingMqttConnectionContextBenchmark>(new AllowNonOptimized());
+                    break;
             }
 
             Console.ReadLine();

Tests/MQTTnet.Benchmarks/TcpPipesBenchmark.cs

@@ -4,7 +4,7 @@ using System.Net.Sockets;
 using System.Threading;
 using System.Threading.Tasks;
 using BenchmarkDotNet.Attributes;
-using MQTTnet.Benchmarks.Tcp;
+using MQTTnet.AspNetCore.Client.Tcp;
 
 namespace MQTTnet.Benchmarks
 {
@@ -25,10 +25,12 @@ namespace MQTTnet.Benchmarks
 
             var clientConnection = new TcpConnection(new IPEndPoint(IPAddress.Loopback, 1883));
 
-            _client = clientConnection.StartAsync().GetAwaiter().GetResult();
+            clientConnection.StartAsync().GetAwaiter().GetResult();
+            _client = clientConnection.Transport;
 
             var serverConnection = new TcpConnection(task.GetAwaiter().GetResult());
-            _server = serverConnection.StartAsync().GetAwaiter().GetResult();
+            serverConnection.StartAsync().GetAwaiter().GetResult();
+            _server = serverConnection.Transport;
         }
 
 

Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj

@@ -10,6 +10,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <PackageReference Include="Microsoft.AspNetCore" Version="2.1.0" />
     <PackageReference Include="Microsoft.AspNetCore.App" Version="2.1.0" />
   </ItemGroup>