Are your examples outdated?

[Tockra]

Hi Guys,

I tried to compile 2 of your examples: https://github.com/eclipse/paho.mqtt.rust/blob/master/examples/ssl_publish.rs https://github.com/eclipse/paho.mqtt.rust/blob/master/examples/async_subscribe.rs

Here the compiler output: ssl-example:

Tims-MacBook-Pro:src timtannert$ cargo build
   Compiling piper v0.1.3
   Compiling futures-executor v0.3.5
   Compiling paho-mqtt-sys v0.3.0
   Compiling futures v0.3.5
   Compiling smol v0.1.11
   Compiling async-std v1.6.0
   Compiling paho-mqtt v0.7.1
   Compiling example_mqtt v0.1.0 (/Users/timtannert/example_mqtt)
error[E0433]: failed to resolve: use of undeclared type or module `env_logger`
  --> src/main.rs:51:6
   |
51 |      env_logger::init();
   |      ^^^^^^^^^^ use of undeclared type or module `env_logger`

error[E0412]: cannot find type `Result` in crate `mqtt`
  --> src/main.rs:49:21
   |
49 |  fn main() -> mqtt::Result<()> {
   |                     ^^^^^^ not found in `mqtt`
   |
help: possible candidates are found in other modules, you can import them into scope
   |
43 |  use async_std::io::Result;
   |
43 |  use core::fmt::Result;
   |
43 |  use core::prelude::v1::Result;
   |
43 |  use core::result::Result;
   |
     and 6 other candidates

error[E0412]: cannot find type `Error` in crate `mqtt`
   --> src/main.rs:116:25
    |
116 |          Ok::<(), mqtt::Error>(())
    |                         ^^^^^ not found in `mqtt`
    |
help: possible candidates are found in other modules, you can import them into scope
    |
43  |  use async_std::io::Error;
    |
43  |  use core::fmt::Error;
    |
43  |  use futures::io::Error;
    |
43  |  use std::error::Error;
    |
      and 2 other candidates

error[E0599]: no method named `create_client` found for struct `paho_mqtt::create_options::CreateOptionsBuilder` in the current scope
  --> src/main.rs:92:27
   |
92 |                          .create_client()?;
   |                           ^^^^^^^^^^^^^ method not found in `paho_mqtt::create_options::CreateOptionsBuilder`

error[E0308]: mismatched types
  --> src/main.rs:95:27
   |
95 |              .trust_store(trust_store)?
   |                           ^^^^^^^^^^^ expected `&str`, found struct `std::path::PathBuf`

error[E0277]: the `?` operator can only be applied to values that implement `std::ops::Try`
  --> src/main.rs:94:25
   |
94 |            let ssl_opts = mqtt::SslOptionsBuilder::new()
   |  _________________________^
95 | |              .trust_store(trust_store)?
   | |_______________________________________^ the `?` operator cannot be applied to type `&mut paho_mqtt::ssl_options::SslOptionsBuilder`
   |
   = help: the trait `std::ops::Try` is not implemented for `&mut paho_mqtt::ssl_options::SslOptionsBuilder`
   = note: required by `std::ops::Try::into_result`

error: aborting due to 6 previous errors

Some errors have detailed explanations: E0277, E0308, E0412, E0433, E0599.
For more information about an error, try `rustc --explain E0277`.
error: could not compile `example_mqtt`.

async_output

Compiling example_mqtt v0.1.0 (/Users/timtannert/example_mqtt)
error[E0433]: failed to resolve: use of undeclared type or module `env_logger`
  --> src/main.rs:54:5
   |
54 |     env_logger::init();
   |     ^^^^^^^^^^ use of undeclared type or module `env_logger`

error[E0425]: cannot find value `QOS_1` in crate `mqtt`
  --> src/main.rs:79:44
   |
79 | ...                   mqtt::QOS_1);
   |                             ^^^^^ not found in `mqtt`

error[E0412]: cannot find type `Error` in crate `mqtt`
   --> src/main.rs:114:24
    |
114 |         Ok::<(), mqtt::Error>(())
    |                        ^^^^^ not found in `mqtt`
    |
help: possible candidates are found in other modules, you can import them into scope
    |
35  |  use async_std::io::Error;
    |
35  |  use core::fmt::Error;
    |
35  |  use futures::io::Error;
    |
35  |  use std::error::Error;
    |
      and 2 other candidates

error[E0277]: the trait bound `paho_mqtt::token::Token: core::future::future::Future` is not satisfied
  --> src/main.rs:90:9
   |
90 |         cli.connect(conn_opts).await?;
   |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ the trait `core::future::future::Future` is not implemented for `paho_mqtt::token::Token`
   |
   = note: required by `core::future::future::Future::poll`

error[E0277]: the trait bound `paho_mqtt::token::Token: core::future::future::Future` is not satisfied
  --> src/main.rs:93:9
   |
93 |         cli.subscribe_many(TOPICS, QOS).await?;
   |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ the trait `core::future::future::Future` is not implemented for `paho_mqtt::token::Token`
   |
   = note: required by `core::future::future::Future::poll`

error[E0599]: no method named `next` found for struct `futures::sync::mpsc::Receiver<std::option::Option<paho_mqtt::message::Message>>` in the current scope
   --> src/main.rs:98:40
    |
98  |         while let Some(msg_opt) = strm.next().await {
    |                                        ^^^^ method not found in `futures::sync::mpsc::Receiver<std::option::Option<paho_mqtt::message::Message>>`
    | 
   ::: /Users/timtannert/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.29/src/sync/mpsc/mod.rs:123:1
    |
123 | pub struct Receiver<T> {
    | ----------------------
    | |
    | doesn't satisfy `_: futures_core::stream::Stream`
    | doesn't satisfy `_: futures_util::stream::stream::StreamExt`
    |
    = note: the method `next` exists but the following trait bounds were not satisfied:
            `futures::sync::mpsc::Receiver<std::option::Option<paho_mqtt::message::Message>>: futures_core::stream::Stream`
            which is required by `futures::sync::mpsc::Receiver<std::option::Option<paho_mqtt::message::Message>>: futures_util::stream::stream::StreamExt`

error[E0277]: the trait bound `paho_mqtt::token::Token: core::future::future::Future` is not satisfied
   --> src/main.rs:105:38
    |
105 |                 while let Err(err) = cli.reconnect().await {
    |                                      ^^^^^^^^^^^^^^^^^^^^^ the trait `core::future::future::Future` is not implemented for `paho_mqtt::token::Token`
    |
    = note: required by `core::future::future::Future::poll`

warning: unused import: `stream::StreamExt`
  --> src/main.rs:42:5
   |
42 |     stream::StreamExt,
   |     ^^^^^^^^^^^^^^^^^
   |
   = note: `#[warn(unused_imports)]` on by default

error: aborting due to 7 previous errors; 1 warning emitted

Some errors have detailed explanations: E0277, E0412, E0425, E0433, E0599.
For more information about an error, try `rustc --explain E0277`.
error: could not compile `example_mqtt`.

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

I'm just raging. I just wanted to write fast a mqtt consumer and I'm trying to code it for 4 hours... Nothing works, everytime I get a error.

Please help me... T

[Tockra]

I see they work just for the newest git version...

[fpagliughi]

The Git sources are changing quickly at the moment in preparation for a new release. There will be several breaking changes in the v0.8 release. The samples may only work with the code from the specific branch. If they don't work within a branch, please let me know, so I can fix them before the release.

[Tockra]

So I found a real bug now. https://github.com/eclipse/paho.mqtt.rust/blob/master/src/async_client.rs The reconnect part isn't correct. You need to subscribe the topics after reconnection again. Otherwise you get a connection but you don't get the messages anymore.

But for your information, the reconnect is realy slow. When I disable my wifi and enable it again. I need something between 20-60 seconds to reconnect to the mqtt broker.

[fpagliughi]

No that's not a bug. Reconnect just makes a connection. If you're using a persistent (non-clean) session, the server will remember, not only your subscriptions, but also any QoS>0 messages that would have been sent to you while you were disconnected. This is a very important feature of MQTT. It might be worth reading through the MQTT spec or some tutorials. The protocol has a lot of useful features around connection and session management.

As for the slow reconnect, this is likely something that can be adjusted in your application to account for your particular network setup.

Is it taking a long time for the client to notice the connection is down? This often happens with wireless setups. Then maybe decrease your keep alive interval. That would help.

Are you using an example that has a retry backoff? When you have 100,000 devices in the field, you don't want them all calling back in the immediate second the connection to the server is restored, or wasting their own CPU in a fast retry cycle when the server is down. So some examples demonstrate that.

The best thing may be to check the logs or spy on the network traffic. That may give you some hints.

[Tockra]

No that's not a bug.

Okay my fault. I wanted to link the example and not the async client: https://github.com/eclipse/paho.mqtt.rust/blob/master/examples/async_subscribe.rs Your example provides a subscriber which reconnects after network issue but doesn't subscribe the stream again? I think your example is missing a

        println!("Subscribing to topics: {:?}", TOPICS);
        cli.subscribe_many(TOPICS, QOS).await?;

after line 109.

Is it taking a long time for the client to notice the connection is down? This often happens with wireless setups. Then maybe decrease your keep alive interval. That would help.

No it recognize the failing connection within ~5 seconds. But the reconnect needs some time. I'm using your example: https://github.com/eclipse/paho.mqtt.rust/blob/master/examples/async_subscribe.rs I just changed it to use tokio. Not a own application.

[fpagliughi]

I think your example is missing a

        println!("Subscribing to topics: {:?}", TOPICS);
        cli.subscribe_many(TOPICS, QOS).await?;

after line 109.

Did you try it?

[Tockra]

I think your example is missing a

        println!("Subscribing to topics: {:?}", TOPICS);
        cli.subscribe_many(TOPICS, QOS).await?;

after line 109.

Did you try it?

Yes. Before I wrote my post. Without these lines it reconnects but doesn't get new messages. After adding this lines it worked.

[fpagliughi]

Something must be wrong. The example connects with a persistent session:

    let conn_opts = mqtt::ConnectOptionsBuilder::new()
        .keep_alive_interval(Duration::from_secs(20))
        .mqtt_version(mqtt::MQTT_VERSION_3_1_1)
        .clean_session(false)                   <- Request persistent session
        .will_message(lwt)
        .finalize();

The broker should not only remember the subscriptions, but also queue up any messages with QoS>0 on those topics while you're client is disconnected.

So in other words, you should be able to disconnect your board (pull the network cable, turn off the WiFi, or whatever), then send a few more messages to the broker from another client that is still connected using QoS=1 or 2. When you turn the WiFi back on, your board should automatically reconnect and then receive the messages that were sent while your board was disconnected.

[fpagliughi]

If you connect with a clean (non-persistent) session, then, yes, you need to re-subscribe to your topics to receive messages again.

[Tockra]

In my test after reconnecting I didn't get the message of the topic which was subscribed. If you wanted to do that, then everything is right. But I think that isn't in the meaning of the example.

[GopherJ]

@fpagliughi Hi, I'm wondering when will be the next stable release? I just had a new project and I would definitely want to try new version of paho which has future0.3 support, if there is musl that's even better.

[fpagliughi]

The new release was delayed because of my work schedule. But it was getting pretty close. You can definitely start testing the master branch in a new project. It has futures 0.3, but the musl support is not quite there yet,

[fpagliughi]

In v0.8