Pure rust mqtt cilent

This is a follow up on #142 without any change on the reconnection behaviour.

Check channel state before sending notifications and distibute connection state changes. The notification Receiver is wrapped into a struct that contains a oneshot Receiver that allows the client to check whether the notification struct is dropped or not. Upon disconnections the corresponding Error is sent.

I'm trying the new tokio2 branch with a simple subscriber.

I'm trying to subscribe to two topics:

let (mut client, receiver) = MqttClient::start(client_options);
let topics = vec![
    ("+/devices/+/activations", QoS::AtMostOnce),
    ("+/devices/+/up", QoS::AtMostOnce),
];
client.subscribe(topics).expect("Subcription failure");

thread::spawn(move || {
    println!("--> Listening!");
    for packet in receiver {
        println!("received {:?}", packet);
    }
}).join().unwrap();

When I start the client:

INFO 2018-02-03T01:04:16Z: rumqtt::client::connection: mqtt connection successful
DEBUG 2018-02-03T01:04:16Z: rumqtt::client::connection: Sending packet. Subscribe(Subscribe { pid: PacketIdentifier(1), topics: [SubscribeTopic { topic_path: "+/devices/+/activations", qos: AtMostOnce }, SubscribeTopic { topic_path: "+/devices/+/up", qos: AtMostOnce }] })
DEBUG 2018-02-03T01:04:16Z: rumqtt::client::connection: Received packet. "Suback(Suback { pid: PacketIdentifier(1), return_codes: [Success(AtMostOnce), Success(AtMostOnce)] })"
DEBUG 2018-02-03T01:04:16Z: rumqtt::client::connection: Sending packet. Suback(Suback { pid: PacketIdentifier(1), return_codes: [Success(AtMostOnce), Success(AtMostOnce)] })
ERROR 2018-02-03T01:04:16Z: rumqtt::client::connection: Reactor stopped. v = ()

Why is the reactor stopped?

And why is the thread not terminated (= the packet receiver closed) if the reactor is stopped?

(Sorry if this is a stupid question, I'm not too familiar with MQTT 🙂)

I'm experiencing a disconnect from the server after a bit more than 1 minute.

Here's a wireshark trace:

(If required I can also send more details or a pcap file)

The ping messages are sent at t = 25, 35 (+10), 50 (+15), 60 (+10), 75 (+15). First of all, it seems a bit strange that the intervals are not consistent. Then, after sending the ping request at t=75, the connection is immediately closed by the server.

In the ping message itself I cannot find anything suspicious. There's no counter that could go out of sync or something like that, is there?

Maybe related to https://github.com/AtherEnergy/rumqtt/issues/76.

This code often disconnects after 15 secs but not every time.

rumqtt 3b5571a40275be65188a678c39d70b75e67a77d7 rustc 1.36.0-nightly (7c71bc320 2019-04-30)

use env_logger::Env;
use rumqtt::{MqttClient, MqttOptions, QoS};

use std::{thread, time::Duration};

fn main() {
    let env = Env::default().filter_or(
        "RUST_LOG",
        "test_rumqtt=debug,rumqtt=info,tokio_reactor=info,debug",
    );
    env_logger::init_from_env(env);

    let mqtt_options = MqttOptions::new("test-id-1", "127.0.0.1", 1883).set_keep_alive(10);
    let (mut mqtt_client, notifications) = MqttClient::start(mqtt_options).unwrap();

    thread::spawn(move || {
        mqtt_client
            .publish("hello/world", QoS::AtLeastOnce, true, "test")
            .unwrap();

        thread::sleep(Duration::from_secs(300));
    });

    for notification in notifications {
        println!("notification: {:?}", notification)
    }
}

[2019-06-13T18:56:35Z DEBUG tokio_reactor] adding I/O source: 0
[2019-06-13T18:56:35Z DEBUG tokio_reactor::registration] scheduling Write for: 0
[2019-06-13T18:56:35Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:35Z INFO  rumqtt::client::connection] Mqtt connect response = Some(Connack(Connack { session_present: false, code: Accepted }))
[2019-06-13T18:56:35Z INFO  rumqtt::client::connection] Mqtt connection successful!!
[2019-06-13T18:56:35Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:35Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:35Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:35Z DEBUG rumqtt::client::connection] Outgoing request = "topic = hello/world, qos = AtLeastOnce, pkid = Some(PacketIdentifier(1)), payload size = 4 bytes"
[2019-06-13T18:56:35Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:35Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:35Z DEBUG rumqtt::client::connection] Incoming packet = "Puback(PacketIdentifier(1))"
[2019-06-13T18:56:35Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:35Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:35Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:43Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:45Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:45Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:45Z DEBUG rumqtt::client::mqttstate] Ping = None. keep alive = 10,
            last incoming packet before 9 secs,
            last outgoing packet before 9 secs
[2019-06-13T18:56:45Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:45Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:45Z DEBUG tokio_reactor::registration] scheduling Read for: 0
[2019-06-13T18:56:50Z DEBUG rumqtt::client::mqttstate] Ping = Ping. keep alive = 10,
            last incoming packet before 15 secs,
            last outgoing packet before 5 secs
[2019-06-13T18:56:50Z DEBUG tokio_reactor] dropping I/O source: 0

$ docker run -it --rm -p 1883:1883 -p 9001:9001 eclipse-mosquitto
1560452193: mosquitto version 1.6.2 starting
1560452193: Config loaded from /mosquitto/config/mosquitto.conf.
1560452193: Opening ipv4 listen socket on port 1883.
1560452193: Opening ipv6 listen socket on port 1883.
1560452195: New connection from 172.17.0.1 on port 1883.
1560452195: New client connected from 172.17.0.1 as test-id-1 (p2, c1, k10).
1560452210: Client test-id-1 has exceeded timeout, disconnecting.

Currently it's rather hard to find out which version on creates.io corresponds to which version of the source code. This makes debugging problems with released versions harder.

Hello, thank you for the great library!

I am looking for Rust MQTT library for my application. I want it to be able to handle incoming messages asynchronously because there will be DB interaction and other blocking things.

I added thread::sleep within on_message callback. Message handling had been queued: one message was handling (i.e. waiting on sleep) and new incoming message was queued and processed only after awaking. Maybe that is totally my fault and I made the main and the only one thread to sleep.

What is the correct way to asynchronous message handling? Can I create some kind of a thread poll within on_message and just push incoming messages to that pool? Or it will affect dealing with QoS? As I can see from the code we firstly send PUBACK and then call user's callback. Does it mean that it doesn't matter whether I would handle incoming message in the same or in separate thread because PUBACK was already sent?

Also, I have found tokio2 branch. Maybe it better suits for async handling?

Thank you.

Hi.

the way rumqtt handles kind of overload situations seems not optimal to me. The current behaviour just blocks for a configured amount of time when the request channel is full. If the application sends any command during this delay, rumqtt will block again when polling next time. This leeds to a rate of 1/configured delay. If the clients keeps sending you will stay at this rate.

I changed this implementation and removed the fixed delay with a stream interception when the publication queue is at a configured level (MqttOptions::in_flight). The essential part is here:

    // Apply outgoing queue limit (in flights) by answering stream poll with not ready if queue is full
    // by returning NotReady.
    fn limit_in_flight_request_stream(&self, requests: impl RequestStream) -> impl RequestStream {
        let mqtt_state = self.mqtt_state.clone();
        let in_flight = self.mqttoptions.in_flight();
        let mut stream = requests.peekable();
        poll_fn(move || -> Poll<Option<Request>, NetworkError> {
            if mqtt_state.borrow().publish_queue_len() >= in_flight {
                match stream.peek() {
                    Err(_) => stream.poll(),
                    _ => Ok(Async::NotReady)
                }
            } else {
                stream.poll()
            }
        })
    }

This affects any command on the request queue. I'm not sure if there is a definition for the in flight window and if it should contain e.g Subscribes or others. I personally think it doesn't matter if they're included.

Next the stream throttling implementation is also a little bit misleading. I replaced the sleep time that happen when a configured rate is reached with a simple StreamExt::throttle that is a prefect match.

This patch changes the client API.

What do you think? I consider this as work in progress and appreciate any feedback!

cheers!

@jamesmunns @rschifflin @ivanovaleksey @nanomad @creativcoder @vorot93

I'm kinda happy with the status of async branch and planning to release it in the next 10 days. However qos2 and retained publishes aren't supported yet. I would like to know if anyone of you are using these features and also would like some help in reviewing and testing. Please let me know if you can spare some time for this

Added ALPN TLS extension support.

• Extend ConnectionMethod with alpn field.

notice

Protocol candidates are represented as Vec<String>, but latest rustls (1.15.0) uses Vec<Vec<u8>>.

Trying to connect to an invalid web-address such as braker.hivemq.com are never resolved,

#[macro_use] extern crate log;
use rumqtt::*;
use rumqtt::mqttoptions::*;

fn main() {
    simple_logger::init_with_level( log::Level::Info )
        .expect("Could not initiate logger");

    let mqtt_options = MqttOptions::new( "Testclient".to_string(), "braker.hivemq.com".to_string(), 1883)
        .set_reconnect_opts(ReconnectOptions::AfterFirstSuccess(0));

    let mqtt_conn = MqttClient::start(mqtt_options);

    loop {
    }
}

yields

but the error is never detected and the code continues anyway

The following code fails because the current implementation of setters require implementation of Copy trait.

let mut opts = rumqtt::MqttOptions::new(client_id, host, port.as_u16());
if let Some(value) = config.clean_session {
    opts.set_clean_session(value);
};

Ok(opts)

error[E0382]: use of moved value: `opts`
   --> /Users/manifest/projects/netology-group/svc-agent-rs/src/mqtt.rs:107:12
    |
103 |             opts.set_clean_session(value);
    |             ---- value moved here
...
107 |         Ok(opts)
    |            ^^^^ value used here after move
    |
    = note: move occurs because `opts` has type `rumqtt::mqttoptions::MqttOptions`, which does not implement the `Copy` trait

error: aborting due to previous error

For more information about this error, try `rustc --explain E0382`.
error: Could not compile `svc-agent`.

To learn more, run the command again with --verbose.

Hoo boy. There's lots of changes here. I think things should be working, and I'll be testing on the corpus of code we've built on rumqtt at Vivint, but I'm definitely no guru with async/await or rumqtt's codebase.

Let me know if there are any questions or concerns. It's my job right now to get this upstreamed if possible, so I'll do my best to resolve them quickly!

This PR has a soft dependency on https://github.com/tekjar/mqtt311/pull/2, which is intended to consolidate the dependency tree.

Hello,

The tls example doesn't work for me, I tried the following code:

use rumqtt::{MqttClient, MqttOptions, QoS};
use std::{thread, time::Duration};

fn main() {
    pretty_env_logger::init();

    let client_id = "tls-test".to_owned();
    let ca = include_bytes!("path/to/cert.pem").to_vec();
    let client_cert = include_bytes!("path/to/client-cert.pem").to_vec();
    let client_key = include_bytes!("path/to/client-key.pem").to_vec();

    let mqtt_options = MqttOptions::new(client_id, "xxxxxxx", 8883)
        .set_ca(ca)
        .set_client_auth(client_cert, client_key)
        .set_keep_alive(10);

    MqttClient::start(mqtt_options);
    // let topic = "hello/world";

    // thread::spawn(move || {
    //     for i in 0..100 {
    //         let payload = format!("publish {}", i);
    //         thread::sleep(Duration::from_secs(1));
    //         if let Err(err) = mqtt_client.publish(topic.clone(), QoS::AtLeastOnce, false, payload) {
    //             println!("{:?}", err);
    //         }
    //     }
    // });

    // for notification in notifications {
    //     println!("{:?}", notification)
    // }
}

I got the following message while executing RUST_BACKTRACE=1 cargo run --example tls, is there any workaround or am I doing something wrong?

thread '<unnamed>' panicked at 'called `Result::unwrap()` on an `Err` value: ()', src/libcore/result.rs:1165:5
stack backtrace:
   0: backtrace::backtrace::libunwind::trace
             at /cargo/registry/src/github.com-1ecc6299db9ec823/backtrace-0.3.40/src/backtrace/libunwind.rs:88
   1: backtrace::backtrace::trace_unsynchronized
             at /cargo/registry/src/github.com-1ecc6299db9ec823/backtrace-0.3.40/src/backtrace/mod.rs:66
   2: std::sys_common::backtrace::_print_fmt
             at src/libstd/sys_common/backtrace.rs:77
   3: <std::sys_common::backtrace::_print::DisplayBacktrace as core::fmt::Display>::fmt
             at src/libstd/sys_common/backtrace.rs:61
   4: core::fmt::write
             at src/libcore/fmt/mod.rs:1030
   5: std::io::Write::write_fmt
             at src/libstd/io/mod.rs:1412
   6: std::sys_common::backtrace::_print
             at src/libstd/sys_common/backtrace.rs:65
   7: std::sys_common::backtrace::print
             at src/libstd/sys_common/backtrace.rs:50
   8: std::panicking::default_hook::{{closure}}
             at src/libstd/panicking.rs:188
   9: std::panicking::default_hook
             at src/libstd/panicking.rs:205
  10: std::panicking::rust_panic_with_hook
             at src/libstd/panicking.rs:464
  11: std::panicking::continue_panic_fmt
             at src/libstd/panicking.rs:373
  12: rust_begin_unwind
             at src/libstd/panicking.rs:302
  13: core::panicking::panic_fmt
             at src/libcore/panicking.rs:141
  14: core::result::unwrap_failed
             at src/libcore/result.rs:1165
  15: core::result::Result<T,E>::unwrap
             at /rustc/4f03f4a989d1c8346c19dfb417a77c09b34408b8/src/libcore/result.rs:933
  16: rumqtt::client::network::stream::NetworkStreamBuilder::connect
             at src/client/network.rs:208
  17: rumqtt::client::connection::Connection::tcp_connect_future
             at src/client/connection.rs:315
  18: rumqtt::client::connection::Connection::mqtt_connect
             at src/client/connection.rs:321
  19: rumqtt::client::connection::Connection::mqtt_eventloop
             at src/client/connection.rs:91
  20: rumqtt::client::connection::Connection::run::{{closure}}
             at src/client/connection.rs:60
note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.

Create an unbounded crossbeam_channel by default. For bounded crossbeam_channel use the 'set_notification_channel_capacity' to a value of choice.

This should fix the issue #166

Hi All. I see that there are a bunch of open issues and pull requests. I'm not working at Ather anymore and it might take some time for these to be addressed. Ather is considering transfer of ownership but it'll take time as internal opensource policies are still being framed.

Meanwhile, I've started my work on the broker and the ecosystem. Head here in case you have any queries

It's certainly not perfect, but it changes a few of the panics into Errs. Unfortunately there's not much in the way of information that can be put into the errors since rusttls has an error type of ().

Closes https://github.com/AtherEnergy/rumqtt/issues/176