Expand description
Poster-rs is an asynchronous, runtime agnostic, zero-copy MQTT 5 library, designed having operation locality in mind.
Set up
Firstly, choose your async runtime. Ready? Lets go!
In the below example we will use Tokio.
use std::error::Error;
use poster::{prelude::*, ConnectOpts, Context};
use tokio::net::TcpStream;
use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn Error>> {
let (mut ctx, mut handle) = Context::new();
let ctx_task = tokio::spawn(async move {
// Set up a connection using your async framework of choice. We will need a read end, which is
// AsyncRead, and write end, which is AsyncWrite, so we split the TcpStream
// into ReadHalf and WriteHalf pair.
let (rx, tx) = TcpStream::connect("127.0.0.1:1883").await?.into_split();
// Pass (ReadHalf, WriteHalf) pair into the context and connect with the broker on
// the protocol level.
ctx.set_up((rx.compat(), tx.compat_write())).connect(ConnectOpts::default()).await?;
// Awaiting the Context::run invocation will block the current task.
if let Err(err) = ctx.run().await {
eprintln!("[context] Error occured: \"{}\", exiting...", err);
} else {
println!("[context] Context exited.");
}
Ok::<(), Box<dyn Error + Send + Sync>>(())
});
/* ... */
ctx_task.await?;
Ok(())
}
At this point, our context is up and running.
Let’s break down the above example.
poster-rs
is a runtime agnostic library, which means that all the asynchronous operations are abstracted
using traits from the futures-rs
crate. The result of this approach is that connection with the broker
must be established manually and the library only cares about receving (AsyncRead,
AsyncWrite) pair during the context creation.
This pair is usually obtained using some sort of split
functions on streams/sockets in the networking libraries.
(See tokio, smol)
new factory method gives us (Context, ContextHandle) tuple. Context is responsible for handling the traffic between the client and the server. ContextHandle however, is a cloneable handle to the Context actor and is used to perform all the MQTT operations.
Method run blocks the task (on .await) until one of the following conditions is met:
- Graceful disconnection is performed (using ContextHandle::disconnect method). The result is then ().
- Error occurs, resulting in MqttError. This may be the result of socket closing, receiving DISCONNECT from the server, etc.
Publishing
Publishing is performed via the publish method.
// ...
let opts = PublishOpts::default().topic("topic").data("hello there".as_bytes());
handle.publish(opts).await?;
See PublishOpts.
Subscriptions
Subscriptions are represented as async streams, obtained via the stream method. The general steps of subscribing are:
- await the invocation of subscribe method
- validate the result (optionally)
- use stream method in order to create a stream for the subscription.
Note that under the hood, the library uses subscription identifiers to group subscriptions.
See SubscribeOpts.
// ...
let opts = SubscribeOpts::default().subscription("topic", SubscriptionOpts::default());
let rsp = handle.subscribe(opts).await?;
let mut subscription = rsp.stream();
while let Some(msg) = subscription.next().await {
println!("topic: {}; payload: {}", msg.topic_name(), str::from_utf8(msg.payload()).unwrap());
}
User may subscribe to multiple topics in one subscription request.
// ...
let opts = SubscribeOpts::default()
.subscription("topic1", SubscriptionOpts::default())
.subscription("topic2", SubscriptionOpts::default());
let mut subscription = handle.subscribe(opts).await?.stream();
while let Some(msg) = subscription.next().await {
println!("topic: {}; payload: {}", msg.topic_name(), str::from_utf8(msg.payload()).unwrap());
}
Each subscription may be customized using the SubscriptionOpts.
let opts = SubscribeOpts::default()
.subscription("topic", SubscriptionOpts::new().maximum_qos(QoS::AtLeastOnce));
SubscribeRsp struct represents the result of the subscription request. In order to access per-topic reason codes, payload method is used:
// ...
let rsp = handle.subscribe(opts).await?;
let all_ok = rsp.payload().iter().copied().all(|reason| reason == SubackReason::GranteedQoS2);
Unsubscribing
Unsubscribing is performed by the unsubscribe method. Note that it does NOT close the subscription stream (it could lead to logic errors).
// ...
let opts = UnsubscribeOpts::default().topic("topic");
let rsp = handle.unsubscribe(opts).await?;
As with subscribing, per topic reason codes can be obtained by the payload method.
See UnsubscribeOpts.
Keep alive and ping
If the keep_alive interval is set during the connection request, the user must use the ping method periodically.
Disconnection
Disconnection may be initiated either by user or the broker. When initiated by the broker, the run method returns Disconnected error.
Graceful disconnection may be also performed by the user by using disconnect method. When disconnection is finished, run method returns ().
// ...
handle.disconnect(DisconnectOpts::default()).await?;
See DisconnectOpts.
Error handling
The main library error type is MqttError enum found in error module.
TSL/SSL
TSL/SSL libraries are available out there with AsyncRead, AsyncWrite TSL/SSL streams. These may be supplied to the set_up method. The library does not handle encription on its own.