Я пытаюсь обернуть синхронный клиент MQTT библиотека с помощью Tokio. Код должен непрерывно получать сообщения по каналу std::sync::mpsc
и отправлять их в код asyn c. Я понимаю, как использовать spawn_blocking
для переноса кода, который возвращает одно значение. Но как это можно применить к обертке oop, которая постоянно получает сообщения из std::sync::mpsc
канала?
Вот код, который я использую для отправки сообщений в канал.
let (mut tx, mut rx) = std::sync::mpsc::channel();
tokio::spawn(async move {
let mut mqtt_options = MqttOptions::new("bot", settings.mqtt.host, settings.mqtt.port);
let (mut mqtt_client, notifications) = MqttClient::start(mqtt_options).unwrap();
mqtt_client.subscribe(settings.mqtt.topic_name, QoS::AtLeastOnce).unwrap();
tokio::task::spawn_blocking(move || {
println!("Waiting for notifications");
for notification in notifications {
match notification {
rumqtt::Notification::Publish(publish) => {
let payload = Arc::try_unwrap(publish.payload).unwrap();
let text: String = String::from_utf8(payload).expect("Can't decode payload for notification");
println!("Recieved message: {}", text);
let msg: Message = serde_json::from_str(&text).expect("Error while deserializing message");
println!("Deserialized message: {:?}", msg);
println!("{}", msg);
tx.send(msg);
}
_ => println!("{:?}", notification)
}
}
});
});
Но я не уверен, как мне использовать Tokio API для получения этих сообщений внутри другого asyn c замыкания.
tokio::task::spawn(async move || {
// How to revieve messages via `rx` here? I can't use tokio::sync::mpsc channels
// since the code that sends messages is blocking.
});