
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 }
    }
}
