async

Author: lattice0

examples/cc253x.rs

@@ -20,7 +20,7 @@ fn main() {
         let b = async {
             println!("version: {:?}", cc2531.version().await);
         };
-        futures::join!(a, b);
+        futures::join!(b, a);
     };
     let executor = Executor::default();
 

src/cc253x.rs

@@ -15,7 +15,8 @@ const MAXIMUM_ZIGBEE_PAYLOAD_SIZE: usize = 255;
 
 pub struct CC2531X {
     _supports_led: Option<bool>,
-    serial: Arc<Mutex<Box<dyn SerialPort>>>,
+    read: Arc<Mutex<Box<dyn SerialPort>>>,
+    write: Arc<Mutex<Box<dyn SerialPort>>>,
 }
 
 impl CC2531X {
@@ -25,7 +26,10 @@ impl CC2531X {
             .open()
             .map_err(|_e| CoordinatorError::SerialOpen)?;
         Ok(Self {
-            serial: Arc::new(Mutex::new(serial)),
+            read: Arc::new(Mutex::new(
+                serial.try_clone().map_err(|_e| CoordinatorError::Io)?,
+            )),
+            write: Arc::new(Mutex::new(serial)),
             _supports_led: None,
         })
     }
@@ -37,12 +41,12 @@ impl CC2531X {
         subsystem: Subsystem,
         _timeout: Option<std::time::Duration>,
     ) -> Result<ParametersValueMap, CoordinatorError> {
+        log!("waiting for {:?}", name);
         let command =
             get_command_by_name(&subsystem, name).ok_or(CoordinatorError::NoCommandWithName)?;
         let mut buffer = [0; MAX_FRAME_SIZE];
-        let len = self
-            .serial
-            .lock()
+        let lock = self.read.lock();
+        let len = lock
             .await
             .read(&mut buffer)
             .map_err(|_e| CoordinatorError::Io)?;
@@ -89,14 +93,15 @@ impl Coordinator for CC2531X {
         let command = get_command_by_name(&Subsystem::Sys, "version")
             .ok_or(CoordinatorError::NoCommandWithName)?;
         let send = async {
-            let mut lock = self.serial.lock().await;
+            let mut lock = self.write.lock().await;
             UnpiPacket::from_command_to_serial_async(
                 command.id,
                 command,
                 &[],
                 (MessageType::SREQ, Subsystem::Sys),
                 &mut **lock,
-            ).await
+            )
+            .await
         };
         let wait = self.wait_for("version", MessageType::SRESP, Subsystem::Sys, None);
         let r = futures::try_join!(send, wait)?;
@@ -111,7 +116,7 @@ impl Coordinator for CC2531X {
             ResetType::Hard => &[("type", ParameterValue::U8(0))],
         };
 
-        let mut lock = self.serial.lock().await;
+        let mut lock = self.write.lock().await;
         UnpiPacket::from_command_to_serial(
             command.id,
             command,
@@ -128,7 +133,7 @@ impl Coordinator for CC2531X {
             .ok_or(CoordinatorError::NoCommandWithName)?;
         //TODO: const firmwareControlsLed = parseInt(this.version.revision) >= 20211029;
         let firmware_controls_led = true;
-        let mut lock = self.serial.lock().await;
+        let mut lock = self.write.lock().await;
         let parameters = match led_status {
             LedStatus::Disable => {
                 if firmware_controls_led {
@@ -187,7 +192,7 @@ impl Coordinator for CC2531X {
 
         let command = get_command_by_name(&Subsystem::Zdo, "management_network_update_request")
             .ok_or(CoordinatorError::NoCommandWithName)?;
-        let mut lock = self.serial.lock().await;
+        let mut lock = self.write.lock().await;
         UnpiPacket::from_command_to_serial(
             command.id,
             command,
@@ -207,7 +212,7 @@ impl Coordinator for CC2531X {
 
         let command = get_command_by_name(&Subsystem::Zdo, "stack_tune")
             .ok_or(CoordinatorError::NoCommandWithName)?;
-        let mut lock = self.serial.lock().await;
+        let mut lock = self.write.lock().await;
         UnpiPacket::from_command_to_serial(
             command.id,
             command,

src/coordinator.rs

@@ -58,6 +58,7 @@ pub enum ResetType {
 pub enum CoordinatorError {
     SerialOpen,
     SerialWrite,
+    SerialRead,
     NoCommandWithName,
     Io,
     Parameter(ParameterError),
@@ -66,6 +67,7 @@ pub enum CoordinatorError {
     Map(MapError),
     NoRequest,
     NoResponse,
+    
 }
 
 impl From<std::io::Error> for CoordinatorError {

src/serial.rs

@@ -2,14 +2,79 @@ use crate::{
     coordinator::CoordinatorError,
     unpi::{
         commands::{Command, ParameterValue},
-        LenTypeInfo, MessageType, Subsystem, UnpiPacket, MAX_PAYLOAD_SIZE,
-    }, utils::log,
+        LenTypeInfo, MessageType, Subsystem, UnpiPacket, UnpiPacketSink, MAX_PAYLOAD_SIZE,
+    },
+    utils::log,
+};
+use futures::channel::oneshot;
+use futures::{
+    channel::oneshot::{Receiver, Sender},
+    lock::Mutex,
+    AsyncReadExt, AsyncWrite, AsyncWriteExt,
 };
-use futures::executor::block_on;
 use serialport::SerialPort;
+use std::{future::Future, sync::Arc};
+use futures::task::Spawn;
 
-impl<'a> UnpiPacket<'a> {
+pub struct SimpleSerialPort {
+    read: Arc<Mutex<Box<dyn SerialPort>>>,
+    write: Arc<Mutex<Box<dyn SerialPort>>>,
+    tx: Sender<UnpiPacket<'static>>,
+    rx: Receiver<UnpiPacket<'static>>,
+}
+
+impl SimpleSerialPort {
+    pub fn new(path: &str, baud_rate: u32) -> Result<Self, CoordinatorError> {
+        let (tx, rx) = oneshot::channel();
+        Ok(SimpleSerialPort {
+            read: Arc::new(Mutex::new(
+                serialport::new(path, baud_rate)
+                    .timeout(std::time::Duration::from_millis(10))
+                    .open()
+                    .map_err(|_e| CoordinatorError::SerialOpen)?,
+            )),
+            write: Arc::new(Mutex::new(
+                serialport::new(path, baud_rate)
+                    .timeout(std::time::Duration::from_millis(10))
+                    .open()
+                    .map_err(|_e| CoordinatorError::SerialOpen)?,
+            )),
+            tx,
+            rx,
+        })
+    }
 
+    pub async fn start(&self) -> Result<(), CoordinatorError> {
+        let mut read = self.read.lock().await;
+        let mut write = self.write.lock().await;
+        let (tx, rx) = (self.tx, self.rx);
+        let mut buffer = [0u8; 256];
+        let receive = async {
+            loop {
+                let len = read.read(&mut buffer).map_err(|_e| CoordinatorError::SerialRead)?;
+                let packet = UnpiPacket::from_payload(
+                    (&buffer[..len], LenTypeInfo::OneByte),
+                    (MessageType::SREQ, Subsystem::Sys),
+                    0,
+                )?;
+                log!("<<< {:?}", packet);
+                tx.send(packet).map_err(|_e| CoordinatorError::SerialWrite)?;
+            }
+        };
+        let send = async {
+            loop {
+                let packet = rx.await.map_err(|_e| CoordinatorError::SerialWrite)?;
+                let mut buffer = [0u8; 256];
+                packet.to_bytes(&mut buffer)?;
+                write.write_all(buffer).map_err(|_e| CoordinatorError::SerialWrite)?;
+                
+            }
+        };
+        todo!()
+    }
+}
+
+impl<'a> UnpiPacket<'a> {
     /// Serialized the packet to the serial port
     pub fn to_serial<S: SerialPort + ?Sized>(
         &self,
@@ -52,8 +117,6 @@ impl<'a> UnpiPacket<'a> {
         type_subsystem: (MessageType, Subsystem),
         serial: &mut S,
     ) -> Result<(), CoordinatorError> {
-        block_on(async move {
-            Self::from_command_to_serial(command_id, command, parameters, type_subsystem, serial)
-        })
+        Self::from_command_to_serial(command_id, command, parameters, type_subsystem, serial)
     }
 }

src/unpi/mod.rs

@@ -1,5 +1,5 @@
 use crate::utils::slice_reader::SliceReader;
-use std::io::Write;
+use std::{future::Future, io::Write};
 
 pub mod commands;
 pub mod subsystems;
@@ -24,6 +24,14 @@ pub struct UnpiPacket<'a> {
     pub fcs: u8,
 }
 
+pub trait UnpiPacketSink {
+    fn write(&mut self, packet: &UnpiPacket) -> impl Future<Output = Result<(), UnpiPacketError>>;
+}
+
+pub trait UnpiPacketSource {
+    fn read(&mut self) -> impl Future<Output = Result<UnpiPacket, UnpiPacketError>>;
+}
+
 struct Wrapped<T>(T);
 
 #[derive(Debug, PartialEq, Copy, Clone)]