Repository: bytebeamio/rumqtt (language: rust)

Issue:
Can the sync API be used without a tokio reactor?
I'd like to use `rumqttc` in a strictly synchronous context, but when I do so, my app panics with:

> thread 'main' panicked at 'there is no reactor running, must be called from the context of a Tokio 1.x runtime'

Is there a way to run the client without any async runtime/scaffolding? If not, do you have plans to allow this?


Discussion / hints:
@aeshirey can you check with the latest main branch?  Because this issue was fixed [here.](https://github.com/bytebeamio/rumqtt/pull/492) 

You can do it like:
```toml
rumqttc = { default-features = false, git = "https://github.com/bytebeamio/rumqtt" }
```

If the issue still persist, please provide a minimal setup to reproduce it!
I just tested and found the panic still occurs. Here's my minimum, reproducible example:

Cargo.toml:
```toml
[package]
name = "my-example"
version = "0.1.0"
edition = "2021"

[dependencies]
rumqttc = { default-features = false, git = "https://github.com/bytebeamio/rumqtt" }
```

main.rs:
```rust
use rumqttc::mqttbytes::v4::Packet;
use rumqttc::{Client, Event, MqttOptions, QoS};

fn main() {
    let mqttoptions = MqttOptions::new("my-client-id", "localhost", 1883);
    let (mut client, mut connection) = Client::new(mqttoptions, 10);
    client.subscribe("sub-topic", QoS::ExactlyOnce).ok();

    loop {
        let event = connection.try_recv().unwrap().unwrap();

        if let Event::Incoming(Packet::Publish(p)) = event {
            let s = String::from_utf8_lossy(p.payload.as_ref()).to_string();
            println!("{s}");
        }
    }
}
```

Running it:

```bash
$ ~/cargo-target/release/my-example
thread 'main' panicked at 'there is no reactor running, must be called from the context of a Tokio 1.x runtime', /home/adam/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.2/src/runtime/context.rs:56:26
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
```

The stacktrace includes the following:

```
   9: futures_util::future::future::FutureExt::now_or_never
  10: rumqttc::client::Connection::try_recv
```

So when I look at `Connection::try_recv`, it [calls `f.now_or_never()`](https://github.com/bytebeamio/rumqtt/blob/38786428bb1dae931073a118ef1acc635aeb04e5/rumqttc/src/client.rs#L417), where `f` is the result of [`EventLoop::poll`](https://github.com/bytebeamio/rumqtt/blob/38786428bb1dae931073a118ef1acc635aeb04e5/rumqttc/src/eventloop.rs#L112), an async function. I'm not _quite_ sure what the solution is here, but I see that in the previous PR, you had updated the very next function, [`recv_timeout` to include `self.runtime.enter()`](https://github.com/bytebeamio/rumqtt/blob/38786428bb1dae931073a118ef1acc635aeb04e5/rumqttc/src/client.rs#L432), which `try_recv` doesn't have. Possibly [`recv`](https://github.com/bytebeamio/rumqtt/blob/38786428bb1dae931073a118ef1acc635aeb04e5/rumqttc/src/client.rs#L404) should also include this call?

I don't do much with async, but is this the optimal way of supporting fully synchronous APIs? 

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 99d840675..47fff189d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,10 +1,16 @@
 ### Unreleased
-
+---
+rumqttc
+-------
+- Fixed panicking in `recv_timeout` and `try_recv` by entering tokio runtime context (#492, #497)
 - Removed unused dependencies and updated version of some of used libraries to fix dependabots warning (#475)
-- (rumqttd) Added properties field to `Unsubscribe`, `UnsubAck`, and `Disconnect` packets so its consistent with other packets. (#480)
-- (rumqttd) Changed default segment size in demo config to 100MB (#484)
-- (rumqttc) Fixed panicking of `timeout` in `recv_timeout` by entering tokio runtime context (#492)
-- (rumqttd) Allow subscription on topic's starting with `test` (#494)
+
+rumqttd
+-------
+- Added properties field to `Unsubscribe`, `UnsubAck`, and `Disconnect` packets so its consistent with other packets. (#480)
+- Changed default segment size in demo config to 100MB (#484)
+- Allow subscription on topic's starting with `test` (#494)
+-----------
 
 ### R17
 ---
diff --git a/rumqttc/examples/syncrecv.rs b/rumqttc/examples/syncrecv.rs
new file mode 100644
index 000000000..f64204d44
--- /dev/null
+++ b/rumqttc/examples/syncrecv.rs
@@ -0,0 +1,41 @@
+use rumqttc::{self, Client, LastWill, MqttOptions, QoS};
+use std::thread;
+use std::time::Duration;
+
+fn main() {
+    pretty_env_logger::init();
+
+    let mut mqttoptions = MqttOptions::new("test-1", "localhost", 1883);
+    let will = LastWill::new("hello/world", "good bye", QoS::AtMostOnce, false);
+    mqttoptions
+        .set_keep_alive(Duration::from_secs(5))
+        .set_last_will(will);
+
+    let (client, mut connection) = Client::new(mqttoptions, 10);
+    thread::spawn(move || publish(client));
+
+    if let Ok(notification) = connection.recv() {
+        println!("Notification = {:?}", notification);
+    }
+
+    if let Ok(notification) = connection.try_recv() {
+        println!("Notification = {:?}", notification);
+    }
+
+    if let Ok(notification) = connection.recv_timeout(Duration::from_secs(10)) {
+        println!("Notification = {:?}", notification);
+    }
+}
+
+fn publish(mut client: Client) {
+    client.subscribe("hello/+/world", QoS::AtMostOnce).unwrap();
+    for i in 0..3 {
+        let payload = vec![1; i];
+        let topic = format!("hello/{}/world", i);
+        let qos = QoS::AtLeastOnce;
+
+        client.publish(topic, qos, true, payload).unwrap();
+    }
+
+    thread::sleep(Duration::from_secs(1));
+}
diff --git a/rumqttc/examples/syncrecv_v5.rs b/rumqttc/examples/syncrecv_v5.rs
new file mode 100644
index 000000000..acd6a5f9f
--- /dev/null
+++ b/rumqttc/examples/syncrecv_v5.rs
@@ -0,0 +1,42 @@
+use rumqttc::v5::mqttbytes::{LastWill, QoS};
+use rumqttc::v5::{Client, MqttOptions};
+use std::thread;
+use std::time::Duration;
+
+fn main() {
+    pretty_env_logger::init();
+
+    let mut mqttoptions = MqttOptions::new("test-1", "localhost", 1883);
+    let will = LastWill::new("hello/world", "good bye", QoS::AtMostOnce, false);
+    mqttoptions
+        .set_keep_alive(Duration::from_secs(5))
+        .set_last_will(will);
+
+    let (client, mut connection) = Client::new(mqttoptions, 10);
+    thread::spawn(move || publish(client));
+
+    if let Ok(notification) = connection.recv() {
+        println!("Notification = {:?}", notification);
+    }
+
+    if let Ok(notification) = connection.try_recv() {
+        println!("Notification = {:?}", notification);
+    }
+
+    if let Ok(notification) = connection.recv_timeout(Duration::from_secs(10)) {
+        println!("Notification = {:?}", notification);
+    }
+}
+
+fn publish(client: Client) {
+    client.subscribe("hello/+/world", QoS::AtMostOnce).unwrap();
+    for i in 0..3 {
+        let payload = vec![1; i];
+        let topic = format!("hello/{}/world", i);
+        let qos = QoS::AtLeastOnce;
+
+        client.publish(topic, qos, true, payload).unwrap();
+    }
+
+    thread::sleep(Duration::from_secs(1));
+}
diff --git a/rumqttc/src/client.rs b/rumqttc/src/client.rs
index f763d0d46..851f7d9e8 100644
--- a/rumqttc/src/client.rs
+++ b/rumqttc/src/client.rs
@@ -414,6 +414,9 @@ impl Connection {
     /// [`EvenLoop`]: super::EventLoop
     pub fn try_recv(&mut self) -> Result<Result<Event, ConnectionError>, TryRecvError> {
         let f = self.eventloop.poll();
+        // Enters the runtime context so we can poll the future, as required by `now_or_never()`.
+        // ref: https://docs.rs/tokio/latest/tokio/runtime/struct.Runtime.html#method.enter
+        let _guard = self.runtime.enter();
         let event = f.now_or_never().ok_or(TryRecvError::Empty)?;
 
         resolve_event(event).ok_or(TryRecvError::Disconnected)
@@ -427,13 +430,10 @@ impl Connection {
         &mut self,
         duration: Duration,
     ) -> Result<Result<Event, ConnectionError>, RecvTimeoutError> {
-        // Enter the runtime so we can use Sleep, which is required by timeout.
-        // ref: https://docs.rs/tokio/latest/tokio/runtime/struct.Runtime.html#method.enter
-        let _guard = self.runtime.enter();
-        let f = timeout(duration, self.eventloop.poll());
+        let f = self.eventloop.poll();
         let event = self
             .runtime
-            .block_on(f)
+            .block_on(async { timeout(duration, f).await })
             .map_err(|_| RecvTimeoutError::Timeout)?;
 
         resolve_event(event).ok_or(RecvTimeoutError::Disconnected)
diff --git a/rumqttc/src/v5/client.rs b/rumqttc/src/v5/client.rs
index 246ef031f..bf1380840 100644
--- a/rumqttc/src/v5/client.rs
+++ b/rumqttc/src/v5/client.rs
@@ -417,6 +417,9 @@ impl Connection {
     /// [`EvenLoop`]: super::EventLoop
     pub fn try_recv(&mut self) -> Result<Result<Event, ConnectionError>, TryRecvError> {
         let f = self.eventloop.poll();
+        // Enters the runtime context so we can poll the future, as required by `now_or_never()`.
+        // ref: https://docs.rs/tokio/latest/tokio/runtime/struct.Runtime.html#method.enter
+        let _guard = self.runtime.enter();
         let event = f.now_or_never().ok_or(TryRecvError::Empty)?;
 
         resolve_event(event).ok_or(TryRecvError::Disconnected)
@@ -430,13 +433,10 @@ impl Connection {
         &mut self,
         duration: Duration,
     ) -> Result<Result<Event, ConnectionError>, RecvTimeoutError> {
-        // Enter the runtime so we can use Sleep, which is required by timeout.
-        // ref: https://docs.rs/tokio/latest/tokio/runtime/struct.Runtime.html#method.enter
-        let _guard = self.runtime.enter();
-        let f = timeout(duration, self.eventloop.poll());
+        let f = self.eventloop.poll();
         let event = self
             .runtime
-            .block_on(f)
+            .block_on(async { timeout(duration, f).await })
             .map_err(|_| RecvTimeoutError::Timeout)?;
 
         resolve_event(event).ok_or(RecvTimeoutError::Disconnected)


Test patch: