diff --git a/Source/MQTTnet.Server/MQTTnet.Server.csproj b/Source/MQTTnet.Server/MQTTnet.Server.csproj index f142e237..3e5d224b 100644 --- a/Source/MQTTnet.Server/MQTTnet.Server.csproj +++ b/Source/MQTTnet.Server/MQTTnet.Server.csproj @@ -14,6 +14,7 @@ false false latest + 3.3 @@ -26,10 +27,23 @@ 1701;1702 + + + + Always + + Always + + + Always + + + Always + @@ -37,6 +51,8 @@ + + @@ -58,4 +74,10 @@ + + + Always + + + diff --git a/Source/MQTTnet.Server/Startup.cs b/Source/MQTTnet.Server/Startup.cs index c4eb45a9..bdcc7aaf 100644 --- a/Source/MQTTnet.Server/Startup.cs +++ b/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(); diff --git a/Source/MQTTnet.Server/package.json b/Source/MQTTnet.Server/package.json new file mode 100644 index 00000000..5b2909f1 --- /dev/null +++ b/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" + } +} diff --git a/Source/MQTTnet.Server/tsconfig.json b/Source/MQTTnet.Server/tsconfig.json new file mode 100644 index 00000000..5794ac5b --- /dev/null +++ b/Source/MQTTnet.Server/tsconfig.json @@ -0,0 +1,12 @@ +{ + "compilerOptions": { + "noImplicitAny": false, + "noEmitOnError": true, + "removeComments": false, + "sourceMap": true, + "target": "es5" + }, + "exclude": [ + "node_modules" + ] +} diff --git a/Source/MQTTnet.Server/wwwroot/Index.html b/Source/MQTTnet.Server/wwwroot/Index.html new file mode 100644 index 00000000..5ee25faf --- /dev/null +++ b/Source/MQTTnet.Server/wwwroot/Index.html @@ -0,0 +1,35 @@ + + + + + MQTTnet Server test + + + + + +
+

+ + + +
+ +
+ +
+ + + \ No newline at end of file diff --git a/Source/MQTTnet.Server/wwwroot/app/app.ts b/Source/MQTTnet.Server/wwwroot/app/app.ts new file mode 100644 index 00000000..b7167701 --- /dev/null +++ b/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 = document.getElementById("topic"); +var msgInput = document.getElementById("msg"); +var stateParagraph = document.getElementById("state"); +var msgsList = 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]); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 45b99c88..a09f6b53 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -510,25 +510,27 @@ namespace MQTTnet.Client } else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) { - await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false); - - await SendAsync(new MqttPubAckPacket + if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false)) { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubAckReasonCode.Success - }, cancellationToken).ConfigureAwait(false); + await SendAsync(new MqttPubAckPacket + { + PacketIdentifier = publishPacket.PacketIdentifier, + ReasonCode = MqttPubAckReasonCode.Success + }, cancellationToken).ConfigureAwait(false); + } } else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) { - await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false); - - var pubRecPacket = new MqttPubRecPacket + if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false)) { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubRecReasonCode.Success - }; + var pubRecPacket = new MqttPubRecPacket + { + 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 HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) { var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket); var handler = ApplicationMessageReceivedHandler; if (handler != null) { - return handler.HandleApplicationMessageReceivedAsync( - new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage)); + var eventArgs = 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) diff --git a/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs b/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs index e81c69a9..6fbaa933 100644 --- a/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs +++ b/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs @@ -13,5 +13,7 @@ namespace MQTTnet public string ClientId { get; } public MqttApplicationMessage ApplicationMessage { get; } + + public bool ProcessingFailed { get; set; } } }