Struct MqttTransport 

pub struct MqttTransport { /* private fields */ }

An MQTT client that implements the core [Transport] and [Receive] traits.

A transport is created disconnected; connect opens the link and spawns the background task that runs the MQTT event loop for the life of the connection. Inbound messages are queued and read with recv.

A connection can end without being asked to: the broker restarts, another client connects with the same client id, or a packet over the limit arrives. From then on is_connected is false, a send answers [Error::Closed], and the next receive reports why the connection ended, until connect opens a new one.

Implementations

impl MqttTransport

pub fn new(config: MqttConfig) -> Self

Creates a transport from the given configuration without connecting.

Arguments

• config - the broker connection settings.

Returns

A disconnected transport ready for connect.

pub async fn publish( &mut self, topic: &str, payload: &[u8], options: PublishOptions, ) -> Result<Delivery>

Publishes a message with options of its own, and follows it to the broker.

The call returns once the message is queued for the broker, as send does. The Delivery it hands back settles when the broker acknowledges the message, so a caller that needs to know the message arrived awaits it, and one that does not drops it.

Arguments

• topic - the topic to publish to, with no wildcard.
• payload - the message.
• options - the quality of service and retain flag for this message.

Returns

The delivery to await for the broker’s acknowledgment.

Errors

Returns [Error::Transport] if MQTT does not allow the topic or the message is over the connection’s packet limit, both before anything is sent, or [Error::Closed] if the transport is not connected.

Examples

use pamoja_core::Transport;
use pamoja_mqtt::{MqttConfig, MqttTransport, PublishOptions};

let mut transport = MqttTransport::new(MqttConfig::new("sensor-1", "localhost", 1883));
transport.connect().await?;

// The last reading stays with the broker for any dashboard that subscribes later,
// and the call waits until the broker says it has it.
transport
    .publish("sites/tank-2/level", b"73", PublishOptions::new().retained())
    .await?
    .confirmed()
    .await?;

pub fn inbox(&self) -> Inbox

Returns a handle on the messages arriving on this transport’s connection.

Returns

The inbox, which a task can wait on while another publishes through the transport.

pub fn is_connected(&self) -> bool

Reports whether the transport currently holds an active connection.

Returns

true once connect has succeeded, until disconnect is called or the connection ends on its own.

pub async fn disconnect(&mut self) -> Result<()>

Closes the connection and stops the background event loop.

The client says goodbye with a DISCONNECT, so the broker discards the will rather than publishing it (OASIS MQTT 3.1.1, section 3.14). Calling this on a transport that is not connected is a no-op. A transport that is dropped without disconnecting closes its connection without a goodbye, and the broker publishes its will.

Returns

Ok(()) once the goodbye has gone out, or a second has passed without the event loop sending it, and the loop has been stopped.

Errors

This call is best-effort and does not surface broker errors raised while tearing down, so it currently always returns Ok(()).

Trait Implementations

impl Drop for MqttTransport

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.

impl Receive for MqttTransport

async fn recv(&mut self) -> Result<Option<Message>>

Awaits the next message from any subscribed topic.

Returns

Some(message) for the next queued message, or None once the connection has ended and its reason has been reported.

Errors

Returns [Error::Closed] if the transport is not connected, or [Error::Transport] once, saying why, when the connection ended on its own.

impl Transport for MqttTransport

async fn connect(&mut self) -> Result<()>

Opens a connection to the broker, closing any the transport already holds.

The session is clean, so subscriptions from an earlier connection are gone and have to be placed again.

Returns

Ok(()) once the broker has accepted the connection.

Errors

Returns [Error::Transport] if the broker cannot be reached or refuses the connection.

async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()>

Publishes a payload to a topic under the configured quality of service.

Returns

Ok(()) once the message is queued for the broker, which is before the broker acknowledges it; publish waits for that.

Errors

Returns [Error::Transport] if MQTT does not allow the topic or the message is over the connection’s packet limit, both before anything is sent, or [Error::Closed] if the transport is not connected.

async fn subscribe(&mut self, topic: &str) -> Result<()>

Subscribes to a topic filter under the configured quality of service.

Returns

Ok(()) once the subscription is queued for the broker.

Errors

Returns [Error::Transport] if the filter places a wildcard where MQTT does not allow one, or [Error::Closed] if the transport is not connected.

fn send_text( &mut self, topic: &str, text: &str, ) -> impl Future<Output = Result<(), Error>> + Send

Publishes text to a topic: words, or a number written out.