You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
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?
pubasyncfnmqtt_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 connectionlet 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 connectionlet 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 brokerinfo!("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)}pubfnsubscriber() -> (){ifletErr(err) = block_on(async{letmut 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 blockOk::<(), mqtt::Error>(())}){error!("{}", err);}}asyncfnmessage_loop(cli:AsyncClient,mutstrm: mqtt::AsyncReceiver<Option<Message>>) -> (){whileletSome(msg_opt) = strm.next().await{ifletSome(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.");whileletErr(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
The text was updated successfully, but these errors were encountered: