Merge branch 'feature/server-project-bootstrap' of https://github.com/chkr1011/MQTTnet into feature/server-project-bootstrap

Source/MQTTnet.Server/MQTTnet.Server.csproj

@@ -14,6 +14,7 @@
     <SignAssembly>false</SignAssembly>
     <DelaySign>false</DelaySign>
     <LangVersion>latest</LangVersion>
+    <TypeScriptToolsVersion>3.3</TypeScriptToolsVersion>
   </PropertyGroup>
 
   <PropertyGroup>
@@ -26,10 +27,23 @@
     <NoWarn>1701;1702</NoWarn>
   </PropertyGroup>
 
+  <ItemGroup>
+    <Content Remove="wwwroot\app\app.ts" />
+  </ItemGroup>
+
   <ItemGroup>
     <Content Update="appsettings.json">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </Content>
+    <Content Update="wwwroot\app\app.js">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </Content>
+    <Content Update="wwwroot\app\app.js.map">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </Content>
+    <Content Update="wwwroot\Index.html">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </Content>
   </ItemGroup>
 
   <ItemGroup>
@@ -37,6 +51,8 @@
     <PackageReference Include="IronPython.StdLib" Version="2.7.9" />
     <PackageReference Include="Microsoft.AspNetCore.App" />
     <PackageReference Include="Microsoft.AspNetCore.Razor.Design" Version="2.2.0" PrivateAssets="All" />
+    <PackageReference Include="Microsoft.AspNetCore.StaticFiles" Version="2.2.0" />
+    <PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.2.1" />
     <PackageReference Include="Swashbuckle.AspNetCore.SwaggerGen" Version="4.0.1" />
     <PackageReference Include="Swashbuckle.AspNetCore.SwaggerUI" Version="4.0.1" />
   </ItemGroup>
@@ -58,4 +74,10 @@
     </None>
   </ItemGroup>
 
+  <ItemGroup>
+    <TypeScriptCompile Include="wwwroot\app\app.ts">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </TypeScriptCompile>
+  </ItemGroup>
+
 </Project>

Source/MQTTnet.Server/Startup.cs

@@ -1,4 +1,6 @@
-using Microsoft.AspNetCore.Builder;
+using System;
+using System.Net.WebSockets;
+using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.Extensions.Configuration;
@@ -29,7 +31,8 @@ namespace MQTTnet.Server
             IHostingEnvironment environment, 
             MqttServerService mqttServerService,
             PythonScriptHostService pythonScriptHostService,
-            DataSharingService dataSharingService)
+            DataSharingService dataSharingService,
+            MqttWebSocketServerAdapter mqttWebSocketServerAdapter)
         {
             if (environment.IsDevelopment())
             {
@@ -40,9 +43,42 @@ namespace MQTTnet.Server
                 application.UseHsts();
             }
 
+            application.UseStaticFiles();
+
+            var webSocketOptions = new WebSocketOptions
+            {
+                KeepAliveInterval = TimeSpan.FromSeconds(120),
+                ReceiveBufferSize = 4 * 1024
+            };
+
+            application.UseWebSockets(webSocketOptions);
+
             application.UseHttpsRedirection();
             application.UseMvc();
 
+            application.Use(async (context, next) =>
+            {
+                if (context.Request.Path == "/mqtt") // TODO: Full path to config.
+                {
+                    if (context.WebSockets.IsWebSocketRequest)
+                    {
+                        using (var webSocket = await context.WebSockets.AcceptWebSocketAsync().ConfigureAwait(false))
+                        {
+                            await mqttWebSocketServerAdapter.RunWebSocketConnectionAsync(webSocket,
+                                $"{context.Connection.RemoteIpAddress}:{context.Connection.RemotePort}").ConfigureAwait(false);
+                        }
+                    }
+                    else
+                    {
+                        context.Response.StatusCode = 400;
+                    }
+                }
+                else
+                {
+                    await next().ConfigureAwait(false);
+                }
+            });
+
             dataSharingService.Configure();
             pythonScriptHostService.Configure();
             mqttServerService.Configure();

Source/MQTTnet.Server/package.json

@@ -0,0 +1,12 @@
+{
+  "version": "1.0.0",
+  "name": "mqtt.test",
+  "private": true,
+  "devDependencies": {},
+  "dependencies": {
+    "@types/node": "^8.0.46",
+    "mqtt": "^2.15.1",
+    "systemjs": "^0.20.19",
+    "typescript": "^2.3.4"
+  }
+}

Source/MQTTnet.Server/tsconfig.json

@@ -0,0 +1,12 @@
+{
+  "compilerOptions": {
+    "noImplicitAny": false,
+    "noEmitOnError": true,
+    "removeComments": false,
+    "sourceMap": true,
+    "target": "es5"
+  },
+  "exclude": [
+    "node_modules"
+  ]
+}

Source/MQTTnet.Server/wwwroot/Index.html

@@ -0,0 +1,35 @@
+<!DOCTYPE html>
+<html>
+<head>
+    <meta charset="utf-8" />
+    <title>MQTTnet Server test</title>
+</head>
+<body>
+<script src="node_modules/systemjs/dist/system.js"></script>
+<script>
+    System.config({
+        packages: {
+            '': {
+                defaultExtension: 'js'
+            }
+        },
+        map: {
+            mqtt: 'node_modules/mqtt/dist/mqtt.min'
+        }
+    });
+    System.import('app/app');
+</script>
+    
+<div>
+    <p id="state"></p>
+    <input id="topic" placeholder="topic" />
+    <input id="msg" placeholder="msg" />
+    <button id="publish">send</button>
+</div>
+
+<div>
+    <ul id="msgs" ></ul>
+</div>
+
+</body>
+</html>

Source/MQTTnet.Server/wwwroot/app/app.ts

@@ -0,0 +1,58 @@
+import { connect } from "mqtt";
+
+var client = connect('ws://' + location.host + '/mqtt',
+    {
+        clientId: "client" + Math.floor(Math.random() * 6) + 1
+    });
+
+window.onbeforeunload = () => {
+    client.end();
+};
+
+var publishButton = document.getElementById("publish");
+var topicInput = <HTMLInputElement>document.getElementById("topic");
+var msgInput = <HTMLInputElement>document.getElementById("msg");
+var stateParagraph = document.getElementById("state");
+var msgsList = <HTMLUListElement>document.getElementById("msgs");
+
+publishButton.onclick = click => {
+    var topic = topicInput.value;
+    var msg = msgInput.value;
+    client.publish(topic, msg);
+};
+
+client.on('connect', () => {
+    client.subscribe('#', { qos: 0 }, (err, granted) => {
+        console.log(err);
+    });
+    client.publish('presence', 'Hello mqtt');
+
+    stateParagraph.innerText = "connected";
+    showMsg("[connect]");
+});
+
+client.on("error", e => {
+    showMsg("error: " + e.message);
+});
+
+client.on("reconnect", () => {
+    stateParagraph.innerText = "reconnecting";
+    showMsg("[reconnect]");
+});
+
+client.on('message', (topic, message) => {
+    showMsg(topic + ": " + message.toString());
+});
+
+function showMsg(msg: string) {
+    //console.log(msg);
+
+    var node = document.createElement("LI");
+    node.appendChild(document.createTextNode(msg));
+
+    msgsList.appendChild(node);
+
+    if (msgsList.childElementCount > 50) {
+        msgsList.removeChild(msgsList.childNodes[0]);
+    }
+}

Source/MQTTnet/Client/MqttClient.cs

@@ -510,25 +510,27 @@ namespace MQTTnet.Client
                 }
                 else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
                 {
-                    await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
+                    if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false))
-
-                    await SendAsync(new MqttPubAckPacket
                     {
-                        PacketIdentifier = publishPacket.PacketIdentifier,
+                        await SendAsync(new MqttPubAckPacket
-                        ReasonCode = MqttPubAckReasonCode.Success
+                        {
-                    }, cancellationToken).ConfigureAwait(false);
+                            PacketIdentifier = publishPacket.PacketIdentifier,
+                            ReasonCode = MqttPubAckReasonCode.Success
+                        }, cancellationToken).ConfigureAwait(false);
+                    }
                 }
                 else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
                 {
-                    await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
+                    if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false))
-
-                    var pubRecPacket = new MqttPubRecPacket
                     {
-                        PacketIdentifier = publishPacket.PacketIdentifier,
+                        var pubRecPacket = new MqttPubRecPacket
-                        ReasonCode = MqttPubRecReasonCode.Success
+                        {
-                    };
+                            PacketIdentifier = publishPacket.PacketIdentifier,
+                            ReasonCode = MqttPubRecReasonCode.Success
+                        };
 
-                    await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false);
+                        await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false);
+                    }                    
                 }
                 else
                 {
@@ -572,18 +574,19 @@ namespace MQTTnet.Client
             return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(pubRecPacket, pubCompPacket);
         }
 
-        private Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
+        private async Task<bool> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
         {
             var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket);
 
             var handler = ApplicationMessageReceivedHandler;
             if (handler != null)
             {
-                return handler.HandleApplicationMessageReceivedAsync(
+                var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage);
-                    new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage));
+                await handler.HandleApplicationMessageReceivedAsync(eventArgs).ConfigureAwait(false);
+                return !eventArgs.ProcessingFailed;
             }
 
-            return Task.FromResult(0);
+            return true;
         }
 
         private static async Task WaitForTaskAsync(Task task, Task sender)

Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs

@@ -13,5 +13,7 @@ namespace MQTTnet
         public string ClientId { get; }
 
         public MqttApplicationMessage ApplicationMessage { get; }
+
+        public bool ProcessingFailed { get; set; }
     }
 }