use crate::master::core::EtherCATMaster;
use crate::sugar::state_stream::{SlaveStateChangeEvent, EmergencyEvent};
use std::sync::Mutex;
use tokio::sync::mpsc;
pub struct MasterStateStreamAsync {
rx: mpsc::UnboundedReceiver<SlaveStateChangeEvent>,
}
impl MasterStateStreamAsync {
pub async fn recv(&mut self) -> Option<SlaveStateChangeEvent> {
self.rx.recv().await
}
pub fn into_inner(self) -> mpsc::UnboundedReceiver<SlaveStateChangeEvent> {
self.rx
}
}
pub struct EmergencyStreamAsync {
rx: mpsc::UnboundedReceiver<EmergencyEvent>,
}
impl EmergencyStreamAsync {
pub async fn recv(&mut self) -> Option<EmergencyEvent> {
self.rx.recv().await
}
pub fn into_inner(self) -> mpsc::UnboundedReceiver<EmergencyEvent> {
self.rx
}
}
pub trait MasterStreamAsyncExt {
fn state_stream_async(&self) -> MasterStateStreamAsync;
fn emergency_stream_async(&self) -> EmergencyStreamAsync;
}
impl MasterStreamAsyncExt for EtherCATMaster {
fn state_stream_async(&self) -> MasterStateStreamAsync {
let (tx, rx) = mpsc::unbounded_channel::<SlaveStateChangeEvent>();
let tx = Mutex::new(tx);
self.events().on_slave_state_changed_async(move |master, slave, old_state, new_state| {
let ev = SlaveStateChangeEvent { master, slave, old_state, new_state };
if let Ok(guard) = tx.lock() {
let _ = guard.send(ev);
}
});
MasterStateStreamAsync { rx }
}
fn emergency_stream_async(&self) -> EmergencyStreamAsync {
let (tx, rx) = mpsc::unbounded_channel::<EmergencyEvent>();
let tx = Mutex::new(tx);
self.events().on_emergency(move |master, slave, error_code, error_register, b1, w1, w2| {
let ev = EmergencyEvent {
master, slave, error_code, error_register, b1, w1, w2,
};
if let Ok(guard) = tx.lock() {
let _ = guard.send(ev);
}
});
EmergencyStreamAsync { rx }
}
}