Merge pull request #355 from JanEggers/ConnectionHandlerWithWebsockets

Connection handler with websockets

Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs

@@ -3,6 +3,7 @@ using Microsoft.AspNetCore.Builder;
 using Microsoft.Extensions.DependencyInjection;
 using System.Linq;
 using MQTTnet.Server;
+using System.Collections.Generic;
 
 namespace MQTTnet.AspNetCore
 {
@@ -23,10 +24,7 @@ namespace MQTTnet.AspNetCore
 
                 if (context.Request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues))
                 {
-                    // Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc.
-                    subProtocol = requestedSubProtocolValues
-                        .OrderByDescending(p => p.Length)
-                        .FirstOrDefault(p => p.ToLower().StartsWith("mqtt"));
+                    subProtocol = SelectSubProtocol(requestedSubProtocolValues);
                 }
 
                 var adapter = app.ApplicationServices.GetRequiredService<MqttWebSocketServerAdapter>();
@@ -40,6 +38,14 @@ namespace MQTTnet.AspNetCore
             return app;
         }
 
+        public static string SelectSubProtocol(IList<string> requestedSubProtocolValues)
+        {
+            // Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc.
+            return requestedSubProtocolValues
+                .OrderByDescending(p => p.Length)
+                .FirstOrDefault(p => p.ToLower().StartsWith("mqtt"));
+        }
+
         public static IApplicationBuilder UseMqttServer(this IApplicationBuilder app, Action<IMqttServer> configure)
         {
             var server = app.ApplicationServices.GetRequiredService<IMqttServer>();

Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs

@@ -1,4 +1,5 @@
 using Microsoft.AspNetCore.Connections;
+using Microsoft.AspNetCore.Connections.Features;
 using MQTTnet.Adapter;
 using MQTTnet.Serializer;
 using MQTTnet.Server;
@@ -13,6 +14,14 @@ namespace MQTTnet.AspNetCore
 
         public override async Task OnConnectedAsync(ConnectionContext connection)
         {
+            // required for websocket transport to work
+            var transferFormatFeature = connection.Features.Get<ITransferFormatFeature>();
+            if (transferFormatFeature != null)
+            {
+                transferFormatFeature.ActiveFormat = TransferFormat.Binary;
+            }
+
+
             var serializer = new MqttPacketSerializer();
             using (var adapter = new MqttConnectionContext(serializer, connection))
             {

Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs

@@ -22,13 +22,18 @@ namespace MQTTnet.TestApp.AspNetCore2
                 .Build();
             services
                 .AddHostedMqttServer(mqttServerOptions)
-                .AddMqttConnectionHandler();
+                .AddMqttConnectionHandler()
+                .AddConnections();
         }
 
         // In class _Startup_ of the ASP.NET Core 2.0 project.
         public void Configure(IApplicationBuilder app, IHostingEnvironment env)
         {
-            app.UseMqttEndpoint();
+            app.UseConnections(c => c.MapConnectionHandler<MqttConnectionHandler>("/mqtt", options => {
+                options.WebSockets.SubProtocolSelector = MQTTnet.AspNetCore.ApplicationBuilderExtensions.SelectSubProtocol;
+            }));
+
+            //app.UseMqttEndpoint();
             app.UseMqttServer(server =>
             {
                 server.Started += async (sender, args) =>