diff --git a/Source/MQTTnet.Server/Configuration/SettingsModel.cs b/Source/MQTTnet.Server/Configuration/SettingsModel.cs
index ca1ebb47..42c660e2 100644
--- a/Source/MQTTnet.Server/Configuration/SettingsModel.cs
+++ b/Source/MQTTnet.Server/Configuration/SettingsModel.cs
@@ -27,12 +27,17 @@
///
/// Listen Settings
///
- public ListenModel EndPoint { get; set; } = new ListenModel();
+ public TcpEndpointModel TcpEndPoint { get; set; } = new TcpEndpointModel();
///
/// Encryption Listen Settings
///
- public ListenModel EndPointEncryption { get; set; } = new ListenModel();
+ public TcpEndpointModel EncryptedTcpEndPoint { get; set; } = new TcpEndpointModel();
+
+ ///
+ /// Settings for the Web Socket endpoint.
+ ///
+ public WebSocketEndpointModel WebSocketEndpoint { get; set; } = new WebSocketEndpointModel();
///
/// Set limit for max pending messages per client
diff --git a/Source/MQTTnet.Server/Configuration/ListenModel.cs b/Source/MQTTnet.Server/Configuration/TcpEndpointModel.cs
similarity index 97%
rename from Source/MQTTnet.Server/Configuration/ListenModel.cs
rename to Source/MQTTnet.Server/Configuration/TcpEndpointModel.cs
index 52c5de42..ccec56bb 100644
--- a/Source/MQTTnet.Server/Configuration/ListenModel.cs
+++ b/Source/MQTTnet.Server/Configuration/TcpEndpointModel.cs
@@ -6,12 +6,12 @@ namespace MQTTnet.Server.Configuration
///
/// Listen Entry Settings Model
///
- public class ListenModel
+ public class TcpEndpointModel
{
///
/// Constructor
///
- public ListenModel()
+ public TcpEndpointModel()
{
}
diff --git a/Source/MQTTnet.Server/Configuration/WebSocketEndpointModel.cs b/Source/MQTTnet.Server/Configuration/WebSocketEndpointModel.cs
new file mode 100644
index 00000000..46610792
--- /dev/null
+++ b/Source/MQTTnet.Server/Configuration/WebSocketEndpointModel.cs
@@ -0,0 +1,18 @@
+
+using System.Collections.Generic;
+
+namespace MQTTnet.Server.Configuration
+{
+ public class WebSocketEndpointModel
+ {
+ public bool Enabled { get; set; } = true;
+
+ public string Path { get; set; } = "/mqtt";
+
+ public int ReceiveBufferSize { get; set; } = 4096;
+
+ public int KeepAliveInterval { get; set; } = 120;
+
+ public List AllowedOrigins { get; set; }
+ }
+}
diff --git a/Source/MQTTnet.Server/Mqtt/MqttServerService.cs b/Source/MQTTnet.Server/Mqtt/MqttServerService.cs
index c17b2413..44a93702 100644
--- a/Source/MQTTnet.Server/Mqtt/MqttServerService.cs
+++ b/Source/MQTTnet.Server/Mqtt/MqttServerService.cs
@@ -1,8 +1,11 @@
using System;
using System.Collections.Generic;
using System.Linq;
+using System.Net.WebSockets;
using System.Text;
+using System.Threading.Tasks;
using IronPython.Runtime;
+using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging;
using MQTTnet.Adapter;
using MQTTnet.AspNetCore;
@@ -26,6 +29,7 @@ namespace MQTTnet.Server.Mqtt
private readonly IMqttServer _mqttServer;
private readonly MqttSubscriptionInterceptor _mqttSubscriptionInterceptor;
private readonly PythonScriptHostService _pythonScriptHostService;
+ private readonly MqttWebSocketServerAdapter _webSocketServerAdapter;
public MqttServerService(
Configuration.SettingsModel settings,
@@ -51,97 +55,36 @@ namespace MQTTnet.Server.Mqtt
_pythonScriptHostService = pythonScriptHostService ?? throw new ArgumentNullException(nameof(pythonScriptHostService));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
- WebSocketServerAdapter = new MqttWebSocketServerAdapter(mqttFactory.Logger.CreateChildLogger());
+ _webSocketServerAdapter = new MqttWebSocketServerAdapter(mqttFactory.Logger.CreateChildLogger());
var adapters = new List
{
new MqttTcpServerAdapter(mqttFactory.Logger.CreateChildLogger()),
- WebSocketServerAdapter
+ _webSocketServerAdapter
};
_mqttServer = mqttFactory.CreateMqttServer(adapters);
}
- public MqttWebSocketServerAdapter WebSocketServerAdapter { get; }
-
public void Configure()
{
_pythonScriptHostService.RegisterProxyObject("publish", new Action(Publish));
- var options = new MqttServerOptionsBuilder()
- .WithMaxPendingMessagesPerClient(_settings.MaxPendingMessagesPerClient)
- .WithDefaultCommunicationTimeout(TimeSpan.FromSeconds(_settings.CommunicationTimeout))
- .WithConnectionValidator(_mqttConnectionValidator)
- .WithApplicationMessageInterceptor(_mqttApplicationMessageInterceptor)
- .WithSubscriptionInterceptor(_mqttSubscriptionInterceptor);
-
- // Configure unencrypted connections
- if (_settings.EndPoint.Enabled)
- {
- options.WithDefaultEndpoint();
- if (_settings.EndPoint.TryReadIPv4(out var address4))
- {
- options.WithDefaultEndpointBoundIPAddress(address4);
- }
-
- if (_settings.EndPoint.TryReadIPv6(out var address6))
- {
- options.WithDefaultEndpointBoundIPV6Address(address6);
- }
-
- if (_settings.EndPoint.Port > 0)
- {
- options.WithDefaultEndpointPort(_settings.EndPoint.Port);
- }
- }
- else
- {
- options.WithoutDefaultEndpoint();
- }
-
- // Configure encrypted connections
- if (_settings.EndPointEncryption.Enabled)
- {
- options
- .WithEncryptedEndpoint()
- .WithEncryptionSslProtocol(System.Security.Authentication.SslProtocols.Tls12)
- .WithEncryptionCertificate(_settings.EndPointEncryption.ReadCertificate());
-
- if (_settings.EndPointEncryption.TryReadIPv4(out var address4))
- {
- options.WithEncryptedEndpointBoundIPAddress(address4);
- }
-
- if (_settings.EndPointEncryption.TryReadIPv6(out var address6))
- {
- options.WithEncryptedEndpointBoundIPV6Address(address6);
- }
-
- if (_settings.EndPoint.Port > 0)
- {
- options.WithEncryptedEndpointPort(_settings.EndPointEncryption.Port);
- }
- }
- else
- {
- options.WithoutEncryptedEndpoint();
- }
-
- if (_settings.ConnectionBacklog > 0)
- {
- options.WithConnectionBacklog(_settings.ConnectionBacklog);
- }
-
_mqttServer.ClientConnectedHandler = _mqttClientConnectedHandler;
_mqttServer.ClientDisconnectedHandler = _mqttClientDisconnectedHandler;
_mqttServer.ClientSubscribedTopicHandler = _mqttClientSubscribedTopicHandler;
_mqttServer.ClientUnsubscribedTopicHandler = _mqttClientUnsubscribedTopicHandler;
- _mqttServer.StartAsync(options.Build()).GetAwaiter().GetResult();
+ _mqttServer.StartAsync(CreateMqttServerOptions()).GetAwaiter().GetResult();
_logger.LogInformation("MQTT server started.");
}
+ public Task RunWebSocketConnectionAsync(WebSocket webSocket, HttpContext httpContext)
+ {
+ return _webSocketServerAdapter.RunWebSocketConnectionAsync(webSocket, httpContext);
+ }
+
private void Publish(PythonDictionary parameters)
{
try
@@ -187,5 +130,75 @@ namespace MQTTnet.Server.Mqtt
_logger.LogError(exception, "Error while publishing application message from server.");
}
}
+
+ private IMqttServerOptions CreateMqttServerOptions()
+ {
+ var options = new MqttServerOptionsBuilder()
+ .WithMaxPendingMessagesPerClient(_settings.MaxPendingMessagesPerClient)
+ .WithDefaultCommunicationTimeout(TimeSpan.FromSeconds(_settings.CommunicationTimeout))
+ .WithConnectionValidator(_mqttConnectionValidator)
+ .WithApplicationMessageInterceptor(_mqttApplicationMessageInterceptor)
+ .WithSubscriptionInterceptor(_mqttSubscriptionInterceptor)
+ .WithStorage(new MqttServerStorage(_settings));
+
+ // Configure unencrypted connections
+ if (_settings.TcpEndPoint.Enabled)
+ {
+ options.WithDefaultEndpoint();
+ if (_settings.TcpEndPoint.TryReadIPv4(out var address4))
+ {
+ options.WithDefaultEndpointBoundIPAddress(address4);
+ }
+
+ if (_settings.TcpEndPoint.TryReadIPv6(out var address6))
+ {
+ options.WithDefaultEndpointBoundIPV6Address(address6);
+ }
+
+ if (_settings.TcpEndPoint.Port > 0)
+ {
+ options.WithDefaultEndpointPort(_settings.TcpEndPoint.Port);
+ }
+ }
+ else
+ {
+ options.WithoutDefaultEndpoint();
+ }
+
+ // Configure encrypted connections
+ if (_settings.EncryptedTcpEndPoint.Enabled)
+ {
+ options
+ .WithEncryptedEndpoint()
+ .WithEncryptionSslProtocol(System.Security.Authentication.SslProtocols.Tls12)
+ .WithEncryptionCertificate(_settings.EncryptedTcpEndPoint.ReadCertificate());
+
+ if (_settings.EncryptedTcpEndPoint.TryReadIPv4(out var address4))
+ {
+ options.WithEncryptedEndpointBoundIPAddress(address4);
+ }
+
+ if (_settings.EncryptedTcpEndPoint.TryReadIPv6(out var address6))
+ {
+ options.WithEncryptedEndpointBoundIPV6Address(address6);
+ }
+
+ if (_settings.EncryptedTcpEndPoint.Port > 0)
+ {
+ options.WithEncryptedEndpointPort(_settings.EncryptedTcpEndPoint.Port);
+ }
+ }
+ else
+ {
+ options.WithoutEncryptedEndpoint();
+ }
+
+ if (_settings.ConnectionBacklog > 0)
+ {
+ options.WithConnectionBacklog(_settings.ConnectionBacklog);
+ }
+
+ return options.Build();
+ }
}
}
\ No newline at end of file
diff --git a/Source/MQTTnet.Server/Mqtt/MqttServerStorage.cs b/Source/MQTTnet.Server/Mqtt/MqttServerStorage.cs
new file mode 100644
index 00000000..562d362d
--- /dev/null
+++ b/Source/MQTTnet.Server/Mqtt/MqttServerStorage.cs
@@ -0,0 +1,27 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using MQTTnet.Server.Configuration;
+
+namespace MQTTnet.Server.Mqtt
+{
+ public class MqttServerStorage : IMqttServerStorage
+ {
+ private readonly SettingsModel _settings;
+
+ public MqttServerStorage(SettingsModel settings)
+ {
+ _settings = settings ?? throw new ArgumentNullException(nameof(settings));
+ }
+
+ public Task SaveRetainedMessagesAsync(IList messages)
+ {
+ return Task.CompletedTask;
+ }
+
+ public Task> LoadRetainedMessagesAsync()
+ {
+ return Task.FromResult>(null);
+ }
+ }
+}
diff --git a/Source/MQTTnet.Server/Startup.cs b/Source/MQTTnet.Server/Startup.cs
index 058e7e94..ebdf5cc5 100644
--- a/Source/MQTTnet.Server/Startup.cs
+++ b/Source/MQTTnet.Server/Startup.cs
@@ -2,8 +2,11 @@
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Mvc;
+using Microsoft.EntityFrameworkCore.Internal;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Scripting.Utils;
+using MQTTnet.Server.Configuration;
using MQTTnet.Server.Logging;
using MQTTnet.Server.Mqtt;
using MQTTnet.Server.Scripting;
@@ -21,15 +24,16 @@ namespace MQTTnet.Server
Configuration = builder.Build();
}
-
+
public IConfigurationRoot Configuration { get; }
public void Configure(
- IApplicationBuilder application,
- IHostingEnvironment environment,
+ IApplicationBuilder application,
+ IHostingEnvironment environment,
MqttServerService mqttServerService,
PythonScriptHostService pythonScriptHostService,
- DataSharingService dataSharingService)
+ DataSharingService dataSharingService,
+ SettingsModel settings)
{
if (environment.IsDevelopment())
{
@@ -42,38 +46,10 @@ namespace MQTTnet.Server
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 mqttServerService.WebSocketServerAdapter.RunWebSocketConnectionAsync(webSocket, context).ConfigureAwait(false);
- }
- }
- else
- {
- context.Response.StatusCode = 400;
- }
- }
- else
- {
- await next().ConfigureAwait(false);
- }
- });
+ ConfigureWebSocketEndpoint(application, mqttServerService, settings);
dataSharingService.Configure();
pythonScriptHostService.Configure();
@@ -84,12 +60,8 @@ namespace MQTTnet.Server
{
services.AddMvc().SetCompatibilityVersion(CompatibilityVersion.Version_2_2);
- // Read settings
- var settings = new Configuration.SettingsModel();
- Configuration.Bind("MQTT", settings);
- services.AddSingleton(settings);
+ services.AddSingleton(ReadSettings());
- // Wire up dependencies
services.AddSingleton();
services.AddSingleton();
services.AddSingleton();
@@ -106,5 +78,63 @@ namespace MQTTnet.Server
services.AddSingleton();
services.AddSingleton();
}
+
+ private SettingsModel ReadSettings()
+ {
+ var settings = new Configuration.SettingsModel();
+ Configuration.Bind("MQTT", settings);
+ return settings;
+ }
+
+ private static void ConfigureWebSocketEndpoint(
+ IApplicationBuilder application,
+ MqttServerService mqttServerService,
+ SettingsModel settings)
+ {
+ if (settings?.WebSocketEndpoint?.Enabled != true)
+ {
+ return;
+ }
+
+ if (string.IsNullOrEmpty(settings.WebSocketEndpoint.Path))
+ {
+ return;
+ }
+
+ var webSocketOptions = new WebSocketOptions
+ {
+ KeepAliveInterval = TimeSpan.FromSeconds(settings.WebSocketEndpoint.KeepAliveInterval),
+ ReceiveBufferSize = settings.WebSocketEndpoint.ReceiveBufferSize
+ };
+
+ if (settings.WebSocketEndpoint.AllowedOrigins?.Any() == true)
+ {
+ webSocketOptions.AllowedOrigins.AddRange(settings.WebSocketEndpoint.AllowedOrigins);
+ }
+
+ application.UseWebSockets(webSocketOptions);
+
+ application.Use(async (context, next) =>
+ {
+ if (context.Request.Path == settings.WebSocketEndpoint.Path)
+ {
+ if (context.WebSockets.IsWebSocketRequest)
+ {
+ using (var webSocket = await context.WebSockets.AcceptWebSocketAsync().ConfigureAwait(false))
+ {
+ await mqttServerService.RunWebSocketConnectionAsync(webSocket, context).ConfigureAwait(false);
+ }
+ }
+ else
+ {
+ context.Response.StatusCode = 400;
+ }
+ }
+ else
+ {
+ await next().ConfigureAwait(false);
+ }
+ });
+ }
}
}
\ No newline at end of file
diff --git a/Source/MQTTnet.Server/appsettings.json b/Source/MQTTnet.Server/appsettings.json
index c913f388..c7419e7b 100644
--- a/Source/MQTTnet.Server/appsettings.json
+++ b/Source/MQTTnet.Server/appsettings.json
@@ -16,23 +16,35 @@
localhost - Localhost only
disable - Skip address assignment
*/
- "EndPoint": {
+ "TcpEndPoint": {
"Enabled": true,
"IPv4": "localhost",
"IPv6": "localhost",
"Port": 1883
},
- "EndPointEncryption": {
+ "EncryptedTcpEndPoint": {
"Enabled": false,
"IPv4": "localhost",
"IPv6": "localhost",
"Port": 8883,
"CertificatePath": "/absolute/path/to/pfx"
},
- "CommunicationTimeout": 15, /* In seconds */
+ "WebSocketEndpoint": {
+ "Enabled": true,
+ "Path": "/mqtt",
+ "KeepAliveInterval": 120, // In seconds.
+ "ReceiveBufferSize": 4096,
+ "AllowedOrigins": [] // List of strings with URLs.
+ },
+ "CommunicationTimeout": 15, // In seconds.
"ConnectionBacklog": 0, /* Set 0 to disable */
"EnablePersistentSessions": false,
- "MaxPendingMessagesPerClient": 250
+ "MaxPendingMessagesPerClient": 250,
+ "RetainedApplicationMessages": {
+ "Persist": false,
+ "Filename": "RetainedApplicationMessages.json",
+ "WriteInterval": 10 // In seconds.
+ }
},
"Logging": {
"LogLevel": {
diff --git a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs
index 7fb7b533..f9c5e9c1 100644
--- a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs
+++ b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs
@@ -33,13 +33,16 @@ namespace MQTTnet.Server
try
{
var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync().ConfigureAwait(false);
-
- using (await _messagesLock.WaitAsync().ConfigureAwait(false))
+ if (retainedMessages?.Any() == true)
{
- _messages.Clear();
- foreach (var retainedMessage in retainedMessages)
+ using (await _messagesLock.WaitAsync().ConfigureAwait(false))
{
- _messages[retainedMessage.Topic] = retainedMessage;
+ _messages.Clear();
+
+ foreach (var retainedMessage in retainedMessages)
+ {
+ _messages[retainedMessage.Topic] = retainedMessage;
+ }
}
}
}