fixes

Signed-off-by: Jess Frazelle <[email protected]>

Author: jessfraz

bambulabs/src/client.rs

@@ -13,61 +13,67 @@ use crate::{
     sequence_id::SequenceId,
 };
 
+const MQTT_PORT: u16 = 8883;
+const MAX_PACKET_SIZE: usize = 1024 * 1024;
+
 /// The Bambu MQTT client.
 #[derive(Clone)]
 pub struct Client {
-    /// The MQTT host.
-    pub host: String,
+    /// The IP address of the MQTT host.
+    pub ip: String,
     /// The access code.
     pub access_code: String,
     /// The serial number.
     pub serial: String,
 
-    client: rumqttc::AsyncClient,
+    topic_device_request: String,
+    topic_device_report: String,
+
+    client: Arc<rumqttc::AsyncClient>,
     event_loop: Arc<Mutex<rumqttc::EventLoop>>,
 
     responses: Arc<DashMap<SequenceId, Message>>,
-
-    topic_device_request: String,
-    topic_device_report: String,
 }
 
-const MAX_PACKET_SIZE: usize = 1024 * 1024;
-
 impl Client {
     /// Creates a new Bambu printer MQTT client.
     pub fn new<S: Into<String> + Clone>(ip: S, access_code: S, serial: S) -> Result<Self> {
         let access_code = access_code.into();
+        let ip = ip.into();
         let serial = serial.into();
-        let host = format!("mqtts://{}:8883", ip.clone().into());
 
+        let opts = Self::get_config(&ip, &access_code)?;
+        let (client, event_loop) = rumqttc::AsyncClient::new(opts, 25);
+
+        Ok(Self {
+            ip,
+            access_code,
+            topic_device_request: format!("device/{}/request", &serial),
+            topic_device_report: format!("device/{}/report", &serial),
+            serial,
+            client: Arc::new(client),
+            event_loop: Arc::new(Mutex::new(event_loop)),
+            responses: Arc::new(DashMap::new()),
+        })
+    }
+
+    fn get_config(ip: &str, access_code: &str) -> Result<rumqttc::MqttOptions> {
         let client_id = format!("bambu-api-{}", nanoid::nanoid!(8));
 
         let ssl_config = rustls::ClientConfig::builder()
             .dangerous()
             .with_custom_certificate_verifier(Arc::new(crate::no_auth::NoAuth::new()))
             .with_no_client_auth();
 
-        let mut opts = rumqttc::MqttOptions::new(client_id, ip, 8883);
+        let mut opts = rumqttc::MqttOptions::new(client_id, ip, MQTT_PORT);
         opts.set_max_packet_size(MAX_PACKET_SIZE, MAX_PACKET_SIZE);
         opts.set_keep_alive(Duration::from_secs(5));
-        opts.set_credentials("bblp", &access_code);
+        opts.set_credentials("bblp", access_code);
         opts.set_transport(rumqttc::Transport::Tls(rumqttc::TlsConfiguration::Rustls(Arc::new(
             ssl_config,
         ))));
 
-        let (client, event_loop) = rumqttc::AsyncClient::new(opts, 25);
-
-        Ok(Self {
-            host,
-            access_code,
-            topic_device_request: format!("device/{}/request", &serial),
-            topic_device_report: format!("device/{}/report", &serial),
-            serial,
-            client,
-            event_loop: Arc::new(Mutex::new(event_loop)),
-            responses: Arc::new(DashMap::new()),
-        })
+        Ok(opts)
     }
 
     /// Polls for a message from the MQTT event loop.
@@ -81,7 +87,27 @@ impl Client {
     ///
     /// Returns an error if there was a problem polling for a message or parsing the event.
     async fn poll(&mut self) -> Result<()> {
-        let msg_opt = self.event_loop.lock().await.poll().await?;
+        let mut ep = self.event_loop.lock().await;
+        let msg_opt = match ep.poll().await {
+            Ok(msg_opt) => msg_opt,
+            Err(err) => {
+                if let rumqttc::ConnectionError::MqttState(rumqttc::StateError::Io(err)) = err {
+                    tracing::error!("Error polling for message: {:?}", err);
+                    tracing::warn!("Reconnecting...");
+                    // We are in a bad state and should reconnect.
+                    let opts = Self::get_config(&self.ip, &self.access_code)?;
+                    let (client, event_loop) = rumqttc::AsyncClient::new(opts, 25);
+                    drop(ep);
+                    self.client = Arc::new(client);
+                    self.event_loop = Arc::new(Mutex::new(event_loop));
+                    tracing::warn!("Reconnected.");
+                    return Ok(());
+                }
+
+                tracing::error!("Error polling for message: {:?}", err);
+                return Ok(());
+            }
+        };
 
         let message = parse_message(&msg_opt);
 
@@ -163,14 +189,16 @@ impl Client {
             if let Some(response) = self.responses.get(sequence_id) {
                 return Ok(response.value().clone());
             }
+            // This sleep is important since it frees up the thread.
+            tokio::time::sleep(Duration::from_secs(1)).await;
         }
 
         anyhow::bail!("Timeout waiting for response to command: {:?}", command)
     }
 
     /// Upload a file.
     pub async fn upload_file(&self, path: &std::path::Path) -> Result<()> {
-        let host_url = url::Url::parse(&self.host)?;
+        let host_url = url::Url::parse(&format!("mqtts://{}:{}", self.ip, MQTT_PORT))?;
         let host = host_url
             .host_str()
             .ok_or(anyhow::anyhow!("not a valid hostname"))?

src/config.rs

@@ -16,8 +16,8 @@ pub struct Config {
 
 impl Config {
     /// Parse a configuration from a toml file.
-    pub fn from_file(file: &PathBuf) -> Result<Self> {
-        let config = std::fs::read_to_string(file)?;
+    pub async fn from_file(file: &PathBuf) -> Result<Self> {
+        let config = tokio::fs::read_to_string(file).await?;
         Self::from_str(&config)
     }
 

src/main.rs

@@ -151,7 +151,7 @@ pub struct Server {
     pub address: String,
 }
 
-#[tokio::main]
+#[tokio::main(flavor = "multi_thread", worker_threads = 128)]
 async fn main() -> Result<()> {
     let opts: Opts = Opts::parse();
 
@@ -221,7 +221,7 @@ async fn main() -> Result<()> {
         delouse::init()?;
     }
 
-    let config = config::Config::from_file(&opts.config)?;
+    let config = config::Config::from_file(&opts.config).await?;
 
     if let Err(err) = run_cmd(&opts, &config).await {
         bail!("running cmd `{:?}` failed: {:?}", &opts.subcmd, err);
@@ -244,7 +244,7 @@ async fn run_cmd(opts: &Opts, config: &Config) -> Result<()> {
 
             let machines = api_context.list_machines()?;
             for (id, machine) in machines.iter() {
-                println!("{}: {:?}", id, machine);
+                println!("{}: {:#?}\n", id, machine);
             }
         }
         SubCommand::PrintFile {

src/server/endpoints.rs

@@ -150,14 +150,16 @@ pub(crate) async fn print_file(
             active_jobs.insert(job_id.to_string(), handle);
         }
         crate::machine::MachineHandle::NetworkPrinter(printer) => {
-            printer
+            let result = printer
                 .client
                 .slice_and_print(&params.job_name, &filepath)
                 .await
                 .map_err(|e| {
                     tracing::error!("failed to print file: {:?}", e);
                     HttpError::for_bad_request(None, "failed to print file".to_string())
                 })?;
+
+            tracing::info!("result: {:?}", result);
         }
     }
 

src/usb_printer.rs

@@ -96,9 +96,10 @@ impl UsbPrinter {
         Ok(gcode)
     }
 
-    fn print(&mut self, file: &std::path::Path) -> Result<Message> {
+    async fn print(&mut self, file: &std::path::Path) -> Result<Message> {
         // Read the gcode file.
-        let lines: Vec<String> = std::fs::read_to_string(file)?
+        let lines: Vec<String> = tokio::fs::read_to_string(file)
+            .await?
             .lines() // split the string into an iterator of string slices
             .map(|s| {
                 let s = String::from(s);
@@ -124,7 +125,7 @@ impl UsbPrinter {
 
     pub async fn slice_and_print(&mut self, file: &std::path::Path) -> Result<Message> {
         let gcode = self.slice(file).await?;
-        self.print(&gcode)
+        self.print(&gcode).await
     }
 
     pub fn status(&self) -> Result<Message> {