Spawn task for serial communication

red/src/lib.rs

@@ -1 +1 @@
-pub mod printer;
+pub mod printer;

red/src/main.rs

@@ -1,10 +1,11 @@
 #![warn(rust_2018_idioms)]
 
-use red::printer;
+use red::printer::Printer;
 
 #[tokio::main]
 async fn main() -> tokio_serial::Result<()> {
-    printer::communicate().await?;
-
+    let printer = Printer::connect("").await.unwrap();
+    printer.printer_in.send("M114\n".to_string()).await;
+    tokio::time::sleep(std::time::Duration::from_secs(15)).await;
     Ok(())
-}
+}

red/src/printer/mod.rs

@@ -1,13 +1,69 @@
-use futures::stream::StreamExt;
+use bytes::BytesMut;
 use futures::sink::SinkExt;
-use std::{env, io, str, fmt::Write};
+use futures::stream::StreamExt;
+use std::{fmt::Write, io, rc::Rc, str};
+use tokio;
+use tokio_serial::{SerialPortBuilderExt};
 use tokio_util::codec::{Decoder, Encoder};
 
-use bytes::BytesMut;
-use tokio_serial::SerialPortBuilderExt;
-
 const DEFAULT_TTY: &str = "/dev/ttyUSB0";
 
+#[derive(Debug)]
+pub enum PrinterError {
+    IO(std::io::Error),
+}
+
+pub struct State {
+    position: (f64, f64),
+}
+
+pub struct Printer {
+    pub printer_in: tokio::sync::mpsc::Sender<String>,
+    pub state: Rc<State>,
+}
+
+impl Printer {
+    pub async fn connect(port_path: &str) -> Result<Self, PrinterError> {
+        let mut port = tokio_serial::new(DEFAULT_TTY, 115200)
+            .open_native_async()
+            .expect("Unable to open serial port");
+
+        port.set_exclusive(false)
+            .expect("Unable to set serial port exclusive to false");
+
+        let connection = LineCodec.framed(port);
+        let (tx, mut rx) = tokio::sync::mpsc::channel(32);
+        tokio::spawn(async move {
+            let (mut printer_tx, mut printer_rx) = connection.split();
+            while let Some(stuff) = printer_rx.next().await {
+                if stuff.map(|s| s.contains("Loaded")).unwrap_or(false) {
+                    break;
+                }
+            }
+
+            while let Some(command) = rx.recv().await {
+                printer_tx.send(command).await.unwrap();
+                while let Some(reply) = printer_rx.next().await {
+                    let reply = reply.unwrap();
+                    if reply.contains("Ok") {
+                        break;
+                    }
+                    else {
+                        println!("got reply: {}", reply);
+                    }
+                }
+            }
+        });
+
+        Ok(Printer {
+            printer_in: tx,
+            state: Rc::new(State {
+                position: (0.0, 0.0),
+            }),
+        })
+    }
+}
+
 struct LineCodec;
 
 impl Decoder for LineCodec {
@@ -19,7 +75,10 @@ impl Decoder for LineCodec {
         if let Some(n) = newline {
             let line = src.split_to(n + 1);
             return match str::from_utf8(line.as_ref()) {
-                Ok(s) => Ok(Some(s.to_string())),
+                Ok(s) => {
+                    println!(">>{}", s);
+                    Ok(Some(s.to_string()))
+                }
                 Err(_) => Err(io::Error::new(io::ErrorKind::Other, "Invalid String")),
             };
         }
@@ -31,39 +90,9 @@ impl Encoder<String> for LineCodec {
     type Error = io::Error;
 
     fn encode(&mut self, item: String, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        println!("<<{}", item);
         dst.write_str(&item)
-            .map_err(|e| {io::Error::new(io::ErrorKind::InvalidData, e.to_string())})?;
+            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?;
         Ok(())
     }
-}
-
-pub async fn communicate() -> tokio_serial::Result<()> {
-    let mut args = env::args();
-    let tty_path = args.nth(1).unwrap_or_else(|| DEFAULT_TTY.into());
-
-    let mut port = tokio_serial::new(tty_path, 115200).open_native_async()
-        .expect("Unable to open serial port");
-
-    port.set_exclusive(false)
-        .expect("Unable to set serial port exclusive to false");
-
-    let (mut writer, mut reader) = LineCodec.framed(port).split();
-    println!("Starting read.");
-    let reader_task = async move {
-        let mut counter = 0;
-        while let Some(line_result) = reader.next().await {
-            let line = line_result.expect("Failed to read line");
-            if line.contains("ok") || line.contains("Loaded") {
-                if let Err(e) = writer.send("M114\n".into()).await {
-                    println!("{:?}", e);
-                    break;
-                }
-            }
-            counter += 1;
-            println!("updates received: {}", counter);
-        }
-    };
-    reader_task.await;
-
-    Ok(())
 }