Repository: bytebeamio/rumqtt (language: rust)

Issue:
v5::EventLoop not Send when websocket is enabled
# Expected Behavior

When enabling the feature "websocket", the v5::EventLoop should still be Send as it is the case when "websocket" is not enabled.

# Current Behavior

The v5::EventLoop is not Send anymore, even though the documentation states that the trait Send is implemented.

# Failure Information (for bugs)

```
error: future cannot be sent between threads safely
--> src\main.rs:24:24
|
24  |       tokio::task::spawn(async move {
|  ________________________^
25  | |         loop {
26  | |             match event_loop.poll().await {
27  | |                 Ok(event) => {
...   |
39  | |         }
40  | |     })
| |_____^ future created by async block is not `Send`
|
= help: the trait `Sync` is not implemented for `(dyn Fn(http::request::Request<()>) -> Pin<Box<(dyn Future<Output = http::request::Request<()>> + 'static)>> + 'static)`
note: captured value is not `Send`
--> src\main.rs:26:19
|
26  |             match event_loop.poll().await {
|                   ^^^^^^^^^^ has type `rumqttc::v5::EventLoop` which is not `Send`
note: required by a bound in `tokio::spawn`
--> C:\..path..to..\index.crates.io-6f17d22bba15001f\tokio-1.35.0\src\task\spawn.rs:166:21
|
164 |     pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
|            ----- required by a bound in this function
165 |     where
166 |         F: Future + Send + 'static,
|                     ^^^^ required by this bound in `spawn`
```

## Context

Please provide any relevant information about your setup. This is important in case the issue is not reproducible except for under certain conditions.

* Operating System: Windows
* Toolchain version: 1.73.0
* not tested with non-v5 AsyncClient, only v5

Repository with example code: https://github.com/kaans/rumqttc_websocket_error

## Failure Logs

Please include any relevant log snippets or files here.



Discussion / hints:
Thank you for reporting and providing example to reproduce the bug!

looks like the issue is only in v5 due to use of `http::Request` [here](https://github.com/bytebeamio/rumqtt/blob/6530b1abb510ec431a22a2ab2aa4178d1162c863/rumqttc/src/v5/mod.rs#L103). 

I will have to see why we have that request modifier and what can we do about it! thanks for reporting it! will keep you updated.

Fix patch:
diff --git a/rumqttc/CHANGELOG.md b/rumqttc/CHANGELOG.md
index 842aed732..0c8b65382 100644
--- a/rumqttc/CHANGELOG.md
+++ b/rumqttc/CHANGELOG.md
@@ -25,6 +25,7 @@ To update your code simply remove `Key::ECC()` or `Key::RSA()` from the initiali
 
 ### Fixed
 - Lowered the MSRV to 1.64.0
+- Request modifier function should be Send and Sync and removed unnecessary Box
 
 ### Security
 
diff --git a/rumqttc/src/v5/mod.rs b/rumqttc/src/v5/mod.rs
index a7c1c636d..4b6793b03 100644
--- a/rumqttc/src/v5/mod.rs
+++ b/rumqttc/src/v5/mod.rs
@@ -50,8 +50,11 @@ pub enum Request {
 }
 
 #[cfg(feature = "websocket")]
-type RequestModifierFn =
-    dyn Fn(http::Request<()>) -> Pin<Box<dyn Future<Output = http::Request<()>>>>;
+type RequestModifierFn = Arc<
+    dyn Fn(http::Request<()>) -> Pin<Box<dyn Future<Output = http::Request<()>> + Send>>
+        + Send
+        + Sync,
+>;
 
 // TODO: Should all the options be exposed as public? Drawback
 // would be loosing the ability to panic when the user options
@@ -100,7 +103,7 @@ pub struct MqttOptions {
     /// The server may set its own maximum inflight limit, the smaller of the two will be used.
     outgoing_inflight_upper_limit: Option<u16>,
     #[cfg(feature = "websocket")]
-    request_modifier: Option<Arc<Box<RequestModifierFn>>>,
+    request_modifier: Option<RequestModifierFn>,
 }
 
 impl MqttOptions {
@@ -201,21 +204,20 @@ impl MqttOptions {
     #[cfg(feature = "websocket")]
     pub fn set_request_modifier<F, O>(&mut self, request_modifier: F) -> &mut Self
     where
-        F: Fn(http::Request<()>) -> O + 'static,
-        O: IntoFuture<Output = http::Request<()>>,
+        F: Fn(http::Request<()>) -> O + Send + Sync + 'static,
+        O: IntoFuture<Output = http::Request<()>> + 'static,
+        O::IntoFuture: Send,
     {
-        let request_modifier = Arc::new(Box::new(request_modifier));
-        self.request_modifier = Some(Arc::new(Box::new(move |request| {
-            Box::pin({
-                let request_modifier = Arc::clone(&request_modifier);
-                async move { request_modifier(request).into_future().await }
-            })
-        })));
+        self.request_modifier = Some(Arc::new(move |request| {
+            let request_modifier = request_modifier(request).into_future();
+            Box::pin(request_modifier)
+        }));
+
         self
     }
 
     #[cfg(feature = "websocket")]
-    pub fn request_modifier(&self) -> Option<Arc<Box<RequestModifierFn>>> {
+    pub fn request_modifier(&self) -> Option<RequestModifierFn> {
         self.request_modifier.clone()
     }
 


Test patch: