YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-16 09:13:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

Extend configuration.

Browse Source
This commit is contained in:
Christian Kratky
2019-03-11 18:43:39 +01:00
parent cd76f434e3
commit e1799ea6a3
8 changed files with 228 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.Server/Configuration/SettingsModel.cs
+7 -2
View File
@@ -27,12 +27,17 @@
/// <summary>
/// Listen Settings
/// </summary>
public ListenModel EndPoint { get; set; } = new ListenModel();
public TcpEndpointModel TcpEndPoint { get; set; } = new TcpEndpointModel();
/// <summary>
/// Encryption Listen Settings
/// </summary>
public ListenModel EndPointEncryption { get; set; } = new ListenModel();
public TcpEndpointModel EncryptedTcpEndPoint { get; set; } = new TcpEndpointModel();
/// <summary>
/// Settings for the Web Socket endpoint.
/// </summary>
public WebSocketEndpointModel WebSocketEndpoint { get; set; } = new WebSocketEndpointModel();
/// <summary>
/// Set limit for max pending messages per client
Source/MQTTnet.Server/Configuration/ListenModel.cs → Source/MQTTnet.Server/Configuration/TcpEndpointModel.cs
+2 -2
View File
@@ -6,12 +6,12 @@ namespace MQTTnet.Server.Configuration
/// <summary>
/// Listen Entry Settings Model
/// </summary>
public class ListenModel
public class TcpEndpointModel
{
/// <summary>
/// Constructor
/// </summary>
public ListenModel()
public TcpEndpointModel()
{
}
Source/MQTTnet.Server/Configuration/WebSocketEndpointModel.cs
+18
View File
@@ -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<string> AllowedOrigins { get; set; }
}
}
Source/MQTTnet.Server/Mqtt/MqttServerService.cs
+82 -69
View File
@@ -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<IMqttServerAdapter>
{
new MqttTcpServerAdapter(mqttFactory.Logger.CreateChildLogger()),
WebSocketServerAdapter
_webSocketServerAdapter
};
_mqttServer = mqttFactory.CreateMqttServer(adapters);
}
public MqttWebSocketServerAdapter WebSocketServerAdapter { get; }
public void Configure()
{
_pythonScriptHostService.RegisterProxyObject("publish", new Action<PythonDictionary>(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();
}
}
}
Source/MQTTnet.Server/Mqtt/MqttServerStorage.cs
+27
View File
@@ -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<MqttApplicationMessage> messages)
{
return Task.CompletedTask;
}
public Task<IList<MqttApplicationMessage>> LoadRetainedMessagesAsync()
{
return Task.FromResult<IList<MqttApplicationMessage>>(null);
}
}
}
Source/MQTTnet.Server/Startup.cs
+68 -38
View File
@@ -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<PythonIOStream>();
services.AddSingleton<PythonScriptHostService>();
services.AddSingleton<DataSharingService>();
@@ -106,5 +78,63 @@ namespace MQTTnet.Server
services.AddSingleton<MqttSubscriptionInterceptor>();
services.AddSingleton<MqttApplicationMessageInterceptor>();
}
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);
}
});
}
}
}
Source/MQTTnet.Server/appsettings.json
+16 -4
View File
@@ -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": {
Source/MQTTnet/Server/MqttRetainedMessagesManager.cs
+8 -5
View File
@@ -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;
}
}
}
}