mqrstt

https://docs.rs/mqrstt/
Admin
mqrstt is an easy-to-use, efficient, and runtime-agnostic pure Rust MQTT v5 client providing both...
Tokens:311,282
Snippets:1,267
Trust Score:9.1
Update:1 month ago
Show doc for...
Tokens:
Raw
### publish_with_properties_blocking example (truncated)

Source: https://docs.rs/mqrstt/latest/src/mqrstt/client.rs.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Starts an example for publish_with_properties_blocking, showing the setup of a PublishProperties struct with a response_topic. The code is truncated in the source; the full example is not visible on this page.

```rust
# let (network, mqtt_client) = mqrstt::NetworkBuilder::<(), smol::net::TcpStream>::new_from_client_id("Example").smol_network();
# smol::block_on(async {

use mqrstt::packets::QoS;
use mqrstt::packets::PublishProperties;

let properties = PublishProperties{
    response_topic: Some("response/topic".into()),
```

--------------------------------

### ConnectOptions::get_clean_start

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.ConnectOptions.html

Gets the clean start flag.

```APIDOC
## ConnectOptions::get_clean_start

### Description
Returns whether the connection starts a new session or continues an existing session. This is specified by the MQTT v5.0 specification.

### Method
`pub fn get_clean_start(&self) -> bool`

### Returns
- `bool` - `true` if a new session is started, `false` if an existing session is continued.
```

--------------------------------

### Test handler setup with AvailablePacketIds and ConnectOptions

Source: https://docs.rs/mqrstt/latest/src/mqrstt/state_handler.rs.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Test helper that creates a StateHandler with a fresh AvailablePacketIds and a ConnectOptions with receive maximum 100 and configurable clean start. Returns the handler and the receiver for available packet IDs.

```rust
use async_channel::Receiver;

use crate::{
    ConnectOptions, StateHandler,
    available_packet_ids::AvailablePacketIds,
    packets::{
        Packet, PubComp, PubCompProperties, PubCompReasonCode, PubRec, PubRecProperties, PubRecReasonCode, PubRel, PubRelProperties, PubRelReasonCode, QoS, SubAck, SubAckProperties,
        SubAckReasonCode, UnsubAck, UnsubAckProperties, UnsubAckReasonCode,
    },
};

use crate::tests::test_packets::*;

fn handler(clean_start: bool) -> (StateHandler, Receiver<u16>) {
    let (apkids, apkids_r) = AvailablePacketIds::new(100);

    let mut opt = ConnectOptions::new("test123123");
    opt.set_receive_maximum(100).set_clean_start(clean_start);

    (StateHandler::new(&opt, apkids), apkids_r)
}
```

--------------------------------

### Tokio MQTT client example with NetworkBuilder

Source: https://docs.rs/mqrstt

Use this example to connect to an MQTT broker using the tokio runtime. It creates a network and client with a no-op handler, connects to broker.emqx.io on port 1883, subscribes to the topic "mqrstt", runs the network in a spawned task for 30 seconds, then disconnects and asserts the network status is OutgoingDisconnect. The handler must not wait too long, and a new connection should be created after an error or disconnect.

```rust
use mqrstt::{
    example_handlers::NOP,
    NetworkBuilder, NetworkStatus,
};

use tokio::time::Duration;

#[tokio::main]
async fn main() {
    let (mut network, client) = NetworkBuilder::new_from_client_id("TokioTcpPingPongExample").tokio_network();
    // Construct a no op handler
    let mut nop = NOP {};
    // In normal operations you would want to loop this connection
    // To reconnect after a disconnect or error
    let stream = tokio::net::TcpStream::connect(("broker.emqx.io", 1883)).await.unwrap();
    network.connect(stream, &mut nop).await.unwrap();

    client.subscribe("mqrstt").await.unwrap();
    // Run the network
    let network_handle = tokio::spawn(async move { network.run(&mut nop).await });

    tokio::time::sleep(Duration::from_secs(30)).await;
    client.disconnect().await.unwrap();
    let result = network_handle.await;
    assert!(result.is_ok());
    assert_eq!(result.unwrap().unwrap(), NetworkStatus::OutgoingDisconnect);
}
```

--------------------------------

### Smol MQTT client example with NetworkBuilder

Source: https://docs.rs/mqrstt

Use this example to connect to an MQTT broker using the smol runtime. It creates a network and client with a no-op handler, connects to broker.emqx.io on port 1883, subscribes to the topic "mqrstt", runs the network for 30 seconds, then disconnects and asserts the network status is OutgoingDisconnect. The handler must not wait too long, and a new connection should be created after an error or disconnect.

```rust
use mqrstt::{example_handlers::NOP, NetworkBuilder, NetworkStatus};

smol::block_on(async {
    // Construct a no op handler
    let mut nop = NOP {};

    // In normal operations you would want to loop this connection
    // To reconnect after a disconnect or error
    let (mut network, client) = NetworkBuilder::new_from_client_id("mqrsttSmolExample").smol_network();
    let stream = smol::net::TcpStream::connect(("broker.emqx.io", 1883)).await.unwrap();
    network.connect(stream, &mut nop).await.unwrap();

    // This subscribe is only processed when we run the network
    client.subscribe("mqrstt").await.unwrap();

    let (result, _) = futures::join!(network.run(&mut nop), async {
        smol::Timer::after(std::time::Duration::from_secs(30)).await;
        client.disconnect().await.unwrap();
    });
    assert!(result.is_ok());
    assert_eq!(result.unwrap(), NetworkStatus::OutgoingDisconnect);
});
```

--------------------------------

### subscribe method with examples

Source: https://docs.rs/mqrstt/latest/src/mqrstt/client.rs.html

The subscribe method creates a subscribe packet and sends it to the network stack. Examples show various input types: string, tuple with QoS, vector of tuples, slice of tuples, and tuple with SubscriptionOptions.

```rust
/// Creates a subscribe packet that is then asynchronously transferred to the Network stack for transmission
///
/// Can be called with anything that can be converted into [`SubscribeTopics`]
///
/// # Examples
/// ```
/// # let (network, mqtt_client) = mqrstt::NetworkBuilder::<(), smol::net::TcpStream>::new_from_client_id("example_id").smol_network();
/// # smol::block_on(async {
///
/// use mqrstt::packets::QoS;
/// use mqrstt::packets::{SubscriptionOptions, RetainHandling};
///
/// // retain_handling: RetainHandling::ZERO, retain_as_publish: false, no_local: false, qos: QoS::AtMostOnce,
/// mqtt_client.subscribe("test/topic").await;
///
/// // retain_handling: RetainHandling::ZERO, retain_as_publish: false, no_local: false, qos: QoS::ExactlyOnce,
/// mqtt_client.subscribe(("test/topic", QoS::ExactlyOnce)).await;
///
/// let vec = vec![("test/topic1", QoS::ExactlyOnce), ("test/topic2", QoS::AtMostOnce)];
/// mqtt_client.subscribe(vec).await;
///
/// let vec = [("test/topic1", QoS::ExactlyOnce), ("test/topic2", QoS::AtLeastOnce)];
/// mqtt_client.subscribe(vec.as_slice()).await;
///
/// let sub_options = SubscriptionOptions{
///   retain_handling: RetainHandling::TWO,
///   retain_as_publish: false,
///   no_local: false,
///   qos: QoS::AtLeastOnce,
/// };
/// mqtt_client.subscribe(("final/test/topic", sub_options)).await;
/// # });
/// ```
pub async fn subscribe<A: Into<SubscribeTopics>>(&self, into_subscribtions: A) -> Result<(), ClientError> {
    let pkid = self.available_packet_ids_r.recv().await.map_err(|_| ClientError::NoNetworkChannel)?;
    let subscription: SubscribeTopics = into_subscribtions.into();
    let sub = Subscribe::new(pkid, subscription.0);

    sub.validate(self.max_packet_size)?;
    self.to_network_s.send(Packet::Subscribe(sub)).await.map_err(|_| ClientError::NoNetworkChannel)?;
    #[cfg(feature = "logs")]
    info!("Send to network: Subscribe with ID {:?}", pkid);
    Ok(())
}
```

--------------------------------

### disconnect usage example

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.MqttClient.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Demonstrates how to disconnect the MQTT client and await the result.

```rust
mqtt_client.disconnect().await.unwrap();
```

--------------------------------

### publish_with_properties usage with QoS and retain flags

Source: https://docs.rs/mqrstt/latest/src/mqrstt/client.rs.html

Shows how to call publish_with_properties with different QoS levels, retain flags, and payload types (string, byte slice, Vec<u8>). The example is part of the method's doc comment and uses a hidden network setup.

```rust
///     response_topic: Some("response/topic".into()),
///     correlation_data: Some("correlation_data".into()),
///     ..Default::default()
/// };
///
/// # let properties_clone = properties.clone();
///
/// // publish a message with QoS 0, without a packet identifier
/// mqtt_client.publish_with_properties("test/topic", QoS::AtMostOnce, false, "Hello world", properties).await;
///
/// # let properties = properties_clone.clone();
///
/// // publish a message with QoS 1, with a packet identifier
/// mqtt_client.publish_with_properties("test/topic", QoS::AtLeastOnce, false, "Hello world", properties).await;
///
/// # let properties = properties_clone.clone();
///
/// // publish a message with QoS 2, with a packet identifier
/// mqtt_client.publish_with_properties("test/topic", QoS::ExactlyOnce, false, "Hello world", properties).await;
///
/// # let properties = properties_clone.clone();
///
/// // publish a message with QoS 1, with a packet identifier, and the "retain" flag set
/// let payload = "Hello World!".as_bytes();
/// mqtt_client.publish_with_properties("test/topic", QoS::AtLeastOnce, true, payload, properties).await;
///
/// # let properties = properties_clone.clone();
///
/// // publish a message with QoS 1, with a packet identifier, and the "retain" flag set
/// let payload = "Hello World!".as_bytes().to_vec();
/// mqtt_client.publish_with_properties("test/topic", QoS::AtMostOnce, true, payload, properties).await;
///
/// # });
/// # let _network = std::hint::black_box(network);
```

--------------------------------

### publish_with_properties_blocking usage examples

Source: https://docs.rs/mqrstt/latest/src/mqrstt/client.rs.html

Shows how to call publish_with_properties_blocking with different QoS levels, retain flags, and payload types (string, byte slice, Vec<u8>). The example is part of the method's doc comment and uses a pre-configured mqtt_client from the surrounding context.

```rust
///     correlation_data: Some("correlation_data".into()),
///     ..Default::default()
/// };
///
/// # let properties_clone = properties.clone();
///
/// // publish a message with QoS 0, without a packet identifier
/// mqtt_client.publish_with_properties_blocking("test/topic", QoS::AtMostOnce, false, "Hello world", properties).unwrap();
///
/// # let properties = properties_clone.clone();
///
/// // publish a message with QoS 1, with a packet identifier
/// mqtt_client.publish_with_properties_blocking("test/topic", QoS::AtLeastOnce, false, "Hello world", properties).unwrap();
///
/// # let properties = properties_clone.clone();
///
/// // publish a message with QoS 2, with a packet identifier
/// mqtt_client.publish_with_properties_blocking("test/topic", QoS::ExactlyOnce, false, "Hello world", properties).unwrap();
///
/// # let properties = properties_clone.clone();
///
/// // publish a message with QoS 1, with a packet identifier, and the "retain" flag set
/// let payload = b"Hello World!";
/// mqtt_client.publish_with_properties_blocking("test/topic", QoS::AtLeastOnce, true, payload, properties).unwrap();
///
/// # let properties = properties_clone.clone();
///
/// // publish a message with QoS 1, with a packet identifier, and the "retain" flag set
/// let payload = b"Hello World!".to_vec();
/// mqtt_client.publish_with_properties_blocking("test/topic", QoS::AtMostOnce, true, payload, properties).unwrap();
///
/// # });
/// ```
```

--------------------------------

### ConnectOptions::set_clean_start

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.ConnectOptions.html

Sets the clean start flag.

```APIDOC
## ConnectOptions::set_clean_start

### Description
Sets whether the connection starts a new session or continues an existing session. This is specified by the MQTT v5.0 specification.

### Method
`pub fn set_clean_start(&mut self, clean_start: bool) -> &mut Self`

### Parameters
- **clean_start** (`bool`) - Required - `true` to start a new session, `false` to continue an existing session.

### Returns
- `&mut Self` - A mutable reference to the `ConnectOptions` for chaining.
```

--------------------------------

### smol_network example with new_from_client_id

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.NetworkBuilder.html?search=

Creates a Network and MqttClient using smol's TcpStream. Uses new_from_client_id to set the client ID.

```rust
let (mut network, client) = mqrstt::NetworkBuilder::<(), smol::net::TcpStream>
    ::new_from_client_id("ExampleClient")
    .smol_network();
```

--------------------------------

### unsubscribe_with_properties usage examples

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.MqttClient.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Shows how to unsubscribe from MQTT topics with additional properties, using different topic input types and a UnsubscribeProperties with user_properties.

```rust
use mqrstt::packets::UnsubscribeProperties;

let properties = UnsubscribeProperties{
    user_properties: vec![("property".into(), "value".into())],
};

// Unsubscribe from a single topic specified as a string:
let topic = "test/topic";
mqtt_client.unsubscribe_with_properties(topic, properties).await;


// Unsubscribe from multiple topics specified as an array of string slices:
let topics = &["test/topic1", "test/topic2"];
mqtt_client.unsubscribe_with_properties(topics.as_slice(), properties).await;

  
// Unsubscribe from a single topic specified as a String:
let topic = String::from("test/topic");
mqtt_client.unsubscribe_with_properties(topic, properties).await;

  
// Unsubscribe from multiple topics specified as a Vec<String>:
let topics = vec![String::from("test/topic1"), String::from("test/topic2")];
mqtt_client.unsubscribe_with_properties(topics, properties).await;

  
// Unsubscribe from multiple topics specified as an array of String:
let topics = &[String::from("test/topic1"), String::from("test/topic2")];
mqtt_client.unsubscribe_with_properties(topics.as_slice(), properties).await;
```

--------------------------------

### Publish messages with different QoS and retain flags

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.MqttClient.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Examples of using the publish method with various QoS levels, retain flags, and payload types.

```rust
use mqrstt::packets::QoS;

// publish a message with QoS 0, without a packet identifier
mqtt_client.publish("test/topic", QoS::AtMostOnce, false, "Hello world").await;

// publish a message with QoS 1, with a packet identifier
mqtt_client.publish("test/topic", QoS::AtLeastOnce, false, b"Hello world").await;

// publish a message with QoS 2, with a packet identifier
mqtt_client.publish("test/topic", QoS::ExactlyOnce, false, "Hello world").await;

// publish a message with QoS 1, with a packet identifier, and the "retain" flag set
let payload: &[u8] = "Hello World!".as_bytes();
mqtt_client.publish("test/topic", QoS::AtLeastOnce, true, payload).await;

// publish a message with QoS 1, with a packet identifier, and the "retain" flag set
let payload: Vec<u8> = "Hello World!".as_bytes().to_vec();
mqtt_client.publish("test/topic", QoS::AtMostOnce, true, payload).await;
```

--------------------------------

### Test outgoing subscribe setup

Source: https://docs.rs/mqrstt/latest/src/mqrstt/state_handler.rs.html

Starts a test for outgoing subscribe by creating a handler, receiving a packet identifier, and constructing a subscribe packet. The test is incomplete in the source.

```rust
#[tokio::test]
async fn outgoing_subscribe() {
    let (mut handler, apkid) = handler(false);

    let pkid = apkid.recv().await.unwrap();

    let sub_packet = create_subscribe_packet(pkid);
```

--------------------------------

### tokio_network example with ConnectOptions

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.NetworkBuilder.html?search=

Creates a Network and MqttClient using tokio's TcpStream. Requires the tokio feature and a stream implementing tokio's AsyncRead and AsyncWrite.

```rust
use mqrstt::ConnectOptions;

let options = ConnectOptions::new("ExampleClient");
let (mut network, client) = mqrstt::NetworkBuilder::<(), tokio::net::TcpStream>
    ::new_from_options(options)
    .tokio_network();
```

--------------------------------

### Connect Method

Source: https://docs.rs/mqrstt/latest/mqrstt/smol/struct.Network.html?search=

Initializes an MQTT connection with the provided configuration and stream.

```rust
pub async fn connect(&mut self, stream: S, handler: &mut H) -> Result<(), ConnectionError>
```

--------------------------------

### publish_with_properties usage examples

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.MqttClient.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Shows how to publish MQTT messages with additional properties, demonstrating usage with different QoS levels, retain flags, and payload types.

```rust
use mqrstt::packets::QoS;
use mqrstt::packets::PublishProperties;

let properties = PublishProperties{
    response_topic: Some("response/topic".into()),
    correlation_data: Some("correlation_data".into()),
    ..Default::default()
};


// publish a message with QoS 0, without a packet identifier
mqtt_client.publish_with_properties("test/topic", QoS::AtMostOnce, false, "Hello world", properties).await;


// publish a message with QoS 1, with a packet identifier
mqtt_client.publish_with_properties("test/topic", QoS::AtLeastOnce, false, "Hello world", properties).await;


// publish a message with QoS 2, with a packet identifier
mqtt_client.publish_with_properties("test/topic", QoS::ExactlyOnce, false, "Hello world", properties).await;


// publish a message with QoS 1, with a packet identifier, and the "retain" flag set
let payload = "Hello World!".as_bytes();
mqtt_client.publish_with_properties("test/topic", QoS::AtLeastOnce, true, payload, properties).await;


// publish a message with QoS 1, with a packet identifier, and the "retain" flag set
let payload = "Hello World!".as_bytes().to_vec();
mqtt_client.publish_with_properties("test/topic", QoS::AtMostOnce, true, payload, properties).await;
```

--------------------------------

### unsubscribe usage with various topic types

Source: https://docs.rs/mqrstt/latest/src/mqrstt/client.rs.html

Shows how to call unsubscribe with different topic types: a string, an array of string slices, a String, a Vec<String>, and an array of String. The example is part of the method's doc comment and uses a hidden network setup.

```rust
# let (network, mqtt_client) = mqrstt::NetworkBuilder::<(), smol::net::TcpStream>::new_from_client_id("Example").smol_network();
# smol::block_on(async {

// Unsubscribe from a single topic specified as a string:
let topic = "test/topic";
mqtt_client.unsubscribe(topic).await;

// Unsubscribe from multiple topics specified as an array of string slices:
let topics = &["test/topic1", "test/topic2"];
mqtt_client.unsubscribe(topics.as_slice()).await;

// Unsubscribe from a single topic specified as a String:
let topic = String::from("test/topic");
mqtt_client.unsubscribe(topic).await;

// Unsubscribe from multiple topics specified as a Vec<String>:
let topics = vec![String::from("test/topic1"), String::from("test/topic2")];
mqtt_client.unsubscribe(topics).await;

// Unsubscribe from multiple topics specified as an array of String:
let topics = &[String::from("test/topic1"), String::from("test/topic2")];
mqtt_client.unsubscribe(topics.as_slice()).await;

# });
# let _network = std::hint::black_box(network);
```

--------------------------------

### subscribe_with_properties_blocking with Vec, slice, and SubscriptionOptions

Source: https://docs.rs/mqrstt/latest/src/mqrstt/client.rs.html

Shows three ways to call subscribe_with_properties_blocking: with a Vec of (topic, QoS) tuples, with a slice of the same, and with a single (topic, SubscriptionOptions) tuple. The first two use the same QoS for all topics; the third uses per-topic options. The hidden lines are part of the doc-test setup and are not shown in the rendered example.

```rust
let vec = vec![("test/topic1", QoS::ExactlyOnce), ("test/topic2", QoS::ExactlyOnce)];
mqtt_client.subscribe_with_properties_blocking(vec, sub_properties).unwrap();

# let sub_properties = sub_properties_clone.clone();
 
let vec = [("test/topic1", QoS::ExactlyOnce), ("test/topic2", QoS::ExactlyOnce)];
mqtt_client.subscribe_with_properties_blocking(vec.as_slice(), sub_properties).unwrap();

# let sub_properties = sub_properties_clone.clone();
 
let sub_options = SubscriptionOptions{
  retain_handling: RetainHandling::ZERO,
  retain_as_publish: false,
  no_local: false,
  qos: QoS::AtLeastOnce,
};
mqtt_client.subscribe_with_properties_blocking(("final/test/topic", sub_options), sub_properties).unwrap();
```

--------------------------------

### connect

Source: https://docs.rs/mqrstt/latest/mqrstt/tokio/struct.Network.html

Initializes an MQTT connection with the provided configuration and stream. It is recommended to use a buffered stream.

```APIDOC
## connect

### Description
Initializes an MQTT connection with the provided configuration and stream. It is recommended to use a buffered stream.

### Method
async fn connect

### Signature
pub async fn connect(&mut self, stream: S, handler: &mut H) -> Result<(), ConnectionError>

### Parameters
- **stream** (S) - Required - The network stream to use for the connection.
- **handler** (&mut H) - Required - The event handler for MQTT events.

### Returns
- **Result<(), ConnectionError>** - Ok on success, or an error if the connection fails.
```

--------------------------------

### disconnect() example

Source: https://docs.rs/mqrstt/latest/src/mqrstt/client.rs.html

Demonstrates calling `disconnect()` on an MQTT client to send a Disconnect packet. The example is wrapped in `smol::block_on` and uses a `NetworkBuilder` with `smol::net::TcpStream`. After disconnecting, no further packets are transmitted.

```rust
/// # let (network, mqtt_client) = mqrstt::NetworkBuilder::<(), smol::net::TcpStream>::new_from_client_id("Example").smol_network();
/// # smol::block_on(async {
///
/// mqtt_client.disconnect().await.unwrap();
///
/// # });
/// # let _network = std::hint::black_box(network);
```

--------------------------------

### Network connect method

Source: https://docs.rs/mqrstt/latest/src/mqrstt/smol/network.rs.html?search=u32+-%3E+bool

Initializes an MQTT connection with the provided stream and handler, handling CONNACK and keep-alive settings.

```rust
62impl<H, S> Network<H, S>
63where
64    H: AsyncEventHandler,
65    S: smol::io::AsyncRead + smol::io::AsyncWrite + Sized + Unpin,
66{
67    /// Initializes an MQTT connection with the provided configuration an stream
68    pub async fn connect(&mut self, stream: S, handler: &mut H) -> Result<(), ConnectionError> {
69        let (mut network, conn_ack) = Stream::connect(&self.options, stream).await?;
70        self.last_network_action = Instant::now();
71
72        if let Some(keep_alive_interval) = conn_ack.connack_properties.server_keep_alive {
73            self.keep_alive_interval = Duration::from_secs(keep_alive_interval as u64);
74        }
75        if self.keep_alive_interval.is_zero() {
76            self.perform_keep_alive = false;
77        }
78
79        let packets = self.state_handler.handle_incoming_connack(&conn_ack)?;
80        handler.handle(Packet::ConnAck(conn_ack)).await;
81        if let Some(mut packets) = packets {
82            network.write_all(&mut packets).await?;
83            self.last_network_action = Instant::now();
84        }
85
86        self.network = Some(network);
87
88        Ok(())
89    }
```

--------------------------------

### ConnectOptions::get_username

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.ConnectOptions.html

Gets the username.

```APIDOC
## ConnectOptions::get_username

### Description
Returns the username for the MQTT connection, if set.

### Method
`pub fn get_username(&self) -> Option<&str>`

### Returns
- `Option<&str>` - The username if set, otherwise `None`.
```

--------------------------------

### Network::connect

Source: https://docs.rs/mqrstt/latest/mqrstt/smol/struct.Network.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Initializes an MQTT connection with the provided configuration and stream.

```APIDOC
## [ASYNC] Network::connect

### Description
Initializes an MQTT connection with the provided configuration and stream.

### Method
async fn connect

### Signature
pub async fn connect(&mut self, stream: S, handler: &mut H) -> Result<(), ConnectionError>

### Parameters
- **stream** (S) - Required - The async stream (e.g., TCP or TLS) to use for the connection.
- **handler** (&mut H) - Required - The event handler for MQTT events.

### Returns
- **Result<(), ConnectionError>** - Ok(()) on success, or a ConnectionError on failure.
```

--------------------------------

### ConnectOptions::send_maximum

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.ConnectOptions.html

Gets the send maximum.

```APIDOC
## ConnectOptions::send_maximum

### Description
Returns the send maximum for the MQTT connection.

### Method
`pub fn send_maximum(&self) -> u16`

### Returns
- `u16` - The send maximum value.
```

--------------------------------

### ConnectOptions::receive_maximum

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.ConnectOptions.html

Gets the receive maximum.

```APIDOC
## ConnectOptions::receive_maximum

### Description
Returns the receive maximum for the MQTT connection.

### Method
`pub fn receive_maximum(&self) -> u16`

### Returns
- `u16` - The receive maximum value.
```

--------------------------------

### Publish MQTT messages with properties and various QoS levels

Source: https://docs.rs/mqrstt/latest/src/mqrstt/client.rs.html?search=std%3A%3Avec

Shows how to call `publish_with_properties` with different QoS levels (AtMostOnce, AtLeastOnce, ExactlyOnce), the retain flag, and payload types (`&str`, `&[u8]`, `Vec<u8>`). The example uses a `PublishProperties` with `response_topic` and `correlation_data` set, and runs inside `smol::block_on`.

```rust
let properties = PublishProperties {
    response_topic: Some("response/topic".into()),
    correlation_data: Some("correlation_data".into()),
    ..Default::default()
};

# let properties_clone = properties.clone();

// publish a message with QoS 0, without a packet identifier
mqtt_client.publish_with_properties("test/topic", QoS::AtMostOnce, false, "Hello world", properties).await;

# let properties = properties_clone.clone();

// publish a message with QoS 1, with a packet identifier
mqtt_client.publish_with_properties("test/topic", QoS::AtLeastOnce, false, "Hello world", properties).await;

# let properties = properties_clone.clone();

// publish a message with QoS 2, with a packet identifier
mqtt_client.publish_with_properties("test/topic", QoS::ExactlyOnce, false, "Hello world", properties).await;

# let properties = properties_clone.clone();

// publish a message with QoS 1, with a packet identifier, and the "retain" flag set
let payload = "Hello World!".as_bytes();
mqtt_client.publish_with_properties("test/topic", QoS::AtLeastOnce, true, payload, properties).await;

# let properties = properties_clone.clone();

// publish a message with QoS 1, with a packet identifier, and the "retain" flag set
let payload = "Hello World!".as_bytes().to_vec();
mqtt_client.publish_with_properties("test/topic", QoS::AtMostOnce, true, payload, properties).await;

# });
# let _network = std::hint::black_box(network);
```

--------------------------------

### ConnectOptions::maximum_packet_size

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.ConnectOptions.html

Gets the maximum packet size.

```APIDOC
## ConnectOptions::maximum_packet_size

### Description
Returns the maximum packet size for the MQTT connection.

### Method
`pub fn maximum_packet_size(&self) -> usize`

### Returns
- `usize` - The maximum packet size.
```

--------------------------------

### ConnectOptions::get_keep_alive_interval

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.ConnectOptions.html

Gets the keep alive interval.

```APIDOC
## ConnectOptions::get_keep_alive_interval

### Description
Returns the current keep alive interval for the MQTT protocol. This time is used in Ping Pong when natural traffic is absent. The granularity is in seconds.

### Method
`pub fn get_keep_alive_interval(&self) -> Duration`

### Returns
- `Duration` - The keep alive interval.
```

--------------------------------

### ConnectOptions::get_session_expiry_interval

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.ConnectOptions.html

Gets the session expiry interval.

```APIDOC
## ConnectOptions::get_session_expiry_interval

### Description
Returns the session expiry interval in seconds. If absent, the value 0 is used. If set to 0, the session ends when the network connection is closed. If set to `UINT_MAX`, the session does not expire.

### Method
`pub fn get_session_expiry_interval(&self) -> Option<u32>`

### Returns
- `Option<u32>` - The session expiry interval in seconds, or `None` if not set.
```

--------------------------------

### new

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.ConnectOptions.html?search=u32+-%3E+bool

Creates a new `ConnectOptions` instance with the given client identifier.

```APIDOC
## new

### Description
Creates a new `ConnectOptions` instance with the given client identifier.

### Parameters
- **client_id** (`S: AsRef<str>`) - Required - The client identifier for the MQTT connection.

### Returns
- `Self` - A new `ConnectOptions` instance.

```

--------------------------------

### example_handlers module with NOP, PingResp, and PingPong handlers

Source: https://docs.rs/mqrstt/latest/src/mqrstt/event_handlers.rs.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Contains example handler implementations: NOP (no-op), PingResp (counts PingResp packets), and PingPong (responds to publish payloads containing 'ping' with a 'pong' message).

```rust
pub mod example_handlers {
    use std::{ops::AddAssign, sync::atomic::AtomicU16};

    use crate::{
        AsyncEventHandler, EventHandler, MqttClient,
        packets::{self, Packet},
    };

    /// Most basic no op handler
    /// This handler performs no operations on incoming messages.
    pub struct NOP {}

    impl AsyncEventHandler for NOP {
        async fn handle(&mut self, _: Packet) {}
    }

    impl EventHandler for NOP {
        fn handle(&mut self, _: Packet) {}
    }

    #[derive(Debug)]
    pub struct PingResp {
        pub client: MqttClient,
        pub ping_resp_received: u32,
    }

    impl PingResp {
        pub fn new(client: MqttClient) -> Self {
            Self { client, ping_resp_received: 0 }
        }
    }

    impl AsyncEventHandler for PingResp {
        async fn handle(&mut self, event: packets::Packet) {
            use Packet::*;
            if event == PingResp {
                self.ping_resp_received += 1;
            }
            println!("Received packet: {}", event);
        }
    }

    impl EventHandler for PingResp {
        fn handle(&mut self, event: Packet) {
            use Packet::*;
            if event == PingResp {
                self.ping_resp_received += 1;
            }
            println!("Received packet: {}", event);
        }
    }

    pub struct PingPong {
        pub client: MqttClient,
        pub number: AtomicU16,
    }

    impl PingPong {
        pub fn new(client: MqttClient) -> Self {
            Self { client, number: AtomicU16::new(0) }
        }
    }

    impl AsyncEventHandler for PingPong {
        async fn handle(&mut self, event: packets::Packet) {
            match event {
                Packet::Publish(p) => {
                    if let Ok(payload) = String::from_utf8(p.payload.to_vec()) {
                        let max_len = payload.len().min(10);
                        let a = &payload[0..max_len];
                        if payload.to_lowercase().contains("ping") {
                            self.client.publish(p.topic.clone(), p.qos, p.retain, b"pong").await.unwrap();
                            println!("Received publish payload: {}", a);

                            if !p.retain {
                                self.number.get_mut().add_assign(1);
                            }

                            println!("DBG: \n {}", &Packet::Publish(p));
                        }
                    }
                }
                Packet::ConnAck(_) => {
                    println!("Connected!")
                }
                _ => (),
            }
        }
    }
}
```

--------------------------------

### Subscribe with properties

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.MqttClient.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Examples of using subscribe_with_properties with different input types and a SubscribeProperties object.

```rust
use mqrstt::packets::QoS;
use mqrstt::packets::{SubscribeProperties, SubscriptionOptions, RetainHandling};

let sub_properties = SubscribeProperties{
  subscription_identifier: Some(1),
  user_properties: vec![],
};

let sub_properties_clone = sub_properties.clone();

// retain_handling: RetainHandling::ZERO, retain_as_publish: false, no_local: false, qos: QoS::AtMostOnce,
mqtt_client.subscribe_with_properties("test/topic", sub_properties).await;


// retain_handling: RetainHandling::ZERO, retain_as_publish: false, no_local: false, qos: QoS::ExactlyOnce,
mqtt_client.subscribe_with_properties(("test/topic", QoS::ExactlyOnce), sub_properties).await;


let vec = vec![("test/topic1", QoS::ExactlyOnce), ("test/topic2", QoS::ExactlyOnce)];
mqtt_client.subscribe_with_properties(vec, sub_properties).await;

  
let vec = [("test/topic1", QoS::ExactlyOnce), ("test/topic2", QoS::ExactlyOnce)];
mqtt_client.subscribe_with_properties(vec.as_slice(), sub_properties).await;

  
let sub_options = SubscriptionOptions{
  retain_handling: RetainHandling::ZERO,
  retain_as_publish: false,
  no_local: false,
  qos: QoS::AtLeastOnce,
};
mqtt_client.subscribe_with_properties(("final/test/topic", sub_options), sub_properties).await;
```

--------------------------------

### ConnectOptions Struct Methods

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.ConnectOptions.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Documentation for the ConnectOptions struct and its methods in the mqrstt crate.

```APIDOC
## ConnectOptions::new

### Description
Creates a new `ConnectOptions` instance with the given client identifier. The client identifier is not restricted by this client, but it is recommended to follow the MQTT v5.0 specification (1 to 23 bytes of UTF-8, containing only [a-zA-Z0-9]).

### Method
`pub fn new<S: AsRef<str>>(client_id: S) -> Self`

### Parameters
- **client_id** (`S`) - Required - The client identifier for the MQTT connection.

### Returns
- `Self` - A new `ConnectOptions` instance.

### Example
```rust
let options = ConnectOptions::new("my-client-id");
```
```

```APIDOC
## ConnectOptions::get_client_id

### Description
Returns the client identifier (ClientID) that identifies the client to the server. Each client connecting to the server must have a unique ClientID. Non-unique client IDs can cause connect/disconnect loops due to auto reconnects and forced disconnects.

### Method
`pub fn get_client_id(&self) -> &str`

### Returns
- `&str` - The client identifier.

### Example
```rust
let client_id = options.get_client_id();
```
```

```APIDOC
## ConnectOptions::set_client_id

### Description
Sets the client identifier (ClientID) for the MQTT connection. The client identifier must be unique per client. Non-unique client IDs can cause connect/disconnect loops due to auto reconnects and forced disconnects.

### Method
`pub fn set_client_id<S: AsRef<str>>(&mut self, client_id: S) -> &mut Self`

### Parameters
- **client_id** (`S`) - Required - The new client identifier.

### Returns
- `&mut Self` - A mutable reference to the `ConnectOptions` for chaining.

### Example
```rust
options.set_client_id("new-client-id");
```
```

```APIDOC
## ConnectOptions::get_clean_start

### Description
Returns whether the connection starts a new session or continues an existing session. See the MQTT v5.0 specification for more details.

### Method
`pub fn get_clean_start(&self) -> bool`

### Returns
- `bool` - `true` if a new session is started, `false` if an existing session is continued.

### Example
```rust
let clean_start = options.get_clean_start();
```
```

```APIDOC
## ConnectOptions::set_clean_start

### Description
Sets whether the connection starts a new session or continues an existing session. See the MQTT v5.0 specification for more details.

### Method
`pub fn set_clean_start(&mut self, clean_start: bool) -> &mut Self`

### Parameters
- **clean_start** (`bool`) - Required - `true` to start a new session, `false` to continue an existing session.

### Returns
- `&mut Self` - A mutable reference to the `ConnectOptions` for chaining.

### Example
```rust
options.set_clean_start(true);
```
```

```APIDOC
## ConnectOptions::get_username

### Description
Returns the username used for authentication with the MQTT broker, if set.

### Method
`pub fn get_username(&self) -> Option<&str>`

### Returns
- `Option<&str>` - The username, or `None` if not set.

### Example
```rust
if let Some(username) = options.get_username() {
    println!("Username: {}", username);
}
```
```

```APIDOC
## ConnectOptions::set_username

### Description
Sets the username used for authentication with the MQTT broker.

### Method
`pub fn set_username<S: AsRef<str>>(&mut self, username: S) -> &mut Self`

### Parameters
- **username** (`S`) - Required - The username to set.

### Returns
- `&mut Self` - A mutable reference to the `ConnectOptions` for chaining.

### Example
```rust
options.set_username("my-username");
```
```

```APIDOC
## ConnectOptions::get_password

### Description
Returns the password used for authentication with the MQTT broker, if set.

### Method
`pub fn get_password(&self) -> Option<&str>`

### Returns
- `Option<&str>` - The password, or `None` if not set.

### Example
```rust
if let Some(password) = options.get_password() {
    println!("Password: {}", password);
}
```
```

```APIDOC
## ConnectOptions::set_password

### Description
Sets the password used for authentication with the MQTT broker.

### Method
`pub fn set_password<S: AsRef<str>>(&mut self, password: S) -> &mut Self`

### Parameters
- **password** (`S`) - Required - The password to set.

### Returns
- `&mut Self` - A mutable reference to the `ConnectOptions` for chaining.

### Example
```rust
options.set_password("my-password");
```
```

```APIDOC
## ConnectOptions::get_session_expiry_interval

### Description
Returns the session expiry interval in seconds. If the interval is absent, the value 0 is used. A value of 0 means the session ends when the network connection is closed. A value of `0xFFFFFFFF` (UINT_MAX) means the session does not expire.

### Method
`pub fn get_session_expiry_interval(&self) -> Option<u32>`

### Returns
- `Option<u32>` - The session expiry interval in seconds, or `None` if not set.

### Example
```rust
let interval = options.get_session_expiry_interval();
```
```

```APIDOC
## ConnectOptions::set_session_expiry_interval

### Description
Sets the session expiry interval in seconds. A value of 0 means the session ends when the network connection is closed. A value of `0xFFFFFFFF` (UINT_MAX) means the session does not expire.

### Method
`pub fn set_session_expiry_interval(&mut self, session_expiry_interval: u32) -> &mut Self`

### Parameters
- **session_expiry_interval** (`u32`) - Required - The session expiry interval in seconds.

### Returns
- `&mut Self` - A mutable reference to the `ConnectOptions` for chaining.

### Example
```rust
options.set_session_expiry_interval(3600);
```
```

```APIDOC
## ConnectOptions::get_keep_alive_interval

### Description
Returns the current keep alive interval for the MQTT protocol. This time is used in Ping Pong when natural traffic is absent. The granularity is in seconds.

### Method
`pub fn get_keep_alive_interval(&self) -> Duration`

### Returns
- `Duration` - The keep alive interval.

### Example
```rust
let interval = options.get_keep_alive_interval();
```
```

```APIDOC
## ConnectOptions::set_keep_alive_interval

### Description
Sets the keep alive interval for the MQTT protocol. This time is used in Ping Pong when natural traffic is absent. The granularity is in seconds.

### Method
`pub fn set_keep_alive_interval(&mut self, keep_alive_interval: Duration) -> &mut Self`

### Parameters
- **keep_alive_interval** (`Duration`) - Required - The keep alive interval.

### Returns
- `&mut Self` - A mutable reference to the `ConnectOptions` for chaining.

### Example
```rust
options.set_keep_alive_interval(Duration::from_secs(60));
```
```

```APIDOC
## ConnectOptions::set_last_will

### Description
Sets the last will message for the MQTT connection.

### Method
`pub fn set_last_will(&mut self, last_will: LastWill) -> &mut Self`

### Parameters
- **last_will** (`LastWill`) - Required - The last will message.

### Returns
- `&mut Self` - A mutable reference to the `ConnectOptions` for chaining.

### Example
```rust
options.set_last_will(last_will);
```
```

```APIDOC
## ConnectOptions::get_last_will

### Description
Returns the last will message for the MQTT connection, if set.

### Method
`pub fn get_last_will(&self) -> Option<&LastWill>`

### Returns
- `Option<&LastWill>` - The last will message, or `None` if not set.

### Example
```rust
if let Some(last_will) = options.get_last_will() {
    // use last_will
}
```
```

```APIDOC
## ConnectOptions::set_receive_maximum

### Description
Sets the receive maximum value for the MQTT connection.

### Method
`pub fn set_receive_maximum(&mut self, receive_maximum: u16) -> &mut Self`

### Parameters
- **receive_maximum** (`u16`) - Required - The receive maximum value.

### Returns
- `&mut Self` - A mutable reference to the `ConnectOptions` for chaining.

### Example
```rust
options.set_receive_maximum(10);
```
```

```APIDOC
## ConnectOptions::receive_maximum

### Description
Returns the receive maximum value for the MQTT connection.

### Method
`pub fn receive_maximum(&self) -> u16`

### Returns
- `u16` - The receive maximum value.

### Example
```rust
let max = options.receive_maximum();
```
```

```APIDOC
## ConnectOptions::set_maximum_packet_size

### Description
Sets the maximum packet size for the MQTT connection. Returns an error if the value is invalid.

### Method
`pub fn set_maximum_packet_size(&mut self, maximum_packet_size: u32) -> Result<&mut Self, ConnectOptionsError>`

### Parameters
- **maximum_packet_size** (`u32`) - Required - The maximum packet size in bytes.

### Returns
- `Result<&mut Self, ConnectOptionsError>` - A mutable reference to the `ConnectOptions` on success, or an error.

### Example
```rust
options.set_maximum_packet_size(1024)?;
```
```

```APIDOC
## ConnectOptions::maximum_packet_size

### Description
Returns the maximum packet size for the MQTT connection.

### Method
`pub fn maximum_packet_size(&self) -> usize`

### Returns
- `usize` - The maximum packet size in bytes.

### Example
```rust
let max_size = options.maximum_packet_size();
```
```

```APIDOC
## ConnectOptions::set_send_maximum

### Description
Sets the send maximum value for the MQTT connection.

### Method
`pub fn set_send_maximum(&mut self, send_maximum: u16) -> &mut Self`

### Parameters
- **send_maximum** (`u16`) - Required - The send maximum value.

### Returns
- `&mut Self` - A mutable reference to the `ConnectOptions` for chaining.

### Example
```rust
options.set_send_maximum(10);
```
```

```APIDOC
## ConnectOptions::send_maximum

### Description
Returns the send maximum value for the MQTT connection.

### Method
`pub fn send_maximum(&self) -> u16`

### Returns
- `u16` - The send maximum value.

### Example
```rust
let max = options.send_maximum();
```
```

--------------------------------

### set_clean_start

Source: https://docs.rs/mqrstt/latest/mqrstt/struct.ConnectOptions.html?search=u32+-%3E+bool

Sets whether the connection starts a new session or continues an existing one.

```APIDOC
## set_clean_start

### Description
Sets whether the connection starts a new session or continues an existing one.

### Parameters
- **clean_start** (`bool`) - Required - The new clean start value.

### Returns
- `&mut Self` - The mutable reference to `self` for chaining.

```
Tokens:
Raw