consortium_ipc/
channel.rs

1// Copyright 2026 Ethan Wu
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14//
15// SPDX-License-Identifier: Apache-2.0
16
17//! Typed, directed communication channels over a [`Transport`].
18//!
19//! `Channel` is the main API for application code. It provides typed `send()` and `recv()` methods that encode and decode messages via a [`Codec`] and hand bytes to the underlying transport.
20use consortium_codec::{Codec, CodecFor};
21use core::marker::PhantomData;
22use core::ops::Deref;
23
24use crate::chan::Chan;
25use crate::transport::{RecvTransport, SendTransport, TransportError};
26
27/// Channel direction marker ([`Tx`] or [`Rx`]).
28///
29/// Sealed against external implementations by a private supertrait: downstream
30/// crates can use `Direction` as a bound but cannot add directions of their own.
31pub trait Direction: crate::sealed::Sealed {}
32
33/// Marker: this channel can only send.
34pub struct Tx;
35
36/// Marker: this channel can only receive.
37pub struct Rx;
38
39impl Direction for Tx {}
40impl Direction for Rx {}
41
42/// A typed, directed communication endpoint over a transport half.
43///
44/// # Type parameters
45///
46/// - `D`  - direction: [`Tx`] or [`Rx`]
47/// - `T`  - message type
48/// - `Tr` - send half ([`SendTransport`]) for `Tx`, receive half ([`RecvTransport`]) for `Rx`
49/// - `C`  - codec; any [`CodecFor<T>`] family, e.g. `consortium_codec::PostcardCodec`
50///
51/// # Buffer ownership
52///
53/// `Channel` does not allocate. The caller provides a `&'static mut [u8]`
54/// scratch buffer at construction. In `no_std` environments this is
55/// typically a `static` array — see `consortium_runtime_mcu::static_buf!` for a
56/// helper that takes the raw `static mut` out of application code:
57///
58/// ```rust,ignore
59/// use consortium_codec::PostcardCodec;
60/// use consortium_ipc::{Channel, Tx};
61///
62/// static mut BUF: [u8; 256] = [0u8; 256];
63///
64/// // Taking the one and only reference to BUF.
65/// let buf = unsafe { &mut *core::ptr::addr_of_mut!(BUF) };
66/// let ch = Channel::<Tx, MyMsg, _, PostcardCodec>::new(chan, transport, buf);
67/// ```
68///
69/// The buffer must be at least `transport.max_send_size()` / `max_recv_size()` bytes.
70pub struct Channel<D, T, Tr, C: CodecFor<T>> {
71    chan: Chan,
72    transport: Tr,
73    buf: &'static mut [u8],
74    _dir: PhantomData<D>,
75    _msg: PhantomData<T>,
76    _codec: PhantomData<C>,
77}
78
79impl<D, T, Tr, C: CodecFor<T>> Channel<D, T, Tr, C> {
80    /// Construct a [`Channel`].
81    ///
82    /// `chan`      - validated channel identifier
83    /// `transport` - initialised transport half
84    /// `buf`       - scratch buffer, sized for the transport MTU
85    pub fn new(chan: Chan, transport: Tr, buf: &'static mut [u8]) -> Self {
86        Self {
87            chan,
88            transport,
89            buf,
90            _dir: PhantomData,
91            _msg: PhantomData,
92            _codec: PhantomData,
93        }
94    }
95
96    /// The [`Chan`] this endpoint is bound to.
97    pub fn chan(&self) -> Chan {
98        self.chan
99    }
100}
101
102impl<T, Tr, C> Channel<Tx, T, Tr, C>
103where
104    Tr: SendTransport,
105    C: CodecFor<T>,
106{
107    /// Encode `msg` with the codec and hand the bytes to the transport.
108    ///
109    /// The transport rings the doorbell internally.  
110    /// Returns when bytes are handed off - not when the remote has
111    /// processed them.
112    pub async fn send(&mut self, msg: &T) -> Result<(), ChannelError<C, Tr>> {
113        let n = C::encode(msg, self.buf).map_err(ChannelError::Codec)?;
114
115        consortium_log::info!("Channel::send: encoded {} bytes on {:?}", n, self.chan);
116
117        self.transport
118            .send(self.buf[..n].as_ref())
119            .await
120            .map_err(ChannelError::Transport)
121    }
122}
123
124impl<T, Tr, C> Channel<Rx, T, Tr, C>
125where
126    Tr: RecvTransport,
127    C: CodecFor<T>,
128{
129    /// Await the next message from the transport and decode it.
130    ///
131    /// Blocks until the transport's doorbell fires and bytes arrive.  
132    /// Returns a [`ReceivedMessage`] wrapping the decoded value.
133    ///
134    /// When `C = PostcardCodec` the message is an owned `T`.  
135    /// When `C` is a future zero-copy codec the returned handle will
136    /// borrow the internal buffer - the `'_` lifetime on
137    /// [`ReceivedMessage`] already accommodates this without any
138    /// change to this signature.
139    pub async fn recv(&mut self) -> Result<ReceivedMessage<'_, T, C>, ChannelError<C, Tr>> {
140        let n = self
141            .transport
142            .recv(self.buf)
143            .await
144            .map_err(ChannelError::Transport)?;
145
146        consortium_log::info!("Channel::recv: received {} bytes on {:?}", n, self.chan);
147
148        let decoded = C::decode(&self.buf[..n]).map_err(ChannelError::Codec)?;
149
150        Ok(ReceivedMessage {
151            value: decoded,
152            _marker: PhantomData,
153        })
154    }
155}
156
157/// A handle to a decoded received message.
158///
159/// For an owning codec such as `consortium_codec::PostcardCodec`
160/// (`Decoded<'buf> = T`) this owns the decoded `T` - the `'a`
161/// lifetime is unused and the value can be moved out freely via
162/// [`into_inner`](ReceivedMessage::into_inner).
163///
164/// For a future zero-copy codec `'a` will borrow the channel's
165/// internal receive buffer, preventing a second `recv()` call
166/// while this handle is live. Application code using [`Deref`]
167/// does not need to change.
168pub struct ReceivedMessage<'a, T: 'a, C: CodecFor<T>> {
169    value: C::Decoded<'a>,
170    _marker: PhantomData<&'a T>,
171}
172
173impl<'a, T, C: CodecFor<T>> ReceivedMessage<'a, T, C> {
174    /// Consume the handle and return the inner decoded value.
175    pub fn into_inner(self) -> C::Decoded<'a> {
176        self.value
177    }
178}
179
180// Deref is only implementable when C::Decoded<'a, T> derefs to T.
181// PostcardCodec: Decoded = T, which derefs trivially via blanket impl.
182// RkyvCodec:     Decoded = &T::Archived, which derefs to T::Archived.
183//
184// We provide the postcard case explicitly. The rkyv case will require
185// a separate impl block or a newtype once that codec is written.
186impl<'a, T, C> Deref for ReceivedMessage<'a, T, C>
187where
188    C: CodecFor<T>,
189    C::Decoded<'a>: core::ops::Deref<Target = T>,
190    T: 'a,
191{
192    type Target = T;
193
194    fn deref(&self) -> &Self::Target {
195        &self.value
196    }
197}
198
199impl<'a, T, C> core::fmt::Debug for ReceivedMessage<'a, T, C>
200where
201    C: CodecFor<T>,
202    C::Decoded<'a>: core::fmt::Debug,
203{
204    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
205        self.value.fmt(f)
206    }
207}
208
209/// Unified error for [`Channel`] operations.
210///
211/// Separates codec failures from transport failures so callers
212/// can handle them independently.
213pub enum ChannelError<C: Codec, Tr: TransportError> {
214    Codec(C::Error),
215    Transport(Tr::Error),
216}
217
218impl<C: Codec, Tr: TransportError> core::fmt::Debug for ChannelError<C, Tr>
219where
220    C::Error: core::fmt::Debug,
221    Tr::Error: core::fmt::Debug,
222{
223    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
224        match self {
225            Self::Codec(err) => f.debug_tuple("Codec").field(err).finish(),
226            Self::Transport(err) => f.debug_tuple("Transport").field(err).finish(),
227        }
228    }
229}
230
231impl<C: Codec, Tr: TransportError> core::fmt::Display for ChannelError<C, Tr>
232where
233    C::Error: core::fmt::Display,
234    Tr::Error: core::fmt::Display,
235{
236    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
237        match self {
238            Self::Codec(e) => write!(f, "codec error: {e}"),
239            Self::Transport(e) => write!(f, "transport error: {e}"),
240        }
241    }
242}
243
244#[cfg(test)]
245mod tests {
246    extern crate std;
247
248    use super::*;
249    use core::fmt;
250    use futures::executor::block_on;
251    use std::boxed::Box;
252    use std::vec::Vec;
253
254    #[derive(Debug, PartialEq)]
255    struct Message {
256        sequence: u8,
257    }
258
259    #[derive(Debug, PartialEq)]
260    enum CodecError {
261        Encode,
262        Decode,
263    }
264
265    impl fmt::Display for CodecError {
266        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
267            write!(f, "{self:?}")
268        }
269    }
270
271    struct MockCodec;
272
273    impl Codec for MockCodec {
274        type Error = CodecError;
275    }
276
277    impl CodecFor<Message> for MockCodec {
278        type Decoded<'buf>
279            = Message
280        where
281            Message: 'buf;
282
283        fn encode(msg: &Message, buf: &mut [u8]) -> Result<usize, Self::Error> {
284            if msg.sequence == 0xFF || buf.is_empty() {
285                return Err(CodecError::Encode);
286            }
287            buf[0] = msg.sequence;
288            Ok(1)
289        }
290
291        fn decode<'buf>(buf: &'buf [u8]) -> Result<Self::Decoded<'buf>, Self::Error>
292        where
293            Message: 'buf,
294        {
295            match buf {
296                [0xEE, ..] | [] => Err(CodecError::Decode),
297                [sequence, ..] => Ok(Message {
298                    sequence: *sequence,
299                }),
300            }
301        }
302    }
303
304    #[derive(Debug, PartialEq)]
305    enum TransportError {
306        Send,
307        Recv,
308        BufferTooSmall,
309    }
310
311    impl fmt::Display for TransportError {
312        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
313            write!(f, "{self:?}")
314        }
315    }
316
317    #[derive(Default)]
318    struct MockTransport {
319        sent: Vec<u8>,
320        recv_payload: Vec<u8>,
321        fail_send: bool,
322        fail_recv: bool,
323    }
324
325    impl crate::transport::TransportError for MockTransport {
326        type Error = TransportError;
327    }
328
329    impl crate::transport::SendTransport for MockTransport {
330        async fn send(&mut self, data: &[u8]) -> Result<(), Self::Error> {
331            if self.fail_send {
332                return Err(TransportError::Send);
333            }
334            self.sent.clear();
335            self.sent.extend_from_slice(data);
336            Ok(())
337        }
338
339        fn max_send_size(&self) -> usize {
340            64
341        }
342    }
343
344    impl crate::transport::RecvTransport for MockTransport {
345        async fn recv(&mut self, buf: &mut [u8]) -> Result<usize, Self::Error> {
346            if self.fail_recv {
347                return Err(TransportError::Recv);
348            }
349            if self.recv_payload.len() > buf.len() {
350                return Err(TransportError::BufferTooSmall);
351            }
352            let n = self.recv_payload.len();
353            buf[..n].copy_from_slice(&self.recv_payload);
354            Ok(n)
355        }
356
357        fn max_recv_size(&self) -> usize {
358            64
359        }
360    }
361
362    impl crate::transport::Transport for MockTransport {}
363
364    fn scratch<const N: usize>() -> &'static mut [u8] {
365        Box::leak(Box::new([0; N])).as_mut_slice()
366    }
367
368    #[test]
369    fn tx_channel_encodes_message_and_sends_bytes() {
370        let ch = unsafe { Chan::new_unchecked(3) };
371        let transport = MockTransport::default();
372        let mut channel = Channel::<Tx, Message, _, MockCodec>::new(ch, transport, scratch::<8>());
373
374        block_on(channel.send(&Message { sequence: 42 })).expect("send succeeds");
375
376        assert_eq!(channel.chan(), ch);
377        assert_eq!(channel.transport.sent, [42]);
378    }
379
380    #[test]
381    fn tx_channel_reports_codec_error_before_transport_send() {
382        let ch = unsafe { Chan::new_unchecked(0) };
383        let transport = MockTransport::default();
384        let mut channel = Channel::<Tx, Message, _, MockCodec>::new(ch, transport, scratch::<8>());
385
386        let err = block_on(channel.send(&Message { sequence: 0xFF }))
387            .expect_err("codec encode should fail");
388
389        assert!(matches!(err, ChannelError::Codec(CodecError::Encode)));
390        assert!(channel.transport.sent.is_empty());
391    }
392
393    #[test]
394    fn tx_channel_reports_transport_send_error() {
395        let ch = unsafe { Chan::new_unchecked(0) };
396        let transport = MockTransport {
397            fail_send: true,
398            ..MockTransport::default()
399        };
400        let mut channel = Channel::<Tx, Message, _, MockCodec>::new(ch, transport, scratch::<8>());
401
402        let err = block_on(channel.send(&Message { sequence: 7 }))
403            .expect_err("transport send should fail");
404
405        assert!(matches!(err, ChannelError::Transport(TransportError::Send)));
406    }
407
408    #[test]
409    fn rx_channel_receives_bytes_and_decodes_message() {
410        let ch = unsafe { Chan::new_unchecked(1) };
411        let transport = MockTransport {
412            recv_payload: std::vec![9],
413            ..MockTransport::default()
414        };
415        let mut channel = Channel::<Rx, Message, _, MockCodec>::new(ch, transport, scratch::<8>());
416
417        let decoded = block_on(channel.recv())
418            .expect("recv succeeds")
419            .into_inner();
420
421        assert_eq!(decoded, Message { sequence: 9 });
422    }
423
424    #[test]
425    fn rx_channel_reports_transport_recv_error() {
426        let ch = unsafe { Chan::new_unchecked(1) };
427        let transport = MockTransport {
428            fail_recv: true,
429            ..MockTransport::default()
430        };
431        let mut channel = Channel::<Rx, Message, _, MockCodec>::new(ch, transport, scratch::<8>());
432
433        let err = block_on(channel.recv()).expect_err("transport recv should fail");
434
435        assert!(matches!(err, ChannelError::Transport(TransportError::Recv)));
436    }
437
438    #[test]
439    fn rx_channel_reports_codec_decode_error() {
440        let ch = unsafe { Chan::new_unchecked(1) };
441        let transport = MockTransport {
442            recv_payload: std::vec![0xEE],
443            ..MockTransport::default()
444        };
445        let mut channel = Channel::<Rx, Message, _, MockCodec>::new(ch, transport, scratch::<8>());
446
447        let err = block_on(channel.recv()).expect_err("codec decode should fail");
448
449        assert!(matches!(err, ChannelError::Codec(CodecError::Decode)));
450    }
451}
consortium_ipc/channel.rs

consortium_ipc/
channel.rs
