ezsp 15.0.0

Ember ZNet Serial Protocol
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

use std::future::Future;

use log::{error, trace, warn};
use tokio::sync::mpsc;

use crate::api::Message;
use crate::frame::Frame;
use crate::parameters::configuration;
use crate::{Callback, Error, Parameters, Response};

/// Receives decoded frames from a transport-specific inbound stream.
///
/// The transport performs link-layer I/O and EZSP byte decoding before yielding
/// a complete [`Frame<Parameters>`](Frame). The generic receiver actor then
/// routes responses, synchronous callback-shaped responses, and asynchronous
/// callbacks to their respective consumers.
///
/// The receiver actor passes the currently negotiated protocol version to each
/// call so transports can switch between legacy and extended EZSP header
/// decoding. An `ASHv2` implementation decodes one complete `ASHv2` DATA
/// payload into one [`Frame<Parameters>`](Frame), using legacy headers while
/// the supplied version is `None`.
pub trait Receive {
    /// Receives the next decoded frame using `negotiated_version`, or returns
    /// `None` when the input closes.
    ///
    /// The receiver actor passes `None` until it observes the initial `version`
    /// response. Subsequent calls receive `Some(version)` so the transport can
    /// select version-sensitive decoding without storing negotiation state.
    ///
    /// This method has no error result. The transport implementation therefore
    /// owns its malformed-frame policy, for example logging and skipping an
    /// invalid frame or closing the input.
    fn receive(
        &mut self,
        negotiated_version: Option<u8>,
    ) -> impl Future<Output = Option<Frame<Parameters>>> + Send;
}

/// Routes received EZSP frames to the transmitter actor or callback stream.
#[derive(Debug)]
pub struct Receiver<T> {
    receive: T,
    callbacks: mpsc::Sender<Callback>,
    transmitter: mpsc::Sender<Message>,
    negotiated_version: Option<u8>,
}

impl<T> Receiver<T>
where
    T: Receive,
{
    /// Creates a receiver task around a transport-specific inbound half.
    #[must_use]
    pub const fn new(
        receive: T,
        callbacks: mpsc::Sender<Callback>,
        transmitter: mpsc::Sender<Message>,
    ) -> Self {
        Self {
            receive,
            callbacks,
            transmitter,
            negotiated_version: None,
        }
    }

    async fn handle_frame(&mut self, frame: Frame<Parameters>) -> Result<(), Error> {
        let (header, payload) = frame.into();

        if let Parameters::Response(Response::Configuration(configuration::Response::Version(
            version,
        ))) = &payload
            && let Some(previous_version) =
                self.negotiated_version.replace(version.protocol_version())
        {
            error!(
                "Replaced previous version {previous_version} with version {}.",
                version.protocol_version()
            );
        }

        match payload {
            Parameters::Response(response) => {
                trace!("Forwarding response: {response:?}");
                self.transmitter
                    .send(Message::Response(Frame::new(
                        header,
                        Parameters::Response(response),
                    )))
                    .await?;
            }
            Parameters::Callback(callback) => {
                if header.is_async_callback() {
                    trace!("Forwarding async callback: {callback:?}");
                    self.callbacks.send(callback).await.unwrap_or_else(|error| {
                        warn!("Callback channel closed: {error}");
                    });
                } else {
                    trace!("Forwarding non-async callback as response: {callback:?}");
                    self.transmitter
                        .send(Message::Response(Frame::new(
                            header,
                            Parameters::Callback(callback),
                        )))
                        .await?;
                }
            }
        }

        Ok(())
    }
}

impl<T> Receiver<T>
where
    T: Receive + Send,
{
    /// Runs until the inbound stream or the transmitter actor channel closes.
    pub async fn run(mut self) {
        while let Some(frame) = self.receive.receive(self.negotiated_version).await {
            if let Err(error) = self.handle_frame(frame).await {
                warn!("{error}");
                return;
            }
        }
    }
}