Shulou(Shulou.com)06/02 Report--

This article focuses on "how to use MQTT in Rust". Interested friends may wish to have a look at it. The method introduced in this paper is simple, fast and practical. Let's let the editor learn "how to use MQTT in Rust".

Rust is a general and compiled programming language developed by Mozilla. The design criteria of the language are: secure, concurrent, practical, supporting functional, concurrent, procedural and object-oriented programming styles. The Rust speed is amazing and the memory utilization is extremely high. Without runtime and garbage collection, it is competent for performance-demanding services, can run on embedded devices, and can easily integrate with other languages. Rust's rich type system and ownership model ensures memory safety and thread safety, allowing you to eliminate a variety of errors at compile time.

MQTT is a lightweight Internet of things message transmission protocol based on publish / subscribe mode, which can provide real-time and reliable message service for networking devices with very little code and bandwidth. it is widely used in Internet of things, mobile Internet, intelligent hardware, vehicle Internet, power energy and other industries.

This paper mainly introduces how to use paho-mqtt client library in Rust project to realize the functions of connecting, subscribing, unsubscribing, sending and receiving messages between client and MQTT server.

Project initialization

This project uses Rust 1.44.0 for development and testing, and uses Cargo 1.44.0 package management tools for project management. Readers can view the current Rust version with the following command.

~ rustc-- versionrustc 1.44.0 (49cae5576 2020-06-01) Select MQTT client library

Paho-mqtt is currently a fully functional and widely used MQTT client in Rust. The latest version 0.7.1 supports MQTT v5, 3.1.1, 3.1, supports data transfer through standard TCP, SSL / TLS, WebSockets, and QoS supports 0, 1, 2, etc.

Initialize the project

Execute the following command to create a new Rust project named mqtt-example.

~ cargo new mqtt-example Created binary (application) `mqtt- example` package

Edit the Cargo.toml file in the project, add the address of the paho-mqtt library to the dependencies, and specify the binaries corresponding to the subscription and publication code files.

[dependencies] paho-mqtt = {git = "https://github.com/eclipse/paho.mqtt.rust.git", branch =" master "} [[bin]] name =" sub "path =" src/sub/main.rs "[[bin]] name =" pub "path =" src/pub/main.rs "Rust MQTT to create a client connection

This article will use the free public MQTT server provided by EMQ X as the MQTT server to test the connection, which is based on EMQ X's MQTT Internet of things cloud platform. The server access information is as follows:

Broker: broker.emqx.io

TCP Port: 1883

Websocket Port: 8083

Configure MQTT Broker connection parameters

Configure the MQTT Broker connection address (including ports), topic (here we configured two topic), and client id.

Const DFLT_BROKER:&str = "tcp://broker.emqx.io:1883"; const DFLT_CLIENT:&str = "rust_publish"; const DFLT_TOPICS:& [& str] = & ["rust/mqtt", "rust/test"]; write MQTT connection code

Write MQTT connection code, and to improve the experience, you can pass in the connection address in the form of command line arguments when executing binaries. Usually we need to create a client and then connect the client to the broker.emqx.io.

Let host = env::args () .nth (1) .unwrap_or_else (| | DFLT_BROKER.to_string ()); / / Define the set of options for the create.// Use an ID for a persistent session.let create_opts = mqtt::CreateOptionsBuilder::new () .server _ uri (host) .client _ id (DFLT_CLIENT.to_string ()) .fi nalize () / / Create a client.let cli = mqtt::Client::new (create_opts) .unwrap_or_else (| err | {println! ("Error creating the client: {:?}", err); process::exit (1);}); / / Define the set of options for the connection.let conn_opts = mqtt::ConnectOptionsBuilder::new () .keep _ alive_interval (Duration::from_secs (20)) .clean _ session (true) .finalize () / / Connect and wait for it to complete or fail.if let Err (e) = cli.connect (conn_opts) {println! ("Unable to connect:\ n\ t {:?}", e); process::exit (1);} publish messages

Here we post a total of five messages to rust/mqtt and rust/test respectively according to the parity of the loop.

For num in 0. 5 {let content = "Hello world!" .to _ string () + & num.to_string (); let mut msg = mqtt::Message::new (DFLT_TOPICS [0], content.clone (), QOS); if num% 2 = 0 {println! ("Publishing messages on the {:?} topic", DFLT_TOPICS [1]); msg = mqtt::Message::new (DFLT_TOPICS [1], content.clone (), QOS) } else {println! ("Publishing messages on the {:?} topic", DFLT_TOPICS [0]);} let tok = cli.publish (msg); if let Err (e) = tok {println! ("Error sending message: {:}", e); break }} subscribe to messages

The consumer needs to be initialized before the client can connect. Here we will loop through the message queue in the consumer and print out the topic name of the subscription and the content of the message received.

Fn subscribe_topics (cli: & mqtt::Client) {if let Err (e) = cli.subscribe_many (DFLT_TOPICS, DFLT_QOS) {println! ("Error subscribes topics: {:?}", e); process::exit (1);}} fn main () {. / / Initialize the consumer before connecting. Let rx = cli.start_consuming (); / / Subscribe topics. Subscribe_topics (& cli); println! ("Processing requests..."); for msg in rx.iter () {if let Some (msg) = msg {println! ("{}", msg);} else if! cli.is_connected () {if try_reconnect (& cli) {println! ("Resubscribe topics...") Subscribe_topics (& cli);} else {break;}...} complete code message publishing code use std:: {env, process, time::Duration}; extern crate paho_mqtt as mqtt;const DFLT_BROKER:&str = "tcp://broker.emqx.io:1883"; const DFLT_CLIENT:&str = "rust_publish" Const DFLT_TOPICS:& [& str] = & ["rust/mqtt", "rust/test"]; / / Define the qos.const QOS:i32 = 1 unwrap_or_else FN main () {let host = env::args (). Nth (1). Unwrap_or_else (| | DFLT_BROKER.to_string ()); / / Define the set of options for the create. / / Use an ID for a persistent session. Let create_opts = mqtt::CreateOptionsBuilder::new () .server _ uri (host) .client _ id (DFLT_CLIENT.to_string ()) .finalize (); / / Create a client. Let cli = mqtt::Client::new (create_opts) .unwrap_or_else (| err | {println! ("Error creating the client: {:}", err); process::exit (1);}); / / Define the set of options for the connection. Let conn_opts = mqtt::ConnectOptionsBuilder::new (). Keep _ alive_interval (Duration::from_secs (20)). Clean _ session (true). Finalize (); / / Connect and wait for it to complete or fail. If let Err (e) = cli.connect (conn_opts) {println! ("Unable to connect:\ n\ t {:?}", e); process::exit (1);} / / Create a message and publish it. / / Publish message to 'test' and' hello' topics. For num in 0.. 5 {let content = "Hello world!" .to _ string () + & num.to_string (); let mut msg = mqtt::Message::new (DFLT_TOPICS [0], content.clone (), QOS); if num% 2 = = 0 {println! ("Publishing messages on the {:?} topic", DFLT_TOPICS [1]) Msg = mqtt::Message::new (DFLT_TOPICS [1], content.clone (), QOS);} else {println! ("Publishing messages on the {:?} topic", DFLT_TOPICS [0]);} let tok = cli.publish (msg) If let Err (e) = tok {println! ("Error sending message: {:?}", e); break;}} / / Disconnect from the broker. Let tok = cli.disconnect (None); println! ("Disconnect from the broker"); tok.unwrap ();} message subscription code

In order to improve the user experience, message subscriptions disconnect and reconnect, and re-subscribe topics after re-establishing the connection.

Use std:: {env, process, thread, time::Duration}; extern crate paho_mqtt as mqtt;const DFLT_BROKER:&str = "tcp://broker.emqx.io:1883"; const DFLT_CLIENT:&str = "rust_subscribe"; const DFLT_TOPICS:& [& str] = & ["rust/mqtt", "rust/test"]; / / The qos list that match topics above.const DFLT_QOS:& [i32] = & [0,1] / / Reconnect to the broker when connection is lost.fn try_reconnect (cli: & mqtt::Client)-> bool {println! ("Connection lost. Waiting to retry connection "); for _ in 0.. 12 {thread::sleep (Duration::from_millis (5000)); if cli.reconnect (). Is_ok () {println! (" Successfully reconnected "); return true;}} println! (" Unable to reconnect after several attempts. ") False} / / Subscribes to multiple topics.fn subscribe_topics (cli: & mqtt::Client) {if let Err (e) = cli.subscribe_many (DFLT_TOPICS, DFLT_QOS) {println! ("Error subscribes topics: {:?}", e); process::exit (1) }} fn main () {let host = env::args (). Nth (1). Unwrap_or_else (| | DFLT_BROKER.to_string ()); / / Define the set of options for the create. / / Use an ID for a persistent session. Let create_opts = mqtt::CreateOptionsBuilder::new () .server _ uri (host) .client _ id (DFLT_CLIENT.to_string ()) .finalize (); / / Create a client. Let mut cli = mqtt::Client::new (create_opts) .unwrap_or_else (| err | {println! ("Error creating the client: {:}", err); process::exit (1);}); / / Initialize the consumer before connecting. Let rx = cli.start_consuming (); / / Define the set of options for the connection. Let lwt = mqtt::MessageBuilder::new () .topic ("test") .payload ("Consumer lost connection") .finalize (); let conn_opts = mqtt::ConnectOptionsBuilder::new () .keep _ alive_interval (Duration::from_secs (20)). Clean _ session (false). Will _ message (lwt) .finalize (); / / Connect and wait for it to complete or fail. If let Err (e) = cli.connect (conn_opts) {println! ("Unable to connect:\ n\ t {:?}", e); process::exit (1);} / / Subscribe topics. Subscribe_topics (& cli); println! ("Processing requests..."); for msg in rx.iter () {if let Some (msg) = msg {println! ("{}", msg);} else if! cli.is_connected () {if try_reconnect (& cli) {println! ("Resubscribe topics...") Subscribe_topics (& cli);} else {break;}} / / If still connected, then disconnect now. If cli.is_connected () {println! ("Disconnecting"); cli.unsubscribe_many (DFLT_TOPICS). Unwrap (); cli.disconnect (None). Unwrap ();} println! ("Exiting");} run and test compile binaries

Execute the following command to generate message subscriptions and publish corresponding sub and pub binaries in the mqtt-example/target/debug directory.

Cargo build

Execute the sub binaries and wait for the consumer to publish.

News release

Execute the pub binaries and you can see that messages have been posted to the two topics rust/test and rust/mqtt, respectively.

At the same time, you can see the published message in the message subscription.

At this point, we have finished connecting to the public MQTT server using the paho-mqtt client, and implemented the connection, message publishing and subscription between the test client and the MQTT server.

At this point, I believe you have a deeper understanding of "how to use MQTT in Rust". You might as well do it in practice. Here is the website, more related content can enter the relevant channels to inquire, follow us, continue to learn!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.