The first message comes without any problem but after that I'm getting a StreamError: Closed(..) and no other error messages.
What does this mean and how can I fix this problem?
pub async fn mqtt_connect() -> Result<AsyncClient, paho_mqtt::errors::Error> {
let mqtt_url: String = std::env::var("MQTT_CON").expect("MQTT_CON must be set.");
let host: String = std::env::args()
.nth(1)
.unwrap_or_else(|| mqtt_url);
let id: &str = "isumis_backend";
/*
Create the client. Use an ID for a persistent session.
A real system should try harder to use a unique ID.
*/
let create_opts: CreateOptions = mqtt::CreateOptionsBuilder::new()
.server_uri(host)
.client_id(id)
.finalize();
// Create the client connection
let cli: AsyncClient = AsyncClient::new(create_opts).unwrap_or_else(|e| {
error!("Error creating the client: {}", e);
process::exit(1);
});
// Define the set of options for the connection
let lwt: Message = Message::new("test", "Async subscriber lost connection", mqtt::QOS_1);
let conn_opts: ConnectOptions = mqtt::ConnectOptionsBuilder::new()
.clean_session(false)
.properties(mqtt::properties![mqtt::PropertyCode::SessionExpiryInterval => 3600])
//.keep_alive_interval(Duration::from_secs(15))
.will_message(lwt)
.finalize();
// Make the connection to the broker
info!("Connecting to the MQTT server...");
cli.connect(conn_opts).await?;
match cli.is_connected() {
true => {
warn!("Connection unsuccessful, reconnecting...");
cli.reconnect();
},
false => info!("Connection to MQTT successful.")
};
Ok(cli)
}
pub fn subscriber() -> () {
if let Err(err) = block_on(async {
let mut cli: AsyncClient = mqtt_connect().await?;
// Get message stream before connecting.
let strm: mqtt::AsyncReceiver<Option<Message>> = cli.get_stream(25);
info!("Subscribing to topics: {:?}", TOPICS);
//let sub_opts: Vec<mqtt::SubscribeOptions> = vec![mqtt::SubscribeOptions::with_retain_as_published(); TOPICS.len()];
//cli.subscribe_many_with_options(TOPICS, QOS, &sub_opts, None)
cli.subscribe_many(TOPICS, QOS).await?;
// Just loop on incoming messages.
info!("Waiting for messages...");
/*
Note that we're not providing a way to cleanly shut down and
disconnect. Therefore, when you kill this app (with a ^C or
whatever) the server will get an unexpected drop and then
should emit the LWT message.
*/
message_loop(cli, strm).await;
// Explicit return type for the async block
Ok::<(), mqtt::Error>(())
}) {
error!("{}", err);
}
}
async fn message_loop(cli: AsyncClient, mut strm: mqtt::AsyncReceiver<Option<Message>>) -> () {
while let Some(msg_opt) = strm.next().await {
if let Some(sub_msg) = msg_opt {
if sub_msg.retained() {
info!("(R) ");
}
let msg: &str = match std::str::from_utf8(sub_msg.payload()) {
Ok(v) => v,
Err(e) => panic!("Invalid UTF-8 sequence: {e}"),
};
let payload_json: Value = serde_json::from_str(msg).unwrap();
info!("New message: {}", &payload_json);
match sub_msg.topic() {
TRAFFIC => network_traffic(payload_json)
.await
.expect("Could not handle network traffic"),
FILE_INFO => handle_file_info(msg)
.await
.expect("Could not handle scanned app results"),
BLACKLIGHT => handle_blacklight(payload_json)
.await
.expect("Could not handle blacklight results"),
&_ => error!("No suitable topic name found")
};
} else {
// A "None" means we were disconnected. Try to reconnect...
warn!("Lost connection. Attempting reconnect.");
while let Err(err) = cli.reconnect().await {
error!("Error reconnecting: {}", err);
actix_web::rt::time::sleep(Duration::from_millis(1000)).await;
}
}
}
}
I recieve a part of the message and then I get the error
The first message comes without any problem but after that I'm getting a
StreamError: Closed(..)and no other error messages. What does this mean and how can I fix this problem?I recieve a part of the message and then I get the error