https://github.com/chkr1011/MQTTnet/wiki/Server#intercepting-subscriptions

引用自:https://www.cnblogs.com/zhaoqm999/p/13020836.html

FinitelyFailed edited this page on 28 Apr · 4月28日编辑本页失败46 revisions 46个修订版

Preparation 制作方法

Creating a MQTT server is similar to creating a MQTT client. The following code shows the most simple way of creating a new MQTT server with a TCP endpoint which is listening at the default port 1883.

创建 MQTT 服务器类似于创建 MQTT 客户机。下面的代码展示了创建一个新的 MQTT 服务器的最简单的方法，该服务器的 TCP 端点在缺省端口1883处监听。

// Start a MQTT server.
var mqttServer = new MqttFactory().CreateMqttServer();
await mqttServer.StartAsync(new MqttServerOptions());
Console.WriteLine("Press any key to exit.");
Console.ReadLine();
await mqttServer.StopAsync();

Setting several options for the MQTT server is possible by setting the property values of the MqttServerOptions directly or via using the MqttServerOptionsBuilder (which is recommended). The following code shows how to use the MqttServerOptionsBuilder.

为 MQTT 服务器设置几个选项是可能的，可以直接设置 MqttServerOptions 的属性值，也可以通过使用 MqttServerOptionsBuilder (建议使用)设置。下面的代码演示如何使用 MqttServerOptionsBuilder。

// Configure MQTT server.
var optionsBuilder = new MqttServerOptionsBuilder()
    .WithConnectionBacklog(100)
    .WithDefaultEndpointPort(1884);

var mqttServer = new MqttFactory().CreateMqttServer();
await mqttServer.StartAsync(optionsBuilder.Build());

Validating MQTT clients 验证 MQTT 客户机

The following code shows how to validate an incoming MQTT client connection request:

下面的代码展示了如何验证传入的 MQTT 客户端连接请求:

// Setup client validator.
var optionsBuilder = new MqttServerOptionsBuilder()
    .WithConnectionValidator(c =>
{
    if (c.ClientId.Length < 10)
    {
        c.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid;
        return;
    }

    if (c.Username != "mySecretUser")
    {
        c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
        return;
    }

    if (c.Password != "mySecretPassword")
    {
        c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
        return;
    }

    c.ReasonCode = MqttConnectReasonCode.Success;
});

Using a certificate 使用证书

In order to use an encrypted connection a certificate including the private key is required. The following code shows how to start a server using a certificate for encryption:

为了使用加密连接，需要包含私钥的证书。下面的代码演示如何使用加密证书启动服务器:

using System.Reflection;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
...

var currentPath = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location);
var certificate = new X509Certificate2(Path.Combine(currentPath, "certificate.pfx"),"yourPassword", X509KeyStorageFlags.Exportable);

var optionsBuilder = new MqttServerOptionsBuilder()
    .WithoutDefaultEndpoint() // This call disables the default unencrypted endpoint on port 1883
    .WithEncryptedEndpoint()
    .WithEncryptedEndpointPort(config.Port)
    .WithEncryptionCertificate(certificate.Export(X509ContentType.Pfx))
    .WithEncryptionSslProtocol(SslProtocols.Tls12)

But also other overloads getting a valid certificate blob (byte array) can be used.

但是也可以使用获得有效证书 blob (字节数组)的其他重载。

For creating a self-signed certificate for testing the following command can be used (Windows SDK must be installed):

要创建用于测试的自签名证书，可以使用以下命令(必须安装 Windows SDK) :

makecert.exe -sky exchange -r -n "CN=selfsigned.crt" -pe -a sha1 -len 2048 -ss My "test.cer"

2048-ss My“ test.cer”

OpenSSL can also be used to create a self-signed PFX certificate as described here.

OpenSSL 还可用于创建自签名的 PFX 证书，如本文所述。

Example: 例子:

openssl req -x509 -newkey rsa:4096 -keyout key.pem -out cert.pem -days 365
openssl pkcs12 -export -out certificate.pfx -inkey key.pem -in cert.pem

Publishing messages 发布信息

The server is also able to publish MQTT application messages. The object is the same as for the client implementation. Due to the fact that the server is able to publish its own messages it is not required having a loopback connection in the same process.

服务器还能够发布 MQTT 应用程序消息。该对象与客户端实现相同。由于服务器能够发布自己的消息，因此在相同的进程中不需要环回连接。

This allows also running the server in a Windows IoT Core UWP app. This platform has a network isolation which makes it impossible to communicate via localhost etc.

这也允许在 Windows IoT Core UWP 应用程序中运行服务器。这个平台有一个网络隔离，这使得它不可能通过本地主机等通信。

Examples for publishing a message are described at the client section of this Wiki.

发布消息的示例在 Wiki 的客户端部分进行了描述。

Consuming messages 消费信息

The server is also able to process every application message which was published by any client. The event ApplicationMessageReceived will be fired for every processed message. It has the same format as for the client but additionally has the ClientId.

服务器还能够处理任何客户机发布的每个应用程序消息。事件 ApplicationMessageReceived 将为每个处理的消息触发。它具有与客户机相同的格式，但另外还具有 ClientId。

Details for consuming a application messages are described at the client section of this Wiki.

使用应用程序消息的详细信息在这个 Wiki 的客户端部分进行了描述。

Saving retained application messages 保存保留的应用程序消息

The server supports retained MQTT messages. Those messages are kept and send to clients when they connect and subscribe to them. It is also supported to save all retained messages and loading them after the server has started. This required implementing an interface. The following code shows how to serialize retained messages as JSON:

服务器支持保留的 MQTT 消息。当客户端连接并订阅这些消息时，这些消息将被保存并发送给客户端。它还支持保存所有保留的消息，并在服务器启动后加载这些消息。这需要实现一个接口。下面的代码展示了如何将保留的消息序列化为 JSON:

// Setting the options
options.Storage = new RetainedMessageHandler();

// The implementation of the storage:
// This code uses the JSON library "Newtonsoft.Json".
public class RetainedMessageHandler : IMqttServerStorage
{
    private const string Filename = "C:\\MQTT\\RetainedMessages.json";

    public Task SaveRetainedMessagesAsync(IList<MqttApplicationMessage> messages)
    {
        File.WriteAllText(Filename, JsonConvert.SerializeObject(messages));
        return Task.FromResult(0);
    }

    public Task<IList<MqttApplicationMessage>> LoadRetainedMessagesAsync()
    {
        IList<MqttApplicationMessage> retainedMessages;
        if (File.Exists(Filename))
        {
            var json = File.ReadAllText(Filename);
            retainedMessages = JsonConvert.DeserializeObject<List<MqttApplicationMessage>>(json);
        }
        else
        {
            retainedMessages = new List<MqttApplicationMessage>();
        }
            
        return Task.FromResult(retainedMessages);
    }
}

Intercepting application messages 拦截应用程序消息

A custom interceptor can be set at the server options. This interceptor is called for every application message which is received by the server. This allows extending application messages before they are persisted (in case of a retained message) and before being dispatched to subscribers. This allows use cases like adding a time stamp to every application message if the hardware device does not know the time or time zone etc. The following code shows how to use the interceptor:

可以在服务器选项中设置自定义拦截器。对于服务器接收到的每个应用程序消息，都会调用这个拦截器。这允许在持久化应用程序消息(在保留消息的情况下)之前以及发送到订阅者之前扩展应用程序消息。这允许在硬件设备不知道时间或时区等情况下向每个应用程序消息添加时间戳之类的用例。下面的代码展示了如何使用拦截器:

var optionsBuilder = new MqttServerOptionsBuilder()
    .WithApplicationMessageInterceptor(context =>
    {
        if (context.ApplicationMessage.Topic == "my/custom/topic")
        {
            context.ApplicationMessage.Payload = Encoding.UTF8.GetBytes("The server injected payload.");
        }

        // It is possible to disallow the sending of messages for a certain client id like this:
        if (context.ClientId != "Someone")
        {
            context.AcceptPublish = false;
            return;
        }
        // It is also possible to read the payload and extend it. For example by adding a timestamp in a JSON document.
        // This is useful when the IoT device has no own clock and the creation time of the message might be important.
    })
    .Build();

If you want to stop processing an application message completely (like a delete) then the property context.ApplicationMessage.Payload must be set to null.

如果您希望完全停止处理应用程序消息(如 delete) ，则使用属性上下文。应用信息。有效负载必须设置为空。

Intercepting subscriptions 拦截订阅

A custom interceptor can be set to control which topics can be subscribed by a MQTT client. This allows moving private API-Topics to a protected area which is only available for certain clients. The following code shows how to use the subscription interceptor.

可以将自定义拦截器设置为控制 MQTT 客户机可以订阅哪些主题。这允许将私有 api 主题移动到只对某些客户机可用的受保护区域。下面的代码演示如何使用订阅拦截器。

// Protect several topics from being subscribed from every client.
var optionsBuilder = new MqttServerOptionsBuilder()
    .WithSubscriptionInterceptor(context =>
    {
        if (context.TopicFilter.Topic.StartsWith("admin/foo/bar") && context.ClientId != "theAdmin")
        {
            context.AcceptSubscription = false;
        }

        if (context.TopicFilter.Topic.StartsWith("the/secret/stuff") && context.ClientId != "Imperator")
        {
            context.AcceptSubscription = false;
            context.CloseConnection = true;
        }
    })
    .Build();

It is also supported to use an async method instead of a synchronized one like in the above example.

还支持使用异步方法，而不是像上面示例中那样使用同步方法。

Storing data in the session 在会话中存储数据

From version 3.0.6 and up, there is a Dictionary<object, object> called SessionItems. It allows to store custom data in the session and is available in all interceptors:

从版本3.0.6及以上，有一个 Dictionary < object，object > called SessionItems。它允许在会话中存储自定义数据，并且在所有拦截器中都可用:

var optionsBuilder = new MqttServerOptionsBuilder()
.WithConnectionValidator(c => { c.SessionItems.Add("SomeData", true); }
.WithSubscriptionInterceptor(c => { c.SessionItems.Add("YourData", new List<string>{"a", "b"}); }
.WithApplicationMessageInterceptor(c => { c.SessionItems.Add("Test", 123); }

ASP.NET Core Integration

ASP.NET Core 2.0

This library also has support for a WebSocket based server which is integrated into ASP.NET Core 2.0. This functionality requires an additional library called MQTTnet.AspNetCore. After adding this library a MQTT server can be added to a Kestrel HTTP server.

该库还支持一个基于 WebSocket 的服务器，该服务器集成到 ASP.NET Core 2.0中。这个功能需要一个称为 MQTTnet 的附加库。AspNetCore.添加这个库之后，可以将 MQTT 服务器添加到 kestrelhttp 服务器。

// In class _Startup_ of the ASP.NET Core 2.0 project.
public void ConfigureServices(IServiceCollection services)
{
     // This adds a hosted mqtt server to the services
     services.AddHostedMqttServer(builder => builder.WithDefaultEndpointPort(1883));

     // This adds TCP server support based on System.Net.Socket
     services.AddMqttTcpServerAdapter();

     // This adds websocket support
     services.AddMqttWebSocketServerAdapter();
}

public void Configure(IApplicationBuilder app, IHostingEnvironment env)
{
    // This maps the websocket to an MQTT endpoint
    app.UseMqttEndpoint();
    // Other stuff
}

ASP.NET Core 2.1+

MQTTnet.AspNetCore is compatible with the abstractions present in ASP.NET Core 2.0 but it also offers a new TCP transport based on ASP.NET Core 2.1 Microsoft.AspNetCore.Connections.Abstractions. This transport is mutual exclusive with the old TCP transport so you may only add and use one of them. Our benchmark indicates that the new transport is up to 30 times faster.

MQTTnet.与 ASP.NET Core 2.0中的抽象兼容，但是它也提供了一个基于 ASP.NET Core 2.1 Microsoft 的新的 TCP 传输协议。AspNetCore.联系。抽象概念。这种传输与旧的 TCP 传输是相互排斥的，因此您只能添加和使用其中一种传输。我们的基准测试表明，新的传输速度要快30倍。

// In class _Program_ of the ASP.NET Core 2.1 or 2.2 project.
private static IWebHost BuildWebHost(string[] args) =>
    WebHost.CreateDefaultBuilder(args)
        .UseKestrel(o => {
            o.ListenAnyIP(1883, l => l.UseMqtt()); // MQTT pipeline
            o.ListenAnyIP(5000); // Default HTTP pipeline
        })
    .UseStartup<Startup>()
    .Build();

// In class _Startup_ of the ASP.NET Core 2.1 or 2.2 project.
public void ConfigureServices(IServiceCollection services)
{
     //this adds a hosted mqtt server to the services
     services.AddHostedMqttServer(builder => builder.WithDefaultEndpointPort(1883));

     //this adds tcp server support based on Microsoft.AspNetCore.Connections.Abstractions
     services.AddMqttConnectionHandler();

     //this adds websocket support
     services.AddMqttWebSocketServerAdapter();
}

ASP.NET Core 3.0+ (Since MQTT version 3.0.9)

In ASP.NET Core 3.0+, the server can be configured like this. Remember, that the TLS middleware connection is not yet available, so this will only work for WebSocket connections (Check https://github.com/chkr1011/MQTTnet/issues/464).

在 ASP.NET Core 3.0 + 中，服务器可以这样配置。请记住，TLS 中间件连接还不可用，因此这只适用于 WebSocket 连接(请检查 https://github.com/chkr1011/mqttnet/issues/464连接)。

// In class _Program_ of the ASP.NET Core 3.0+ project.
private static IWebHost BuildWebHost(string[] args) =>
    WebHost.CreateDefaultBuilder(args)
        .UseKestrel(o => {
            o.ListenAnyIP(1883, l => l.UseMqtt()); // MQTT pipeline
            o.ListenAnyIP(5000); // Default HTTP pipeline
        })
    .UseStartup<Startup>()
    .Build();

// In class _Startup_ of the ASP.NET Core 3.0+ project.
public void ConfigureServices(IServiceCollection services)
{
    services
        .AddHostedMqttServer(mqttServer => mqttServer.WithoutDefaultEndpoint())
        .AddMqttConnectionHandler()
        .AddConnections();
}

public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
    app.UseEndpoints(endpoints =>
    {
        endpoints.MapMqtt("/mqtt");
    });

    app.UseMqttServer(server =>
    {
        // Todo: Do something with the server
    });
}

Windows IoT Core and UWP localhost loopback addresses 和 UWP 本地主机回送地址

In Windows IoT Core as well as in UWP, loopback connections (127.0.0.1) are not allowed. If you try to connect to a locally running server (broker), this will fail. See Communicating with localhost (loopback) for enable loopback in Windows 10 IoT Core and UWP-apps.

在 Windows IoT Core 以及 UWP 中，loopback 连接(127.0.0.1)是不允许的。如果尝试连接到本地运行的服务器(代理) ，则会失败。请参阅与本地主机通信(loopback)以便在 Windows 10的物联网核心和 UWP-apps 中启用 loopback。

Special notice for using the server project in Android 在 Android 中使用服务器项目的特别注意事项

Under Android, there is an issue with the default bound IP address. So you have to use the actual address of the device. Check the example below.

在 Android 系统下，缺省绑定 IP 地址有一个问题。所以你必须使用设备的实际地址。查看下面的例子。

IPHostEntry ipHostInfo = Dns.GetHostEntry(Dns.GetHostName());
IPAddress ipAddress = ipHostInfo.AddressList[0];

var server = new MqttFactory().CreateMqttServer();
server.StartAsync(new MqttServerOptionsBuilder()
    .WithDefaultEndpointBoundIPAddress(ipAddress)
    .WithDefaultEndpointBoundIPV6Address(IPAddress.None)
    .Build()).GetAwaiter().GetResult();

Accessing the MQTT server in an ASP.NET MVC controller

If we have an ASP.NET Core application that needs to send MQTT messages from an MVC controller, the MqttService singleton needs to be registered with dependency injection. The trick is to have two methods to correctly setup the MQTT part:

如果我们有一个 ASP.NET Core 应用程序需要从一个 MVC 控制器发送 MQTT 消息，那么 MqttService 单例应用程序需要向依赖注入注册。关键在于有两种方法可以正确设置 MQTT 部分:

Have MqttService implement all the interfaces needed to hook with MqttServer (like IMqttServerClientConnectedHandler, IMqttServerApplicationMessageInterceptor, etc.)

让 MqttService 实现与 MqttServer 挂接所需的所有接口(比如 IMqttServerClientConnectedHandler、 IMqttServerApplicationMessageInterceptor 等)
Write a ConfigureMqttServerOptions(AspNetMqttServerOptionsBuilder options) method that sets up the current object as callback for the needed methods:

编写一个 ConfigureMqttServerOptions (aspnetmqttserveroptionsbilder options)方法，将当前对象设置为所需方法的回调:

public void ConfigureMqttServerOptions(AspNetMqttServerOptionsBuilder options)
{
    options.WithConnectionValidator(this);
     options.WithApplicationMessageInterceptor(this);
}

Write a 写一个ConfigureMqttServer(IMqttServer mqtt) that stores the reference to the MQTT server for later use and setup the handlers: 存储对 MQTT 服务器的引用，以供以后使用，并设置处理程序:

public void ConfigureMqttServer(IMqttServer mqtt)
{
    this.mqtt = mqtt;
    mqtt.ClientConnectedHandler = this;
    mqtt.ClientDisconnectedHandler = this;
}

Then, in your Startup class configure and use the service.

然后，在启动类中配置和使用该服务。

In ConfigureServices:

配置服务:

services.AddSingleton<MqttService>();
services.AddHostedMqttServerWithServices(options => {
    var s = options.ServiceProvider.GetRequiredService<MqttService>();
    s.ConfigureMqttServerOptions(options);
});
services.AddMqttConnectionHandler();
services.AddMqttWebSocketServerAdapter();

In Configure:

配置:

app.UseMqttEndpoint();
app.UseMqttServer(server => app.ApplicationServices.GetRequiredService<MqttService().ConfigureMqttServer(server));

原文地址：https://www.cnblogs.com/wx881208/p/14325043.html

MQTTNetServer补充说明