rs_matter/
respond.rs

1/*
2 *
3 *    Copyright (c) 2024-2026 Project CHIP Authors
4 *
5 *    Licensed under the Apache License, Version 2.0 (the "License");
6 *    you may not use this file except in compliance with the License.
7 *    You may obtain a copy of the License at
8 *
9 *        http://www.apache.org/licenses/LICENSE-2.0
10 *
11 *    Unless required by applicable law or agreed to in writing, software
12 *    distributed under the License is distributed on an "AS IS" BASIS,
13 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 *    See the License for the specific language governing permissions and
15 *    limitations under the License.
16 */
17
18use core::fmt::Display;
19use core::future::Future;
20use core::pin::pin;
21
22use embassy_futures::select::{select, select_slice};
23
24use crate::crypto::Crypto;
25use crate::dm::clusters::net_comm;
26use crate::dm::networks::wireless::NoopWirelessNetCtl;
27use crate::dm::{DataModel, ReportDataHandler};
28use crate::error::Error;
29use crate::im::busy::BusyInteractionModel;
30use crate::im::events::DEFAULT_MAX_EVENTS_BUF_SIZE;
31use crate::im::subscriptions::DEFAULT_MAX_SUBSCRIPTIONS;
32use crate::im::{IMBuffer, InteractionModel, PROTO_ID_INTERACTION_MODEL};
33use crate::persist::KvBlobStoreAccess;
34use crate::sc::busy::BusySecureChannel;
35use crate::sc::SecureChannel;
36use crate::transport::exchange::Exchange;
37use crate::utils::select::Coalesce;
38use crate::utils::storage::pooled::Buffers;
39use crate::Matter;
40
41/// Send a busy response if - after that many ms - the exchange
42/// is still not accepted by the regular handlers.
43const RESPOND_BUSY_MS: u32 = 500;
44
45/// A trait modeling a generic handler for an exchange.
46///
47/// The handler takes ownership of the exchange: once handling is done, the only
48/// thing left to do with the exchange is to drop it (which ends it).
49pub trait ExchangeHandler {
50    async fn handle(&self, exchange: Exchange<'_>) -> Result<(), Error>;
51}
52
53impl<T> ExchangeHandler for &T
54where
55    T: ExchangeHandler,
56{
57    fn handle(&self, exchange: Exchange<'_>) -> impl Future<Output = Result<(), Error>> {
58        (*self).handle(exchange)
59    }
60}
61
62/// A struct for chaining two exchange handlers into a single one,
63/// where each handler is handling one specific protocol (i.e. SC vs IM) in a sequential fashion.
64/// I.e. if the first exchange handler refuses to handle the exchange, the second one is tried.
65pub struct ChainedExchangeHandler<H, T> {
66    pub handler_proto: u16,
67    pub handler: H,
68    pub next: T,
69}
70
71impl<H, T> ChainedExchangeHandler<H, T> {
72    /// Construct a chained handler that works as follows:
73    /// - It will call the provided `handler` instance if the protocol ID of the incoming message does match the supplied `handler_proto` value.
74    /// - Otherwise, it will call the `next` handler
75    pub const fn new(handler_proto: u16, handler: H, next: T) -> Self {
76        Self {
77            handler_proto,
78            handler,
79            next,
80        }
81    }
82
83    /// Chain itself with another exchange handler.
84    ///
85    /// The returned chained handler works as follows:
86    /// - It will call the provided `handler` instance if the protocol ID of the incoming message does match the supplied `handler_proto` value.
87    /// - Otherwise, it will call the `self` handler
88    pub const fn chain<H2>(
89        self,
90        handler_proto: u16,
91        handler: H2,
92    ) -> ChainedExchangeHandler<H2, Self> {
93        ChainedExchangeHandler::new(handler_proto, handler, self)
94    }
95}
96
97impl<H, T> ExchangeHandler for ChainedExchangeHandler<H, T>
98where
99    H: ExchangeHandler,
100    T: ExchangeHandler,
101{
102    async fn handle(&self, mut exchange: Exchange<'_>) -> Result<(), Error> {
103        // Peek the protocol id of the incoming message, then hand the exchange
104        // (by value) to the matching handler.
105        exchange.recv_fetch().await?;
106        let proto_id = exchange.rx()?.meta().proto_id;
107
108        if proto_id == self.handler_proto {
109            self.handler.handle(exchange).await
110        } else {
111            self.next.handle(exchange).await
112        }
113    }
114}
115
116/// An [`ExchangeHandler`] that handles nothing - a convenient terminator for a
117/// [`ChainedExchangeHandler`].
118///
119/// By the time the chain reaches it, no handler matched the incoming protocol, so
120/// it simply drops the exchange (sending no reply); the peer will time out and
121/// retry as it would against any unsupported protocol.
122pub struct EmptyExchangeHandler;
123
124impl ExchangeHandler for EmptyExchangeHandler {
125    async fn handle(&self, _exchange: Exchange<'_>) -> Result<(), Error> {
126        Ok(())
127    }
128}
129
130/// A generic responder utility for accepting and handling exchanges received by the provided `Matter` stack,
131/// by applying the provided `ExchangeHandler` instance to each accepted exchange.
132///
133/// This responder uses an intra-task concurrency model - without an external executor - where all handling is done as a single future.
134pub struct Responder<'a, T> {
135    name: &'a str,
136    handler: T,
137    matter: &'a Matter<'a>,
138    respond_after_ms: u32,
139}
140
141impl<'a, T> Responder<'a, T>
142where
143    T: ExchangeHandler,
144{
145    /// Create a new responder.
146    ///
147    /// The `respond_after_ms` parameter instructs the responder how much time to wait before accepting an exchange.
148    ///
149    /// This is useful when utilizing multiple responders on a single `Matter` instance, where e.g. the first (main) responder is the actual one,
150    /// responsible for handling the incoming exchanges, while e.g. another one - with a non-zero `respond_after_ms` - is answerring all exchanges
151    /// not accepted in time by the main responder with a simple "I'm busy, try again later" handling.
152    #[inline(always)]
153    pub const fn new(
154        name: &'a str,
155        handler: T,
156        matter: &'a Matter<'a>,
157        respond_after_ms: u32,
158    ) -> Self {
159        Self {
160            name,
161            handler,
162            matter,
163            respond_after_ms,
164        }
165    }
166
167    /// Get the name of this responder
168    pub const fn name(&self) -> &str {
169        self.name
170    }
171
172    /// Get a reference to the `ExchangeHandler` instance used by this responder
173    pub fn handler(&self) -> &T {
174        &self.handler
175    }
176
177    /// Run the responder with a given number of handlers.
178    pub async fn run<const N: usize>(&self) -> Result<(), Error> {
179        info!("{}: Creating {} handlers", self.name, N);
180
181        let mut handlers = heapless::Vec::<_, N>::new();
182        debug!(
183            "{}: Handlers size: {}B",
184            self.name,
185            core::mem::size_of_val(&handlers)
186        );
187
188        for handler_id in 0..N {
189            unwrap!(handlers.push(self.handle(handler_id)).map_err(|_| ())); // Cannot fail because the vector has size N
190        }
191
192        let handlers = pin!(handlers);
193        let handlers = unsafe { handlers.map_unchecked_mut(|handlers| handlers.as_mut_slice()) };
194
195        select_slice(handlers).await.0
196    }
197
198    /// A handler for one exchange.
199    #[inline(always)]
200    pub async fn handle(&self, handler_id: impl Display) -> Result<(), Error> {
201        loop {
202            // Ignore the error as it had been logged already
203            let _ = self.respond_once(&handler_id).await;
204        }
205    }
206
207    /// Respond to a single exchange.
208    /// Useful in e.g. integration tests, where we know that we are expecting to respond to a single exchange within the run of the test.
209    #[inline(always)]
210    pub async fn respond_once(&self, handler_id: impl Display) -> Result<(), Error> {
211        let exchange = Exchange::accept_after(self.matter, self.respond_after_ms).await?;
212        // Capture the id up front: the handler takes the exchange by value.
213        let exchange_id = exchange.id();
214
215        if self.log_warn() {
216            warn!(
217                "{}: Handler {} / exchange {}: Starting",
218                self.name,
219                display2format!(&handler_id),
220                exchange_id
221            );
222        } else {
223            debug!(
224                "{}: Handler {} / exchange {}: Starting",
225                self.name,
226                display2format!(&handler_id),
227                exchange_id
228            );
229        }
230
231        let result = self.handler.handle(exchange).await;
232
233        if let Err(err) = &result {
234            error!(
235                "{}: Handler {} / exchange {}: Abandoned because of error {:?}",
236                self.name,
237                display2format!(&handler_id),
238                exchange_id,
239                err
240            );
241        } else if self.log_warn() {
242            warn!(
243                "{}: Handler {} / exchange {}: Completed",
244                self.name,
245                display2format!(&handler_id),
246                exchange_id
247            );
248        } else {
249            debug!(
250                "{}: Handler {} / exchange {}: Completed",
251                self.name,
252                display2format!(&handler_id),
253                exchange_id
254            );
255        }
256
257        result
258    }
259
260    fn log_warn(&self) -> bool {
261        self.respond_after_ms > 0
262    }
263}
264
265/// A type alias for the "default" responder handler, which is a chained handler of the `InteractionModel` and `SecureChannel` handlers.
266pub type DefaultExchangeHandler<
267    'd,
268    'a,
269    C,
270    B,
271    T,
272    K,
273    N,
274    NC = NoopWirelessNetCtl,
275    R = (),
276    const NS: usize = DEFAULT_MAX_SUBSCRIPTIONS,
277    const NE: usize = DEFAULT_MAX_EVENTS_BUF_SIZE,
278> = ChainedExchangeHandler<
279    &'d InteractionModel<'a, C, B, T, K, N, NC, R, NS, NE>,
280    SecureChannel<'d, &'d C>,
281>;
282
283impl<'d, 'a, C, B, T, K, N, NC, R, const NS: usize, const NE: usize>
284    Responder<'a, DefaultExchangeHandler<'d, 'a, C, B, T, K, N, NC, R, NS, NE>>
285where
286    B: Buffers<IMBuffer>,
287{
288    /// Creates a "default" responder. This is a responder that composes and uses the `rs-matter`-provided `ExchangeHandler` implementations
289    /// (`SecureChannel` and `InteractionModel`) for handling the Secure Channel protocol and the Interaction Model protocol.
290    #[inline(always)]
291    pub const fn new_default(
292        data_model: &'d InteractionModel<'a, C, B, T, K, N, NC, R, NS, NE>,
293    ) -> Self
294    where
295        C: Crypto,
296        T: DataModel,
297        K: KvBlobStoreAccess,
298        N: net_comm::Networks,
299        R: ReportDataHandler,
300    {
301        Self::new(
302            "Responder",
303            ChainedExchangeHandler::new(
304                PROTO_ID_INTERACTION_MODEL,
305                data_model,
306                SecureChannel::new(data_model.crypto(), data_model),
307            ),
308            data_model.matter(),
309            0,
310        )
311    }
312}
313
314/// A type alias for the "busy" responder handler, which is a chained handler of the `BusyInteractionModel` and `BusySecureChannel` handlers.
315pub type BusyExchangeHandler = ChainedExchangeHandler<BusyInteractionModel, BusySecureChannel>;
316
317impl<'a> Responder<'a, BusyExchangeHandler> {
318    /// Creates a simple "busy" responder, which is answering all exchanges with a simple "I'm busy, try again later" handling.
319    /// The resonder is using the `rs-matter`-provided `ExchangeHandler` instances (`BusySecureChannel` and `BusyInteractionModel`)
320    /// capable of answering with "busy" messages the SC and IM protocols, respectively.
321    ///
322    /// Exchanges which are not accepted after the specified milliseconds are answered by this responder,
323    /// as the assumption is that the main responder is busy and cannot answer these right now.
324    #[inline(always)]
325    pub const fn new_busy(matter: &'a Matter<'a>, respond_after_ms: u32) -> Self {
326        Self::new(
327            "Busy Responder",
328            ChainedExchangeHandler::new(
329                PROTO_ID_INTERACTION_MODEL,
330                BusyInteractionModel::new(),
331                BusySecureChannel::new(),
332            ),
333            matter,
334            respond_after_ms,
335        )
336    }
337}
338
339/// A composition of the `Responder::new_default` and `Responder::new_busy` responders.
340pub struct DefaultResponder<
341    'd,
342    'a,
343    C,
344    B,
345    T,
346    K,
347    N,
348    NC = NoopWirelessNetCtl,
349    R = (),
350    const NS: usize = DEFAULT_MAX_SUBSCRIPTIONS,
351    const NE: usize = DEFAULT_MAX_EVENTS_BUF_SIZE,
352> where
353    B: Buffers<IMBuffer>,
354{
355    responder: Responder<'a, DefaultExchangeHandler<'d, 'a, C, B, T, K, N, NC, R, NS, NE>>,
356    busy_responder: Responder<'a, BusyExchangeHandler>,
357}
358
359impl<'d, 'a, C, B, T, K, N, NC, R, const NS: usize, const NE: usize>
360    DefaultResponder<'d, 'a, C, B, T, K, N, NC, R, NS, NE>
361where
362    C: Crypto,
363    B: Buffers<IMBuffer>,
364    T: DataModel,
365    K: KvBlobStoreAccess,
366    N: net_comm::Networks,
367    R: ReportDataHandler,
368{
369    /// Creates the responder composition.
370    #[inline(always)]
371    pub const fn new(data_model: &'d InteractionModel<'a, C, B, T, K, N, NC, R, NS, NE>) -> Self {
372        Self {
373            responder: Responder::new_default(data_model),
374            busy_responder: Responder::new_busy(data_model.matter(), RESPOND_BUSY_MS),
375        }
376    }
377
378    /// Run the responder.
379    pub async fn run<const A: usize, const O: usize>(&self) -> Result<(), Error> {
380        let mut actual = pin!(self.responder.run::<A>());
381        let mut busy = pin!(self.busy_responder.run::<O>());
382
383        select(&mut actual, &mut busy).coalesce().await
384    }
385
386    /// Get a reference to the main responder.
387    ///
388    /// Useful when the user would like to organize its own herd of responders rather than using the `run` method.
389    #[allow(clippy::type_complexity)]
390    pub const fn responder(
391        &self,
392    ) -> &Responder<
393        'a,
394        ChainedExchangeHandler<
395            &'d InteractionModel<'a, C, B, T, K, N, NC, R, NS, NE>,
396            SecureChannel<'d, &'d C>,
397        >,
398    > {
399        &self.responder
400    }
401
402    /// Get a reference to the busy responder.
403    ///
404    /// Useful when the user would like to organize its own herd of busy responders rather than using the `run` method.
405    pub const fn busy_responder(
406        &self,
407    ) -> &Responder<'a, ChainedExchangeHandler<BusyInteractionModel, BusySecureChannel>> {
408        &self.busy_responder
409    }
410}
rs_matter/respond.rs

rs_matter/
respond.rs
