Crate poster

source ·
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::new()).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:

  1. Graceful disconnection is performed (using ContextHandle::disconnect method). The result is then ().
  2. 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::new().topic_name("topic").payload("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::new().subscription("topic", SubscriptionOpts::new());
    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::new()
        .subscription("topic1", SubscriptionOpts::new())
        .subscription("topic2", SubscriptionOpts::new());

    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::new()
        .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::new().topic_filter("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::new()).await?;

See DisconnectOpts.

Error handling

The main library error type is MqttError enum found in error module.

TLS/SSL

TLS/SSL libraries are available out there with AsyncRead, AsyncWrite TLS/SSL streams. These may be supplied to the set_up method. The library does not handle encription on its own.

Modules

Structs

  • AuthOpts
    Authorization options, represented as a consuming builder. Used during extended authorization, translated to the AUTH packet.
  • AuthRsp
    Response from connection request, if extended authorization is performed. Accesses data in AUTH packet.
  • ConnectOpts
    Connection options, represented as a consuming builder. Used during connection request, translated to the CONNECT packet.
  • ConnectRsp
    Response from connection request. Accesses data in CONNACK packet.
  • Context
    Client context. Responsible for socket management and direct communication with the broker.
  • ContextHandle
    Cloneable handle to the client Context. The ContextHandle object is used to perform MQTT operations.
  • DisconnectOpts
    Disconnection options, represented as a consuming builder. Used during disconnection request, translated to the DISCONNECT packet.
  • PubackRsp
    Response to the publish request, with QoS==1 representing the PUBACK packet.
  • PubcompRsp
    Response to the publish request, with QoS==2 representing the PUBCOMP packet.
  • PublishData
    Accesses data in the incoming PUBLISH packet.
  • PublishOpts
    Publish options, represented as a consuming builder. Used during publish request, translated to the PUBLISH packet.
  • PubrecRsp
    Response to the publish request, with QoS==2 representing the PUBREC packet.
  • SubscribeOpts
    Subscription options, represented as a consuming builder. Used during subscription request, translated to the SUBSCRIBE packet. Note that multiple topic filters may be supplied.
  • SubscribeRsp
    Response to the subscription request, representing the Suback packet.
  • SubscriptionOpts
    Subscription options set for the topic filter.
  • UnsubscribeOpts
    Unsubscribe options, represented as a consuming builder. Used during unsubscribe request, translated to the UNSUBSCRIBE packet.
  • UnsubscribeRsp
    Response to the unsubscribe request, representing the UNSUBACK packet.
  • UserProperties
    Map collection for reading user properties as key-value pairs from packets.

Enums