Struct MqttStream

pub struct MqttStream<Io> {
    pub io: Framed<Io, MqttCodec>,
    pub remote_addr: SocketAddr,
    pub cfg: Arc<Builder>,
}

MQTT v3.1.1 protocol stream implementation

Fields

io: Framed<Io, MqttCodec>

Framed IO layer with MQTT codec

remote_addr: SocketAddr

Remote client’s network address

cfg: Arc<Builder>

Shared configuration builder

Implementations

impl<Io> MqttStream<Io>

where Io: AsyncRead + AsyncWrite + Unpin,

Examples

use std::net::{SocketAddr, IpAddr, Ipv4Addr};
use std::sync::Arc;
use tokio::net::TcpStream;
use tokio_util::codec::Framed;
use rmqtt_codec::{MqttCodec, types::Publish};
use rmqtt_net::{Builder,v3};

let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 1883);
let stream = TcpStream::connect(addr).await?;
let mut mqtt_stream = v3::MqttStream {
    io: Framed::new(stream, MqttCodec::V3(Default::default())),
    remote_addr: addr,
    cfg: Arc::new(Builder::default()),
};

// Send a PING request
mqtt_stream.send_ping_request().await?;

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

Sends DISCONNECT packet and flushes buffers

pub async fn send_publish(&mut self, publish: Box<Publish>) -> Result<()>

Publishes a message to the broker

pub async fn send_publish_ack(&mut self, packet_id: NonZeroU16) -> Result<()>

Acknowledges a received publish (QoS 1)

pub async fn send_publish_received( &mut self, packet_id: NonZeroU16, ) -> Result<()>

Confirms receipt of a publish (QoS 2 step 1)

pub async fn send_publish_release( &mut self, packet_id: NonZeroU16, ) -> Result<()>

Releases a stored publish (QoS 2 step 2)

pub async fn send_publish_complete( &mut self, packet_id: NonZeroU16, ) -> Result<()>

Confirms publish completion (QoS 2 step 3)

pub async fn send_subscribe_ack( &mut self, packet_id: NonZeroU16, status: Vec<SubscribeReturnCode>, ) -> Result<()>

Acknowledges a subscription request

pub async fn send_unsubscribe_ack( &mut self, packet_id: NonZeroU16, ) -> Result<()>

Acknowledges an unsubscribe request

pub async fn send_connect(&mut self, connect: Connect) -> Result<()>

Initiates connection to the broker

pub async fn send_connect_ack( &mut self, return_code: ConnectAckReason, session_present: bool, ) -> Result<()>

Responds to connection request

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

Sends keep-alive ping request

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

Responds to ping request

pub async fn send(&mut self, packet: Packet) -> Result<()>

Generic packet sending method

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

Flushes write buffers

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

Closes the connection gracefully

pub async fn recv(&mut self, tm: Duration) -> Result<Option<Packet>>

Receives next packet with timeout

pub async fn recv_connect(&mut self, tm: Duration) -> Result<Box<Connect>>

Waits for CONNECT packet with timeout

Trait Implementations

impl<Io> Stream for MqttStream<Io>

where Io: AsyncRead + Unpin,

type Item = Result<Packet, Error>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.