Trouble migrating from v0.3.x to v0.5.x using Tokio spawn

[petewasher-flybrid]

Hello,
Previously I was using a v0.3 which worked great for my system. I am now trying to migrate to v0.5.

I have two main parts of the application:

1. Collects information from a system and provides this as a queue of incoming messages
2. Maintains MQTT connection and delivers these messages to MQTT as they appear from the queue.

This works from two Tokio tasks. I have gone through and migrated from v0.3 syntax to v0.5 syntax, this appears to work fine and the MQTT delivery module I have compiles fine.

When I try and use this from the main application, I have problems with this function from main:

tokio::spawn(async move {
            mqtt_delivery::mqtt_pump(mqtt_socket.clone(), mqtt_provider).await;
        });

Which are various types of:

error: implementation of `Send` is not general enough
   --> src/main.rs:222:9
    |
222 | /         tokio::spawn(async move {
223 | |             mqtt_delivery::mqtt_pump(mqtt_socket.clone(), mqtt_provider).await;
224 | |         });
    | |__________^ implementation of `Send` is not general enough
    |
    = note: `Send` would have to be implemented for the type `&rust_mqtt::client::options::publish::Options<'_>`
    = note: ...but `Send` is actually implemented for the type `&'0 rust_mqtt::client::options::publish::Options<'_>`, for some specific lifetime `'0`

Is there some way I can work through this? It doesn't seem that clear how I can provide either the correct Send or the right lifetime.

Thanks.

[julian-graf]

Hi. I've had the same problem back when getting v0.4 to work and investigated into it quite a bit, but unfortunately I couldn't find a solution.

I have worked around it using async closures and join!, so if you don't have a limitation that your tasks have to run of two actually separate tasks, I would recommend going this route for now. You can take a look at the integration tests, I used this approach there.
Now regarding the problem:
I have seen such errors down at the Readable/Writable trait level, but the compiler messages haven't really been helpful. Back then I thought this was actually related to the implementation, but revisiting this now seems to tell a different story:

• rust_mqtt::client::options::publish::Options<'_> implements Sync (https://docs.rs/rust-mqtt/latest/rust_mqtt/client/options/struct.PublicationOptions.html#synthetic-implementations), so the same type's reference should be Send as the compiler error here dictates. I think the problem might be that tokio::spawn requires a 'static future which requires all references to be 'static as well and since it is impossible to borrow a Options<'non_static_lifetime> for 'static, the future is not Send for 'static. But this is just a guess. Perhaps you might be able to prove to the compiler that the references you pass are lifetime in the Options and all similar types are 'static?

[petewasher-flybrid]

Ok - Thanks for your comments. If it helps here's a fairly simple example following the examples in this repo. You will see the many similar issues if you compile this.

I will see what I can do about the 'static lifetimes. Do you just mean to pass <'static> as the lifetime to the option items?

use tokio::{net::TcpStream, time::sleep};
use embedded_io_adapters::tokio_1::FromTokio;
use rust_mqtt::{
    buffer::AllocBuffer, client::{
        Client,
        options::{
            ConnectOptions,
            PublicationOptions,
            TopicReference
        },
        event::Event,
    },
    config::SessionExpiryInterval,
    types::{
        MqttString,
        TopicName
    },
};

#[derive(Clone)]
pub struct ItemCache {
    pub formatted_value: String,
    pub payload: String,
    pub last_sent: u64,
    is_dirty: bool,
}

pub struct StaticAlloc(AllocBuffer);
impl StaticAlloc {
    pub fn get(&self) -> &'static mut AllocBuffer {
        let inner = &raw const self.0 as *mut AllocBuffer;
        unsafe { &mut *inner }
    }
}
pub static ALLOC: StaticAlloc = StaticAlloc(AllocBuffer);

async fn mqtt_pump(ip_addr: std::net::SocketAddr,
    mut broadcast_rx: tokio::sync::broadcast::Receiver<(String, String, ItemCache)>,
) {
    log::debug!("Starting MQTT pump");

    // Construct the client
    'outer: loop {
        //// rust-mqtt client v0.5.0 boiler plate begins
        // This comes from rust-mqtt/examples/demo.rs
        //
        let mut client = Client::<'_, _, _, 1, 1, 1, 1>::new(ALLOC.get());
        let connection = TcpStream::connect(ip_addr).await.unwrap();
        let connection = FromTokio::new(connection);

        match client.connect(
            connection,
            &ConnectOptions::new()
                .clean_start()
                .session_expiry_interval(SessionExpiryInterval::NeverEnd)
                .user_name(MqttString::from_str("my_datalogger").unwrap()),
            Some(MqttString::from_str("my_datalogger").unwrap()),
        ).await{
            Ok(_) => log::info!("Connected to MQTT broker"),
            Err(e) => {
                log::error!("Error connecting to MQTT broker: {e:?}");
                sleep(std::time::Duration::from_secs(5)).await;
                continue;
            }
        };
        //// rust-mqtt client boiler plate ends
    }
}

#[tokio::main]
async fn main() {

   let (broadcaster, _) = tokio::sync::broadcast::channel::<(
        String,
        String,
        ItemCache,
    )>(64);

   let mqtt_ip: std::net::IpAddr = "127.0.0.1".parse().expect("Whoops");
   let mqtt_socket = core::net::SocketAddr::new(mqtt_ip, 8080);
   let mqtt_subscription_queue = broadcaster.subscribe();

   // Spawn the task
   tokio::spawn(async move {
      mqtt_pump(mqtt_socket.clone(), mqtt_subscription_queue).await;
   });
}

[petewasher-flybrid]

Digging further into your comments, I changed the tokio::spawn to instead create a full thread, which seems to work OK:

std::thread::spawn(move || {
            tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap()
                .block_on(mqtt_delivery::mqtt_pump(mqtt_socket, mqtt_subscription));
        });

[julian-graf]

Thank you for the code snippet. I will play around with it and see if I can find out more.

Interesting that the workaround with a new thread works because std::thread::spawn has basically the same trait bounds...