rmqtt / rmqtt Goto Github PK

View Code? Open in Web Editor NEW

449.0 11.0 56.0 9.24 MB

MQTT Server/MQTT Broker - Scalable Distributed MQTT Message Broker for IoT in the 5G Era

License: MIT License

Rust 99.15% Dockerfile 0.08% Makefile 0.07% Shell 0.71%

broker mqtt iot rust messaging pubsub iot-middleware 5g iot-broker subscriber

rmqtt's Introduction

RMQTT Broker

English | 简体中文

RMQTT broker is a fully open source, highly scalable, highly available distributed MQTT messaging broker for IoT, M2M and mobile applications that can handle millions of concurrent clients on a single service node.

Features

100% Rust safe code;
Based on tokio, ntex , ntex-mqtt;
MQTT v3.1, v3.1.1 and v5.0 protocols support;
- QoS0, QoS1, QoS2 message support;
- Offline message support;
- Retained message support;
- Last Will message support;
Built-in AUTH/ACL;
HTTP AUTH/ACL;
WebHook;
HTTP APIs;
$SYS System Topics;
Store session information;
Store unexpired messages;
Distributed cluster;
Hooks;
TLS support;
WebSocket support;
WebSocket-TLS support;
Shared subscription($share/{group}/topic);
Built-in extensible components;
Extensible plug-in support;
Metrics & Stats;
Rate limit;
Inflight and Queue;
Message resending;
For full list of new features, please read RMQTT Release Notes.

Installation

The RMQTT broker is cross-platform, which supports Linux, Unix, macOS and Windows. It means RMQTT can be deployed on x86_64 architecture servers and ARM devices like Raspberry Pi.

Run RMQTT using Docker

Single node

mkdir -p /app/log/rmqtt
docker run -d --name rmqtt -p 1883:1883 -p 8883:8883 -p 11883:11883 -p 6060:6060 -v /app/log/rmqtt:/var/log/rmqtt  rmqtt/rmqtt:latest

Multi node

  docker run -d --name rmqtt1 -p 1884:1883 -p 8884:8883 -p 11884:11883 -p 6064:6060 -v /app/log/rmqtt/1:/var/log/rmqtt  rmqtt/rmqtt:latest --id 1 --plugins-default-startups "rmqtt-cluster-raft" --node-grpc-addrs "[email protected]:5363" "[email protected]:5363" "[email protected]:5363" --raft-peer-addrs "[email protected]:6003" "[email protected]:6003" "[email protected]:6003"   

  docker run -d --name rmqtt2 -p 1885:1883 -p 8885:8883 -p 11885:11883 -p 6065:6060 -v /app/log/rmqtt/2:/var/log/rmqtt  rmqtt/rmqtt:latest --id 2 --plugins-default-startups "rmqtt-cluster-raft" --node-grpc-addrs "[email protected]:5363" "[email protected]:5363" "[email protected]:5363" --raft-peer-addrs "[email protected]:6003" "[email protected]:6003" "[email protected]:6003"   

  docker run -d --name rmqtt3 -p 1886:1883 -p 8886:8883 -p 11886:11883 -p 6066:6060 -v /app/log/rmqtt/3:/var/log/rmqtt  rmqtt/rmqtt:latest --id 3 --plugins-default-startups "rmqtt-cluster-raft" --node-grpc-addrs "[email protected]:5363" "[email protected]:5363" "[email protected]:5363" --raft-peer-addrs "[email protected]:6003" "[email protected]:6003" "[email protected]:6003"

Node IDs: 1, 2, 3; Node IP Addrs: 172.17.0.3, 172.17.0.4, 172.17.0.5

Create a static cluster by docker-compose

Download docker-compose configuration template
Start docker-compose cluster

docker-compose up -d

View cluster

curl "http://127.0.0.1:6066/api/v1/health/check"

Installing via ZIP Binary Package (Linux、MacOS、Windows)

Get the binary package of the corresponding OS from RMQTT Download page.

Experience

Test

Functional Testing

paho.mqtt.testing(MQTT V3.1.1) client_test.py

client_test.py Test.test_retained_messages [OK]
client_test.py Test.test_zero_length_clientid [OK]
client_test.py Test.will_message_test [OK]
client_test.py Test.test_zero_length_clientid [OK]
client_test.py Test.test_offline_message_queueing [OK]
client_test.py Test.test_overlapping_subscriptions [OK]
client_test.py Test.test_keepalive [OK]
client_test.py Test.test_redelivery_on_reconnect [OK]
client_test.py Test.test_dollar_topics [OK]
client_test.py Test.test_unsubscribe [OK]
client_test.py Test.test_subscribe_failure [OK]
You need to modify the rmqtt-acl.toml configuration and add the following line at the first line: ["deny", "all", "subscribe", ["test/nosubscribe"]],

paho.mqtt.testing(MQTT V5.0) client_test5.py

client_test5.py Test.test_retained_message [OK]
client_test5.py Test.test_will_message [OK]
client_test5.py Test.test_offline_message_queueing [OK]
client_test5.py Test.test_dollar_topics [OK]
client_test5.py Test.test_unsubscribe [OK]
client_test5.py Test.test_session_expiry [OK]
client_test5.py Test.test_shared_subscriptions [OK]
client_test5.py Test.test_basic [OK]
client_test5.py Test.test_overlapping_subscriptions [OK]
client_test5.py Test.test_redelivery_on_reconnect [OK]
client_test5.py Test.test_payload_format [OK]
client_test5.py Test.test_publication_expiry [OK]
client_test5.py Test.test_subscribe_options [OK]
client_test5.py Test.test_assigned_clientid [OK]
client_test5.py Test.test_subscribe_identifiers [OK]
client_test5.py Test.test_request_response [OK]
client_test5.py Test.test_server_topic_alias [OK]
client_test5.py Test.test_client_topic_alias [OK]
client_test5.py Test.test_maximum_packet_size [OK]
client_test5.py Test.test_keepalive [OK]
client_test5.py Test.test_zero_length_clientid [OK]
client_test5.py Test.test_user_properties [OK]
client_test5.py Test.test_flow_control2 [OK]
client_test5.py Test.test_flow_control1 [OK]
client_test5.py Test.test_will_delay [OK]
client_test5.py Test.test_server_keep_alive [OK]
- You need to modify the rmqtt.toml configuration and change max_keepalive to 60.
client_test5.py Test.test_subscribe_failure [OK]
- You need to modify the rmqtt-acl.toml configuration and add the following line at the first line: ["deny", "all", "subscribe", ["test/nosubscribe"]], Modify the test_subscribe_failure() method in client_test5.py by changing 0x80 to 0x87. Because rmqtt returns the error code 0x87, while test_subscribe_failure expects it to return 0x80. UnspecifiedError = 0x80, NotAuthorized = 0x87。

Benchmark Testing

environment

Item	Content
System	x86_64 GNU/Linux	Rocky Linux 9.2 (Blue Onyx)
CPU	Intel(R) Xeon(R) CPU E5-2696 v3 @ 2.30GHz	72(CPU(s)) = 18(Core(s)) * 2(Thread(s) per core) * 2(Socket(s))
Memory	DDR3/2333	128G
Disk		2T
Container	podman	v4.4.1
MQTT Bench	docker.io/rmqtt/rmqtt-bench:latest	v0.1.3
MQTT Broker	docker.io/rmqtt/rmqtt:latest	v0.3.0
Other	MQTT Bench and MQTT Broker coexistence

Connection Concurrency Performance

Item	Single Node	Raft Cluster Mode
Total Concurrent Clients	1,000,000	1,000,000
Connection Handshake Rate	(5500-7000)/second	(5000-7000)/second

Message Throughput Performance

Item	Single Node	Raft Cluster Mode
Subscription Client Count	1,000,000	1,000,000
Publishing Client Count	40	40
Message Throughput Rate	150,000/second	156,000/second

For detailed benchmark test results and information, see documentation.

rmqtt's People

Contributors

Stargazers

Watchers

Forkers

vanlin linecode isgasho niceskylei armersong simon-fu leo-lionni bowmanchow niurenyige muskanmahajan486 tryor lionwang luhaiyong git-try-it sxhxliang gj-zhang daizhen228 eonie kotlin2018 ledao rhello1999 chatgpt1 foxxnuaa phial3 santiagz ychuan pinghe iotlibrary gsxhnd tsingson iq-scm qcjxberin endy-c greenpdx zhanglei jaylinyu sevenhe1205 megatronyy pjlys huangcv darixsamani diablur fiekers fastflair felipefdl aa51513 acproject jdcw martin19838615 yelcartoubi cheemsthx 1185497683 holg werbenhu gongzhengyang lx6317648 fangfsz

rmqtt's Issues

Apple arm release

Darwin arm release

QoS1和QoS2情况下的BUG

操作步骤：

ClientA连接
ClientB连接并订阅QoS1、2主题
ClientB离线（主动）
ClientA发送ClientB所订阅的主体
ClientB尝试连接上线（失败）

Upgrade salvo web server to latest 0.44 and add logging and add webserver configuration

rmqtt uses 0.37.9 salvo and the interface has changed in 0.44.
step one, would be to upgrade rmqtt to use salvo 0.44 api.
step two, is to add logging to http requests.
step three, add configuration for salvo web server in rmqtt config.

I am trying to add static webpages to http-api/api.rs:router function using StaticDir but I am getting 404, not sure where it is looking for the files.

Templates and push

Was checking out the golang examples in templates repo.

there seems to be no push mechanism like SSE or web sockets ?

关于paho.mqtt.testing测试rmqtt

请问官方之前有使用paho.mqtt.testing的测试用例跑一遍rmqtt嘛？

命令执行：python3 client_test.py

2023-08-15 06:10:25.638 WARN rmqtt::broker::v3.26 | [email protected]:1883/192.168.112.1:59802/myclientid2/undefined-1692079825631 Connection Refused, handshake, ack_code: ServiceUnavailable, new_ack_code: V3(ServiceUnavailable), reason: Msg("Handshake try lock failed")

2023-08-15 06:29:37.954 WARN rmqtt::broker::v3.26 | [email protected]:1883/192.168.112.1:61814/myclientid/undefined-1692080977937 Connection Refused, handshake, ack_code: ServiceUnavailable, new_ack_code: V3(ServiceUnavailable), reason: Msg("Keepalive must be greater than 0")

connection error!

/mqtt/subscribe Not Found

I have nodes, broker,clients,subscriptions, routes and publish working, but subscribe returns a 404. I have tried it from a browser using axios and curl. I have used tcpdump to verify the the request on the wire.

I get the error "session does not exist", not sure what this means.

I am using mqttx client and I can subscribe and publish with no problem on port 1883.

http://10.1.42.157:3000/api/mqtt/subscribe
detail | "session does not exist"

curl -i -X POST "http://192.168.42.34:6060/api/v1/mqtt/subscribe" --header 'Content-Type: application/json' -d '{"topic":"testtopic/#","clientid":"6"}'
HTTP/1.1 404 Not Found
content-type: text/html
content-length: 912
date: Wed, 28 Jun 2023 10:15:04 GMT

<title>404: Not Found</title> <style> :root { --bg-color: #fff; --text-color: #222; } body { background: var(--bg-color); color: var(--text-color); text-align: center; } footer{text-align:center;} @media (prefers-color-scheme: dark) { :root { --bg-color: #222; --text-color: #ddd; } a:link { color: red; } a:visited { color: #a8aeff; } a:hover {color: #a8aeff;} a:active {color: #a8aeff;} } </style>

404: Not Found

The requested resource could not be found.

session does not exist

salvo

http://127.0.0.1:9090/mqtt/acl: connection closed before message completed

参考https://github.com/rmqtt/rmqtt/blob/master/docs/zh_CN/auth-http.md，在本地启了个/mqtt/acl服务，从浏览器或者postman都可以调用成功，但是在rmqtt_auth_http中无法调用成功呢？/mqtt/auth也是一样。
2023-08-11 16:28:56.659 ERRO rmqtt_auth_http.244 | error:reqwest::Error { kind: Request, url: Url { scheme: "http", cannot_be_a_base: false, username: "", password: None, host: Some(Ipv4(127.0.0.1)), port: Some(9090), path: "/mqtt/acl", query: None, fragment: None }, source: hyper::Error(IncompleteMessage) }
2023-08-11 16:28:56.660 WARN rmqtt_auth_http.378 | [email protected]:1883/127.0.0.1:63449/mqttx_de420e2a/-1691253670 acl error, Msg("error sending request for url (http://127.0.0.1:9090/mqtt/acl): connection closed before message completed")
2023-08-11 16:29:37.801 ERRO rmqtt_auth_http.244 | error:reqwest::Error { kind: Request, url: Url { scheme: "http", cannot_be_a_base: false, username: "", password: None, host: Some(Ipv4(127.0.0.1)), port: Some(9090), path: "/mqtt/auth", query: None, fragment: None }, source: hyper::Error(IncompleteMessage) }

有k8s的安装示例吗

can describe about how to use it detail

1、build and run
2、install distribution node
3、how to connect it?

Question/Request

Hiya! again amazing server, is there any way of configuring rmqtt as a broker bridge to communicate/proxy messages across brokers?

Thanks!

question: plugin usage

hi there

how can i find and use some plugin such as rmqtt-sys-topic ?

本地搭建集群，验证生产和消费流程。返回错误信息Handshake try lock failed

1：rmqtt配置文件
注：因为只想简单测试，遂只保留了cluster-raft和http-api两部分插件配置。

2：集群服务部署方式
下载了并使用官方中的docker-compose方式，仅修改了日志挂载目录到本地路径。

3：启动后节点健康检查

4：启动消费者订阅；通过官方的http api检查消费者信息

5：启动生产者发布，提示错误（见下两图）

Panic when providing a TLS certificate and a key to a listener.

When i try to run the server with TLS certificates i get an error saying

thread 'main' panicked at rmqtt-bin/src/server.rs:214:53:
removal index (is 0) should be < len (is 0)

seems to happen here

    let key_file = &mut BufReader::new(File::open(listen_cfg.key.as_ref().unwrap())?);


        let cert_chain = certs(cert_file).unwrap();
        let mut keys = rsa_private_keys(key_file).unwrap();


        let mut tls_config = if listen_cfg.cross_certificate {
            let root_chain = cert_chain.clone();
            let mut client_auth_roots = RootCertStore::empty();
            for root in root_chain {
                client_auth_roots.add(&root).unwrap();
            }
            ServerConfig::new(AllowAnyAuthenticatedClient::new(client_auth_roots))
        } else {
            ServerConfig::new(NoClientAuth::new())
        };


    >    tls_config.set_single_cert(cert_chain, keys.remove(0)).map_err(|e| MqttError::from(e.to_string()))?;


        let tls_acceptor = Acceptor::new(tls_config);

Is anyone else having a similar issue ?

The certificate has been validated and checked with openssl and the key has also been checked, it is a 4096 bit rsa key.

你好，我是文强，想认识一下你。

hello，我是文强。一直从事消息领域的工作。最近在研究rust和mq结合的一些可能性，想认识一下你，方便的话加一下微信：loobooya。感谢感谢。

https://time.geekbang.org/column/intro/100552001?utm_campaign=geektime_search&utm_content=geektime_search&utm_medium=geektime_search&utm_source=geektime_search&utm_term=geektime_search

打包的文件大小是否能压缩小的？

tls认证问题

几个问题咨询一下官方
1：目前rmqtt中使用的是单向认证吗？
2：这方面的学习资料是参考resttls吗？
3：是否有考虑实现双向认证？

场景1：我使用golang语言写了一个mqtt的pub和sub的客户端，调用rmqtt的8883端口。
场景2：我使用golang语言写了一个mqtt的pub和sub的客户端，调用rmqtt的8883端口。
注：场景2中，客户端都配置了tls客户端证书；rmqtt也更换了服务端证书（使用openssl生成的client和server证书）
其中场景1和场景2，rmqtt都返回如下信息：

2023-08-25 13:04:40.294 DEBG rustls::server::hs.380 | decided upon suite TLS13_AES_128_GCM_SHA256
2023-08-25 13:04:40.295 DEBG rustls::server::tls13::client_hello.343 | Client unwilling to resume, DHE_KE not offered

麻烦官方空闲的时候看一下，感谢。

Max QOS in Connect ACK Protocol Question ? max_qos_allowed=2 Produce MQTT 5 Protocol Error ?

Hi,

on fast testing I noticed perhaps a little Bug ?

Reading MQTT Spec CONNACK Properties:

https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901084

3.2.2.3.4 Maximum QoS

36 (0x24) Byte, Identifier of the Maximum QoS.

Followed by a Byte with a value of either 0 or 1. It is a Protocol Error to include Maximum QoS more than once, or to have a value other than 0 or 1. If the Maximum QoS is absent, the Client uses a Maximum QoS of 2.

Could it be that the Sevrer in Version 0.3 send the configured max_qos_allowed=2 properties which is a spec validation ?

Reproduced with latest Release , unchanged configuration, and Using a mqtt client which check this violation.

能不能加一下websokect接口

试了下您的项目，觉得真好，只是我平时用是网页创建websokect来接收数据的，不知道能不能给加一下websokect的接口，急切期盼！可惜我不会rust，无法贡献自己的力量

请开通付费赞助

请开通付费赞助, 包括 one-time payment, 以 $ 示支持.

remove unnecessary variable to be mutable

code:after pull origin master
ide:vscode
message:warning: variable does not need to be mutable
image:

topic的问题

在测试时发现，

一个客户端订阅了topic(如：a/b/#) 后，代码改成订阅另一个topic(如A/c/#)
这个客户端的clientid 是指定的，前后两次使用相同的clientid
当客户端订阅topic (A/c/#)时会收到topic（a/b/#）的消息。
如果客户端订阅topic (A/c/#)前unsubscribe( 'a/b/#'),再subscribe( 'A/c/#')，则功能正常。
感觉是订阅的topic被保留到服务器上，每次有相同的clientid连入，不管是否订阅之前的topic，都会推送之前订阅过的消息。

请问有开发mqtt client的计划吗？

我发现rumqtt的client（rumqttc）还不支持native-tls

RMQTT Development Direction Discussion

Thank you for your support and attention to RMQTT. For the future development direction of RMQTT, you can put forward valuable suggestions and opinions according to your specific business needs in the process of use.

thank you all

Publish to crates.io

Publishing to crates.io brings the opportunity to simplify the installation cargo install rmqtt instead of using docker images or pre built releases.

有技术社区吗

这个项目真的很不错，有专门讨论这个项目的技术群吗？

roadmap: where the project go

Firstly, thanks for your awesome job of this project. I haved used it in my own project: fornet.
But I'm wondering where the project go, it seems all things about MQTT protocol have implemented.

您好，感谢您做了如此好用的项目。MQTT 相关的协议都已经实现完毕了，它下一步要如何发展？

Creating Clients

Hello, I cannot understand how the mechanism of authorization through HTTP Auth is implemented.

I did not find in the documentation how to add usernames and passwords for clients, and the plugin description uses the path like http://127.0.0.1:8080/mqtt/superuser, which does not work...

can you tell me how to create users?

Integration

This would work very well with SurrealDB because

it has live queries
It needs a kv Store to make it scale out to many nodes.

Best to just google it and how it uses tikv currrently to scale out.

you can also scale it out by feeding mutations into rmqtt and rmqtt feeds into many SurrealDB. Reads go directly to SurrealDB.

Can't update package on windows and missing authentication.

Great package, and apparently the only package on R to work with mqtt messages on widows.
I'm trying to use its but i need to add an user and password for the host I'm giving to the mqtt_topic_subscribe function, but aparently there isnt that option on the 0.1.0 at least, and I can't update it.
Any idea what could be the reason for not being able to update on windows R studio to the current 0.2.10 ?
I'm using the dtupdate package to try to update this.

Thank you.

Cheers, Luís Pereira.

http-api fails in firefox script => Reason: CORS header ‘Access-Control-Allow-Origin’ missing

I am writing a vue/axios interface to the http-api, rmqtt send the data just fine( status 200), but firefox does not allow the data to script because CORS header ‘Access-Control-Allow-Origin’ missing.

I can turn it off in firefox by about:config about:config -> security.fileuri.strict_origin_policy -> false
But this is a security issue.

I looked at the http-api code but could not find where the headers are set.

How do I add Access-Control-Allow-Origin: *

请教如何基于这个库开发一些功能

小白看了文档感觉有点蒙蒙的，具体来说这个库的应用场景是什么呢？如何集成到现有的Rust项目中去？比如我的tokio consumer要做一个job，完成之后再把结果publish到broker去，broker自动通知producer这个job完成了，这个逻辑用rmqtt该怎样实现呢？另外broker应该怎样部署？每隔一段时间重复执行一个job怎样实现？

怎么样设置通过账号密码才能连接服务端的

集群部署，节点选举出现“Leader ID exception”

动作1：使用官方提供的docker-compose进行部署，并按流程依次操作。
动作2：通过curl "http://127.0.0.1:6066/api/v1/health/check"进行查看。
问题如下图所示：

目前的解决方法：手动停止上图中的“节点3”，然后再重新启动“节点3”。再次重复动作2，“status” 返回OK。

run example err

fisrt run:
cargo run --example basic
then:
cargo run --example client

result:

 (base) ➜  ntex-mqtt git:(master) ✗ cargo run --example client
   Compiling ntex-mqtt v1.0.0 (/Users/lchjczw/work/rust/project/ntex-mqtt)
    Finished dev [unoptimized + debuginfo] target(s) in 2.07s
     Running `target/debug/examples/client`
[2024-02-03T07:19:08Z TRACE ntex_mqtt::v5::client::connector] Connect ack response from server: ConnectAck {
        session_present: false,
        reason_code: Success,
        session_expiry_interval_secs: None,
        receive_max: 15,
        max_qos: AtLeastOnce,
        max_packet_size: None,
        assigned_client_id: None,
        topic_alias_max: 32,
        retain_available: true,
        wildcard_subscription_available: true,
        subscription_identifiers_available: true,
        shared_subscription_available: true,
        server_keepalive_sec: None,
        response_info: None,
        server_reference: None,
        auth_method: None,
        auth_data: None,
        reason_string: None,
        user_properties: [],
    }
[2024-02-03T07:19:08Z TRACE ntex_mqtt::v5::sink] Sending subscribe packet Subscribe {
        packet_id: 1,
        id: None,
        user_properties: [],
        topic_filters: [
            (
                "response",
                SubscriptionOptions {
                    qos: AtLeastOnce,
                    no_local: false,
                    retain_as_published: false,
                    retain_handling: AtSubscribe,
                },
            ),
        ],
    }
[2024-02-03T07:19:08Z DEBUG ntex_mqtt::v5::client::connection] start mqtt client keep-alive task
[2024-02-03T07:19:08Z TRACE ntex_mqtt::v5::client::dispatcher] Dispatch packet: DispatchItem::Item((Disconnect(Disconnect { reason_code: UnspecifiedError, session_expiry_interval_secs: None, server_reference: None, reason_string: None, user_properties: [] }), 2))
[2024-02-03T07:19:08Z TRACE ntex_mqtt::v5::client::dispatcher] Dispatch packet: DispatchItem::Disconnect(None)
[2024-02-03T07:19:08Z TRACE ntex_mqtt::io] TCP-CLIENT: io shutdown completed
[2024-02-03T07:19:08Z TRACE ntex_mqtt::io] TCP-CLIENT: Service shutdown is completed, stop
thread 'main' panicked at examples/client.rs:56:10:
called `Result::unwrap()` on an `Err` value: Disconnected
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
Panic in Arbiter thread.

(base) ➜  ntex-mqtt git:(master) ✗ cargo run --example basic
    Finished dev [unoptimized + debuginfo] target(s) in 0.30s
     Running `target/debug/examples/basic`
[2024-02-03T07:19:00Z INFO  ntex::server::builder] Starting 1 workers
[2024-02-03T07:19:00Z INFO  ntex::server::builder] Starting "mqtt" service on 127.0.0.1:1883
[2024-02-03T07:19:00Z TRACE ntex::server::accept] Starting server accept loop
[2024-02-03T07:19:00Z INFO  ntex::server::accept] Starting socket listener on 127.0.0.1:1883
[2024-02-03T07:19:00Z TRACE ntex::server::worker] Service "mqtt" is available
[2024-02-03T07:19:00Z TRACE ntex::server::accept] Worker is available
[2024-02-03T07:19:08Z TRACE ntex::server::accept] Accepting connection: Tcp(TcpStream { addr: 127.0.0.1:1883, peer: 127.0.0.1:51124, fd: 15 }) bp: false
[2024-02-03T07:19:08Z TRACE ntex::server::accept] Sent to worker 0
[2024-02-03T07:19:08Z TRACE ntex::server::worker] Got socket for service: "mqtt"
[2024-02-03T07:19:08Z TRACE ntex_io::tasks] : New 24 bytes available, wakeup dispatcher
[2024-02-03T07:19:08Z TRACE ntex_mqtt::v5::server] Starting mqtt v5 handshake
[2024-02-03T07:19:08Z INFO  basic] new connection: Connect { clean_start: false, keep_alive: 1, session_expiry_interval_secs: 0, auth_method: None, auth_data: None, request_problem_info: true, request_response_info: false, receive_max: None, topic_alias_max: 0, user_properties: [], max_packet_size: Some(30), last_will: None, client_id: "user", username: None, password: None }
[2024-02-03T07:19:08Z TRACE ntex_mqtt::v5::server] Sending: ConnectAck {
        session_present: false,
        reason_code: Success,
        session_expiry_interval_secs: None,
        receive_max: 15,
        max_qos: AtLeastOnce,
        max_packet_size: None,
        assigned_client_id: None,
        topic_alias_max: 32,
        retain_available: true,
        wildcard_subscription_available: true,
        subscription_identifiers_available: true,
        shared_subscription_available: true,
        server_keepalive_sec: None,
        response_info: None,
        server_reference: None,
        auth_method: None,
        auth_data: None,
        reason_string: None,
        user_properties: [],
    }
[2024-02-03T07:19:08Z TRACE ntex_mqtt::service] : Connection handshake succeeded
[2024-02-03T07:19:08Z TRACE ntex_mqtt::service] : Connection handler is created, starting dispatcher
[2024-02-03T07:19:08Z DEBUG ntex_mqtt::io] : Start keep-alive timer Seconds(1)
[2024-02-03T07:19:08Z DEBUG ntex_io::ioref] : Start timer Seconds(1)
[2024-02-03T07:19:08Z DEBUG ntex_io::timer] : Timer driver does not run, current: 0
[2024-02-03T07:19:08Z TRACE ntex_io::tasks] : New 16 bytes available, wakeup dispatcher
[2024-02-03T07:19:08Z TRACE ntex_mqtt::v5::dispatcher] Dispatch v5 packet: DispatchItem::Item((Subscribe(Subscribe { packet_id: 1, id: None, user_properties: [], topic_filters: [("response", SubscriptionOptions { qos: AtLeastOnce, no_local: false, retain_as_published: false, retain_handling: AtSubscribe })] }), 14))
[2024-02-03T07:19:08Z WARN  ntex_mqtt::v5::default] MQTT5 Control service is not configured, pkt: Subscribe(Subscribe { packet: Subscribe { packet_id: 1, id: None, user_properties: [], topic_filters: [("response", SubscriptionOptions { qos: AtLeastOnce, no_local: false, retain_as_published: false, retain_handling: AtSubscribe })] }, result: SubscribeAck { packet_id: 1, properties: [], reason_string: None, status: [UnspecifiedError] }, size: 14 })
[2024-02-03T07:19:08Z TRACE ntex_io::io] : Initiate io shutdown Flags(RD_READY | WR_PAUSED | DSP_STOP)
[2024-02-03T07:19:08Z TRACE ntex_mqtt::io] : Dispatcher is instructed to stop
[2024-02-03T07:19:08Z DEBUG ntex_io::ioref] : Stop timer
[2024-02-03T07:19:08Z TRACE ntex_tokio::io] : Write task is instructed to shutdown
[2024-02-03T07:19:08Z TRACE ntex_tokio::io] : Read task is instructed to shutdown
[2024-02-03T07:19:08Z TRACE ntex_tokio::io] : Tokio write task is stopped
[2024-02-03T07:19:08Z TRACE ntex_mqtt::io] : io shutdown completed
[2024-02-03T07:19:08Z WARN  ntex_mqtt::v5::default] MQTT5 Control service is not configured, pkt: Closed(Closed)
[2024-02-03T07:19:08Z TRACE ntex_mqtt::io] : Service shutdown is completed, stop
[2024-02-03T07:19:08Z TRACE ntex_io::ioref] : Force close io stream object
^C[2024-02-03T07:19:56Z INFO  ntex::server::builder] SIGINT received, exiting
[2024-02-03T07:19:56Z TRACE ntex::server::accept] Stopping accept loop
[2024-02-03T07:19:56Z INFO  ntex::server::accept] Stopping socket listener on 127.0.0.1:1883
[2024-02-03T07:19:56Z TRACE ntex::server::accept] Accept loop has been stopped
[2024-02-03T07:19:56Z INFO  ntex::server::worker] Shutting down worker, 0 connections

[Question] if auth/acl http callback error message can be sent to client？

If should I create topic which publish the error message for auth/acl？

是否必须通过创建一个专门用来发布认证/授权错误信息的主题来解决这个问题？

how to only provide tls external port

当注释掉 listener.tcp.external.addr = "0.0.0.0:1883" 打开 listener.tls.external.addr = "0.0.0.0:8883"

报错：

thread 'main' panicked at 'called Result::unwrap() on an Err value: ConfigError(missing field addr)', /home/try/workspace/rmqtt/rmqtt/src/settings/mod.rs :100 :42

docker image 版本：0.2.16

请教一下，有没有批量发布的接口？怎么调用？

Subscriptions after server restart and data persistence.

First off, I would like to take a moment to express my sincere gratitude for the incredible work you've done on the Rmqtt MQTT broker.

I recently started using Rmqtt and I have been thoroughly impressed with its features and performance. It has significantly enhanced my projects, and for that, I am truly thankful.

However, I wanted to share a concern I've encountered. It appears that subscriptions are not persisting after server restarts. This has caused inconvenience as clients need to subscribe again after every server restart. I believe a more robust subscription storage solution would greatly improve the user experience.

Additionally, I would appreciate guidance on how to properly stop the broker. Clear documentation on this aspect would be immensely helpful for users like me.

Once again, thank you for your hard work and dedication. I look forward to seeing Rmqtt continue to flourish and improve.

Client ID not randomly generated

I think that when connecting with a client id of length 0, the server is not generating a proper id for the client, causing clients to kick each other from the network.

AFAIK the server must either allow a client id of length 0 an generate a random ID, or reject the client. I'm not sure if MQTT5 has changed this.

From the spec:

A Server MAY allow a Client to supply a ClientId that has a length of zero bytes, however if it does so the Server MUST treat this as a special case and assign a unique ClientId to that Client. It MUST then process the CONNECT packet as if the Client had provided that unique ClientId [MQTT-3.1.3-6].

最新版本能发布一个release么？

最新版本能发布一个release么？我怕生成又会遇到啥问题

Could not compile from source code

Rust 1.69.0
Source Code 0.26.0

The errors are below:

error: failed to run custom build command for rmqtt v0.2.10 (E:\servers\sources\rmqtt\rmqtt)

Caused by:
process didn't exit successfully: E:\servers\sources\rmqtt\target\debug\build\rmqtt-691acc4cc66764fd\build-script-build (exit code: 101)
--- stdout
out: E:\servers\sources\rmqtt\target\debug\build\rmqtt-c73114283d9fa88b\out
cargo:rerun-if-changed=pb.proto
cargo:rerun-if-changed=src/grpc/proto

--- stderr
thread 'main' panicked at 'Could not find protoc installation and this build crate cannot proceed without
this knowledge. If protoc is installed and this crate had trouble finding
it, you can set the PROTOC environment variable with the specific path to your
installed protoc binary.You can download it from https://github.com/protocolbuffers/protobuf/releases or from your package manager.

Caused by:
process didn't exit successfully: E:\servers\sources\rmqtt\target\debug\build\rmqtt-raft-20c6376ba58d4924\build-script-build (exit code: 101)
--- stdout
out: E:\servers\sources\rmqtt\target\debug\build\rmqtt-raft-6b26a9737cfa55aa\out
cargo:rerun-if-changed=raft_service.proto
cargo:rerun-if-changed=proto/

For more information: https://docs.rs/prost-build/#sourcing-protoc
', C:\Users\mekot.cargo\registry\src\github.com-1ecc6299db9ec823\prost-build-0.11.9\src\lib.rs:1457:10
note: run with RUST_BACKTRACE=1 environment variable to display a backtrace
error[E0793]: reference to packed field is unaligned
--> C:\Users\mekot.cargo\registry\src\github.com-1ecc6299db9ec823\ntapi-0.3.7\src\ntexapi.rs:2783:52
|
2783 | *tick_count.QuadPart_mut() = read_volatile(&(*USER_SHARED_DATA).u.TickCountQuad);
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
= note: fields of packed structs are not properly aligned, and creating a misaligned reference is undefined behavior (even if that reference is never dereferenced)
= help: copy the field contents to a local variable, or replace the reference with a raw pointer and use read_unaligned/write_unaligned (loads and stores via *p must be properly aligned even when using raw pointers)

error[E0793]: reference to packed field is unaligned
--> C:\Users\mekot.cargo\registry\src\github.com-1ecc6299db9ec823\ntapi-0.3.7\src\ntexapi.rs:2807:25
|
2807 | ((read_volatile(&(*USER_SHARED_DATA).u.TickCountQuad)
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
= note: fields of packed structs are not properly aligned, and creating a misaligned reference is undefined behavior (even if that reference is never dereferenced)
= help: copy the field contents to a local variable, or replace the reference with a raw pointer and use read_unaligned/write_unaligned (loads and stores via *p must be properly aligned even when using raw pointers)

For more information about this error, try rustc --explain E0793.
error: could not compile ntapi due to 2 previous errors

login does not correctly handle error HTTP response

RMQTT Docker Version

0.2.11

HTTP Auth Response

POST http://dev.fornetcode.com/mqtt/auth

HTTP/1.1 400 Bad Request
Server: nginx/1.23.4
Date: Fri, 22 Sep 2023 15:10:24 GMT
Content-Type: text/plain;charset=utf-8
Content-Length: 18
Connection: keep-alive

.clientId(missing)

Log:

Result:
mqtt 链接正常保持，没有被踢掉。 MQTT 应该关闭链接的

建议新增配置项

加一个配置项
listener.tcp.external.active = true
listener.tcp.internal.active = false