Struct Mosquitto 

pub struct Mosquitto { /* private fields */ }

Mosquitto client

Implementations

impl Mosquitto

pub fn new(id: &str) -> Mosquitto

create a new mosquitto instance, providing a client name. Clients connecting to a broker must have unique names

Examples found in repository

examples/publish_wait.rs (line 14)

fn run() -> Result<(),Box<std::error::Error>> {
    let m = Mosquitto::new("test");

    let t = Instant::now();

    m.connect_wait("localhost",1883,TIMEOUT)?;
    m.publish_wait("/bonzo/dog",b"hello dolly",2,false,TIMEOUT)?;
    m.publish_wait("/bonzo/cat",b"meeeaaaww",2,false,TIMEOUT)?;
    println!("elapsed {:.2} msec",as_millis(t.elapsed()));
    Ok(())
}

examples/verify-publish.rs (line 7)

fn go() -> Result<(),Box<Error>> {
    let m = Mosquitto::new("test");

    m.connect("localhost",1883)?;

    // publish and get a message id
    let our_mid = m.publish("bonzo/dog","hello dolly".as_bytes(), 2, false)?;

    // and wait for confirmation for that message id
    let mut mc = m.callbacks(());
    mc.on_publish(|_,mid| {
        if mid == our_mid {
            m.disconnect().unwrap();
        }
    });

    // wait forever until explicit disconnect
    m.loop_until_disconnect(-1)?;
    Ok(())
}

examples/self-publish-receive-many.rs (line 6)

fn run() -> mosq::Result<()> {
    let m = Mosquitto::new("test");

    m.connect_wait("localhost",1883,300)?;
    let bilbo = m.subscribe("bilbo/#",1)?;

    let mt = m.clone();
    thread::spawn(move || {
        for i in 0..5 {
            let topic = format!("bilbo/{}",10*(i+1));
            let data = format!("hello #{}",i);
            mt.publish(&topic,data.as_bytes(), 1, false).unwrap();
        }
    });

    let msgs = bilbo.receive_many(300)?;
    for msg in msgs {
        println!("topic {} text '{}'",msg.topic(),msg.text());
    }
    Ok(())
}

examples/subscribe-and-listen.rs (line 5)

fn main() {
    let m = Mosquitto::new("test");

    m.will_set("test/will",b"finished!",0,false).expect("can't set will");

    m.connect("localhost",1883).expect("can't connect");
    let bonzo = m.subscribe("bonzo/#",0).expect("can't subscribe to bonzo");
    let frodo = m.subscribe("frodo/#",0).expect("can't subscribe to frodo");

    // not interested in any retained messages!
    let mut mc = m.callbacks(());
    mc.on_message(|_,msg| {
        if ! msg.retained() {
            if bonzo.matches(&msg) {
                println!("bonzo {:?}",msg);
            } else
            if frodo.matches(&msg) {
                println!("frodo {:?}",msg);
                m.disconnect().unwrap();
            }
        }
    });

    m.loop_forever(200).expect("broken loop");
}

examples/self-publish.rs (line 6)

fn main() {
    let m = Mosquitto::new("test");
    
    m.connect("localhost",1883).expect("can't connect");
    m.subscribe("bilbo/#",1).expect("can't subscribe to bonzo");
    
    let mt = m.clone();
    thread::spawn(move || {
        let timeout = time::Duration::from_millis(500);
        for _ in 0..5 {
            mt.publish("bilbo/baggins","hello dolly".as_bytes(), 1, false).unwrap();
            thread::sleep(timeout);
        }
        mt.disconnect().unwrap();
    });
    
    let mut mc = m.callbacks(0);
    mc.on_message(|data,msg| {
        println!("bilbo {:?}",msg);
        *data += 1;
    });
    mc.on_disconnect(|_,rc| println!("disconnect {}",rc));
    
    
    m.loop_until_disconnect(200).expect("broken loop");
    println!("received {} messages",mc.data);
}

pub fn new_session(id: &str, clean_session: bool) -> Mosquitto

create a new mosquitto instance with specified clean session flag. Clients connecting to a broker must have unique names

pub fn callbacks<'a, T>(&'a self, data: T) -> Callbacks<'a, T>

create a Callback object so you can listen to events.

Examples found in repository

examples/verify-publish.rs (line 15)

fn go() -> Result<(),Box<Error>> {
    let m = Mosquitto::new("test");

    m.connect("localhost",1883)?;

    // publish and get a message id
    let our_mid = m.publish("bonzo/dog","hello dolly".as_bytes(), 2, false)?;

    // and wait for confirmation for that message id
    let mut mc = m.callbacks(());
    mc.on_publish(|_,mid| {
        if mid == our_mid {
            m.disconnect().unwrap();
        }
    });

    // wait forever until explicit disconnect
    m.loop_until_disconnect(-1)?;
    Ok(())
}

examples/subscribe-and-listen.rs (line 14)

fn main() {
    let m = Mosquitto::new("test");

    m.will_set("test/will",b"finished!",0,false).expect("can't set will");

    m.connect("localhost",1883).expect("can't connect");
    let bonzo = m.subscribe("bonzo/#",0).expect("can't subscribe to bonzo");
    let frodo = m.subscribe("frodo/#",0).expect("can't subscribe to frodo");

    // not interested in any retained messages!
    let mut mc = m.callbacks(());
    mc.on_message(|_,msg| {
        if ! msg.retained() {
            if bonzo.matches(&msg) {
                println!("bonzo {:?}",msg);
            } else
            if frodo.matches(&msg) {
                println!("frodo {:?}",msg);
                m.disconnect().unwrap();
            }
        }
    });

    m.loop_forever(200).expect("broken loop");
}

examples/self-publish.rs (line 21)

fn main() {
    let m = Mosquitto::new("test");
    
    m.connect("localhost",1883).expect("can't connect");
    m.subscribe("bilbo/#",1).expect("can't subscribe to bonzo");
    
    let mt = m.clone();
    thread::spawn(move || {
        let timeout = time::Duration::from_millis(500);
        for _ in 0..5 {
            mt.publish("bilbo/baggins","hello dolly".as_bytes(), 1, false).unwrap();
            thread::sleep(timeout);
        }
        mt.disconnect().unwrap();
    });
    
    let mut mc = m.callbacks(0);
    mc.on_message(|data,msg| {
        println!("bilbo {:?}",msg);
        *data += 1;
    });
    mc.on_disconnect(|_,rc| println!("disconnect {}",rc));
    
    
    m.loop_until_disconnect(200).expect("broken loop");
    println!("received {} messages",mc.data);
}

pub fn connect(&self, host: &str, port: u32) -> Result<()>

connect to the broker. You can only be fully sure that a connection succeeds after the on_connect callback returns non-zero

Examples found in repository

examples/verify-publish.rs (line 9)

fn go() -> Result<(),Box<Error>> {
    let m = Mosquitto::new("test");

    m.connect("localhost",1883)?;

    // publish and get a message id
    let our_mid = m.publish("bonzo/dog","hello dolly".as_bytes(), 2, false)?;

    // and wait for confirmation for that message id
    let mut mc = m.callbacks(());
    mc.on_publish(|_,mid| {
        if mid == our_mid {
            m.disconnect().unwrap();
        }
    });

    // wait forever until explicit disconnect
    m.loop_until_disconnect(-1)?;
    Ok(())
}

examples/subscribe-and-listen.rs (line 9)

fn main() {
    let m = Mosquitto::new("test");

    m.will_set("test/will",b"finished!",0,false).expect("can't set will");

    m.connect("localhost",1883).expect("can't connect");
    let bonzo = m.subscribe("bonzo/#",0).expect("can't subscribe to bonzo");
    let frodo = m.subscribe("frodo/#",0).expect("can't subscribe to frodo");

    // not interested in any retained messages!
    let mut mc = m.callbacks(());
    mc.on_message(|_,msg| {
        if ! msg.retained() {
            if bonzo.matches(&msg) {
                println!("bonzo {:?}",msg);
            } else
            if frodo.matches(&msg) {
                println!("frodo {:?}",msg);
                m.disconnect().unwrap();
            }
        }
    });

    m.loop_forever(200).expect("broken loop");
}

examples/self-publish.rs (line 8)

fn main() {
    let m = Mosquitto::new("test");
    
    m.connect("localhost",1883).expect("can't connect");
    m.subscribe("bilbo/#",1).expect("can't subscribe to bonzo");
    
    let mt = m.clone();
    thread::spawn(move || {
        let timeout = time::Duration::from_millis(500);
        for _ in 0..5 {
            mt.publish("bilbo/baggins","hello dolly".as_bytes(), 1, false).unwrap();
            thread::sleep(timeout);
        }
        mt.disconnect().unwrap();
    });
    
    let mut mc = m.callbacks(0);
    mc.on_message(|data,msg| {
        println!("bilbo {:?}",msg);
        *data += 1;
    });
    mc.on_disconnect(|_,rc| println!("disconnect {}",rc));
    
    
    m.loop_until_disconnect(200).expect("broken loop");
    println!("received {} messages",mc.data);
}

pub fn connect_wait(&self, host: &str, port: u32, millis: i32) -> Result<()>

connect to the broker, waiting for success.

Examples found in repository

examples/publish_wait.rs (line 18)

fn run() -> Result<(),Box<std::error::Error>> {
    let m = Mosquitto::new("test");

    let t = Instant::now();

    m.connect_wait("localhost",1883,TIMEOUT)?;
    m.publish_wait("/bonzo/dog",b"hello dolly",2,false,TIMEOUT)?;
    m.publish_wait("/bonzo/cat",b"meeeaaaww",2,false,TIMEOUT)?;
    println!("elapsed {:.2} msec",as_millis(t.elapsed()));
    Ok(())
}

examples/self-publish-receive-many.rs (line 8)

fn run() -> mosq::Result<()> {
    let m = Mosquitto::new("test");

    m.connect_wait("localhost",1883,300)?;
    let bilbo = m.subscribe("bilbo/#",1)?;

    let mt = m.clone();
    thread::spawn(move || {
        for i in 0..5 {
            let topic = format!("bilbo/{}",10*(i+1));
            let data = format!("hello #{}",i);
            mt.publish(&topic,data.as_bytes(), 1, false).unwrap();
        }
    });

    let msgs = bilbo.receive_many(300)?;
    for msg in msgs {
        println!("topic {} text '{}'",msg.topic(),msg.text());
    }
    Ok(())
}

pub fn threaded(&self)

call if you wish to use Mosquitto in a multithreaded environment.

pub fn reconnect(&self) -> Result<()>

reconnect to the broker

pub fn reconnect_delay_set( &self, delay: u32, delay_max: u32, exponential_backoff: bool, ) -> Result<()>

pub fn subscribe<'a>(&'a self, sub: &str, qos: u32) -> Result<TopicMatcher<'a>>

subscribe to an MQTT topic, with a desired quality-of-service. The returned value can be used to directly match against received messages, and has a mid field identifying the subscribing request. on_subscribe will be called with this identifier.

Examples found in repository

examples/self-publish-receive-many.rs (line 9)

fn run() -> mosq::Result<()> {
    let m = Mosquitto::new("test");

    m.connect_wait("localhost",1883,300)?;
    let bilbo = m.subscribe("bilbo/#",1)?;

    let mt = m.clone();
    thread::spawn(move || {
        for i in 0..5 {
            let topic = format!("bilbo/{}",10*(i+1));
            let data = format!("hello #{}",i);
            mt.publish(&topic,data.as_bytes(), 1, false).unwrap();
        }
    });

    let msgs = bilbo.receive_many(300)?;
    for msg in msgs {
        println!("topic {} text '{}'",msg.topic(),msg.text());
    }
    Ok(())
}

examples/subscribe-and-listen.rs (line 10)

fn main() {
    let m = Mosquitto::new("test");

    m.will_set("test/will",b"finished!",0,false).expect("can't set will");

    m.connect("localhost",1883).expect("can't connect");
    let bonzo = m.subscribe("bonzo/#",0).expect("can't subscribe to bonzo");
    let frodo = m.subscribe("frodo/#",0).expect("can't subscribe to frodo");

    // not interested in any retained messages!
    let mut mc = m.callbacks(());
    mc.on_message(|_,msg| {
        if ! msg.retained() {
            if bonzo.matches(&msg) {
                println!("bonzo {:?}",msg);
            } else
            if frodo.matches(&msg) {
                println!("frodo {:?}",msg);
                m.disconnect().unwrap();
            }
        }
    });

    m.loop_forever(200).expect("broken loop");
}

examples/self-publish.rs (line 9)

fn main() {
    let m = Mosquitto::new("test");
    
    m.connect("localhost",1883).expect("can't connect");
    m.subscribe("bilbo/#",1).expect("can't subscribe to bonzo");
    
    let mt = m.clone();
    thread::spawn(move || {
        let timeout = time::Duration::from_millis(500);
        for _ in 0..5 {
            mt.publish("bilbo/baggins","hello dolly".as_bytes(), 1, false).unwrap();
            thread::sleep(timeout);
        }
        mt.disconnect().unwrap();
    });
    
    let mut mc = m.callbacks(0);
    mc.on_message(|data,msg| {
        println!("bilbo {:?}",msg);
        *data += 1;
    });
    mc.on_disconnect(|_,rc| println!("disconnect {}",rc));
    
    
    m.loop_until_disconnect(200).expect("broken loop");
    println!("received {} messages",mc.data);
}

pub fn unsubscribe(&self, sub: &str) -> Result<i32>

unsubcribe from an MQTT topic - on_unsubscribe callback will be called.

pub fn publish( &self, topic: &str, payload: &[u8], qos: u32, retain: bool, ) -> Result<i32>

publish an MQTT message to the broker, returning message id. Quality-of-service and whether retained can be specified. To be sure, check the message id passed to the on_publish callback

Examples found in repository

examples/verify-publish.rs (line 12)

fn go() -> Result<(),Box<Error>> {
    let m = Mosquitto::new("test");

    m.connect("localhost",1883)?;

    // publish and get a message id
    let our_mid = m.publish("bonzo/dog","hello dolly".as_bytes(), 2, false)?;

    // and wait for confirmation for that message id
    let mut mc = m.callbacks(());
    mc.on_publish(|_,mid| {
        if mid == our_mid {
            m.disconnect().unwrap();
        }
    });

    // wait forever until explicit disconnect
    m.loop_until_disconnect(-1)?;
    Ok(())
}

examples/self-publish-receive-many.rs (line 16)

fn run() -> mosq::Result<()> {
    let m = Mosquitto::new("test");

    m.connect_wait("localhost",1883,300)?;
    let bilbo = m.subscribe("bilbo/#",1)?;

    let mt = m.clone();
    thread::spawn(move || {
        for i in 0..5 {
            let topic = format!("bilbo/{}",10*(i+1));
            let data = format!("hello #{}",i);
            mt.publish(&topic,data.as_bytes(), 1, false).unwrap();
        }
    });

    let msgs = bilbo.receive_many(300)?;
    for msg in msgs {
        println!("topic {} text '{}'",msg.topic(),msg.text());
    }
    Ok(())
}

examples/self-publish.rs (line 15)

fn main() {
    let m = Mosquitto::new("test");
    
    m.connect("localhost",1883).expect("can't connect");
    m.subscribe("bilbo/#",1).expect("can't subscribe to bonzo");
    
    let mt = m.clone();
    thread::spawn(move || {
        let timeout = time::Duration::from_millis(500);
        for _ in 0..5 {
            mt.publish("bilbo/baggins","hello dolly".as_bytes(), 1, false).unwrap();
            thread::sleep(timeout);
        }
        mt.disconnect().unwrap();
    });
    
    let mut mc = m.callbacks(0);
    mc.on_message(|data,msg| {
        println!("bilbo {:?}",msg);
        *data += 1;
    });
    mc.on_disconnect(|_,rc| println!("disconnect {}",rc));
    
    
    m.loop_until_disconnect(200).expect("broken loop");
    println!("received {} messages",mc.data);
}

pub fn will_set( &self, topic: &str, payload: &[u8], qos: u32, retain: bool, ) -> Result<()>

Examples found in repository

examples/subscribe-and-listen.rs (line 7)

fn main() {
    let m = Mosquitto::new("test");

    m.will_set("test/will",b"finished!",0,false).expect("can't set will");

    m.connect("localhost",1883).expect("can't connect");
    let bonzo = m.subscribe("bonzo/#",0).expect("can't subscribe to bonzo");
    let frodo = m.subscribe("frodo/#",0).expect("can't subscribe to frodo");

    // not interested in any retained messages!
    let mut mc = m.callbacks(());
    mc.on_message(|_,msg| {
        if ! msg.retained() {
            if bonzo.matches(&msg) {
                println!("bonzo {:?}",msg);
            } else
            if frodo.matches(&msg) {
                println!("frodo {:?}",msg);
                m.disconnect().unwrap();
            }
        }
    });

    m.loop_forever(200).expect("broken loop");
}

pub fn will_clear(&self) -> Result<()>

pub fn publish_wait( &self, topic: &str, payload: &[u8], qos: u32, retain: bool, millis: i32, ) -> Result<i32>

publish an MQTT message to the broker, returning message id after waiting for successful publish

Examples found in repository

examples/publish_wait.rs (line 19)

fn run() -> Result<(),Box<std::error::Error>> {
    let m = Mosquitto::new("test");

    let t = Instant::now();

    m.connect_wait("localhost",1883,TIMEOUT)?;
    m.publish_wait("/bonzo/dog",b"hello dolly",2,false,TIMEOUT)?;
    m.publish_wait("/bonzo/cat",b"meeeaaaww",2,false,TIMEOUT)?;
    println!("elapsed {:.2} msec",as_millis(t.elapsed()));
    Ok(())
}

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

explicitly disconnect from the broker.

Examples found in repository

examples/verify-publish.rs (line 18)

fn go() -> Result<(),Box<Error>> {
    let m = Mosquitto::new("test");

    m.connect("localhost",1883)?;

    // publish and get a message id
    let our_mid = m.publish("bonzo/dog","hello dolly".as_bytes(), 2, false)?;

    // and wait for confirmation for that message id
    let mut mc = m.callbacks(());
    mc.on_publish(|_,mid| {
        if mid == our_mid {
            m.disconnect().unwrap();
        }
    });

    // wait forever until explicit disconnect
    m.loop_until_disconnect(-1)?;
    Ok(())
}

examples/subscribe-and-listen.rs (line 22)

fn main() {
    let m = Mosquitto::new("test");

    m.will_set("test/will",b"finished!",0,false).expect("can't set will");

    m.connect("localhost",1883).expect("can't connect");
    let bonzo = m.subscribe("bonzo/#",0).expect("can't subscribe to bonzo");
    let frodo = m.subscribe("frodo/#",0).expect("can't subscribe to frodo");

    // not interested in any retained messages!
    let mut mc = m.callbacks(());
    mc.on_message(|_,msg| {
        if ! msg.retained() {
            if bonzo.matches(&msg) {
                println!("bonzo {:?}",msg);
            } else
            if frodo.matches(&msg) {
                println!("frodo {:?}",msg);
                m.disconnect().unwrap();
            }
        }
    });

    m.loop_forever(200).expect("broken loop");
}

examples/self-publish.rs (line 18)

fn main() {
    let m = Mosquitto::new("test");
    
    m.connect("localhost",1883).expect("can't connect");
    m.subscribe("bilbo/#",1).expect("can't subscribe to bonzo");
    
    let mt = m.clone();
    thread::spawn(move || {
        let timeout = time::Duration::from_millis(500);
        for _ in 0..5 {
            mt.publish("bilbo/baggins","hello dolly".as_bytes(), 1, false).unwrap();
            thread::sleep(timeout);
        }
        mt.disconnect().unwrap();
    });
    
    let mut mc = m.callbacks(0);
    mc.on_message(|data,msg| {
        println!("bilbo {:?}",msg);
        *data += 1;
    });
    mc.on_disconnect(|_,rc| println!("disconnect {}",rc));
    
    
    m.loop_until_disconnect(200).expect("broken loop");
    println!("received {} messages",mc.data);
}

pub fn do_loop(&self, timeout: i32) -> Result<()>

process network events for at most timeout milliseconds. -1 will mean the default, 1000ms.

pub fn loop_forever(&self, timeout: i32) -> Result<()>

process network events. This will handle intermittent disconnects for you, but will return after an explicit disconnect() call

Examples found in repository

examples/subscribe-and-listen.rs (line 27)

fn main() {
    let m = Mosquitto::new("test");

    m.will_set("test/will",b"finished!",0,false).expect("can't set will");

    m.connect("localhost",1883).expect("can't connect");
    let bonzo = m.subscribe("bonzo/#",0).expect("can't subscribe to bonzo");
    let frodo = m.subscribe("frodo/#",0).expect("can't subscribe to frodo");

    // not interested in any retained messages!
    let mut mc = m.callbacks(());
    mc.on_message(|_,msg| {
        if ! msg.retained() {
            if bonzo.matches(&msg) {
                println!("bonzo {:?}",msg);
            } else
            if frodo.matches(&msg) {
                println!("frodo {:?}",msg);
                m.disconnect().unwrap();
            }
        }
    });

    m.loop_forever(200).expect("broken loop");
}

pub fn loop_until_disconnect(&self, timeout: i32) -> Result<()>

loop forever, but do not regard an explicit disconnect as an error.

Examples found in repository

examples/verify-publish.rs (line 23)

fn go() -> Result<(),Box<Error>> {
    let m = Mosquitto::new("test");

    m.connect("localhost",1883)?;

    // publish and get a message id
    let our_mid = m.publish("bonzo/dog","hello dolly".as_bytes(), 2, false)?;

    // and wait for confirmation for that message id
    let mut mc = m.callbacks(());
    mc.on_publish(|_,mid| {
        if mid == our_mid {
            m.disconnect().unwrap();
        }
    });

    // wait forever until explicit disconnect
    m.loop_until_disconnect(-1)?;
    Ok(())
}

examples/self-publish.rs (line 29)

fn main() {
    let m = Mosquitto::new("test");
    
    m.connect("localhost",1883).expect("can't connect");
    m.subscribe("bilbo/#",1).expect("can't subscribe to bonzo");
    
    let mt = m.clone();
    thread::spawn(move || {
        let timeout = time::Duration::from_millis(500);
        for _ in 0..5 {
            mt.publish("bilbo/baggins","hello dolly".as_bytes(), 1, false).unwrap();
            thread::sleep(timeout);
        }
        mt.disconnect().unwrap();
    });
    
    let mut mc = m.callbacks(0);
    mc.on_message(|data,msg| {
        println!("bilbo {:?}",msg);
        *data += 1;
    });
    mc.on_disconnect(|_,rc| println!("disconnect {}",rc));
    
    
    m.loop_until_disconnect(200).expect("broken loop");
    println!("received {} messages",mc.data);
}

pub fn tls_set<P1, P2, P3>( &self, cafile: P1, certfile: P2, keyfile: P3, passphrase: Option<&str>, ) -> Result<()>

where P1: AsRef<Path>, P2: AsRef<Path>, P3: AsRef<Path>,

Set TLS parameters cafile is a file containing the PEM encoded trusted CA certificate certfile is a file containing the PEM encoded certificate file for this client. keyfile is a file containing the PEM encoded private key for this client. password if the private key is encrypted

pub fn tls_psk_set( &self, psk: &str, identity: &str, ciphers: Option<&str>, ) -> Result<()>

Set TLS PSK parameters psk is the pre-shared-key in hex format with no leading “0x” identity is the identity of this client. May be used as the username ciphers is an optional string describing the PSK ciphers available for use

Trait Implementations

impl Clone for Mosquitto

fn clone(&self) -> Mosquitto

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Drop for Mosquitto

fn drop(&mut self)

Executes the destructor for this type.

impl Send for Mosquitto

impl Sync for Mosquitto