active_rpc/
channel.rs

1//! Channels: named streams of messages and capability transfers.
2//!
3//! A channel is a typed, named stream between lanes (or between an agent and a
4//! peer). Direction is declared up front; a receive-only channel refuses sends,
5//! and a send-only channel refuses receives. Messages carry values and may carry
6//! capabilities, which the kernel transfers (moves) rather than copies, so a
7//! transferred capability leaves the sender.
8
9use std::collections::{BTreeSet, VecDeque};
10
11use crate::cap::CapId;
12use crate::error::{Error, Result};
13use crate::lane::LaneId;
14use crate::value::Value;
15
16/// Channel identity.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
18pub struct ChannelId(pub u64);
19
20impl std::fmt::Display for ChannelId {
21    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
22        write!(f, "ch{}", self.0)
23    }
24}
25
26/// Allowed traffic direction.
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub enum Direction {
29    /// The lane may only send.
30    Send,
31    /// The lane may only receive.
32    Receive,
33    /// Both directions.
34    Both,
35}
36
37impl Direction {
38    pub fn can_send(self) -> bool {
39        matches!(self, Direction::Send | Direction::Both)
40    }
41
42    pub fn can_receive(self) -> bool {
43        matches!(self, Direction::Receive | Direction::Both)
44    }
45}
46
47/// One message on a channel.
48#[derive(Debug, Clone, PartialEq)]
49pub struct Message {
50    /// Sending lane.
51    pub from: LaneId,
52    /// Payload value (never secret material).
53    pub payload: Value,
54    /// Capabilities transferred with this message.
55    pub caps: Vec<CapId>,
56    /// Monotonic sequence within the channel.
57    pub sequence: u64,
58}
59
60/// A named channel.
61#[derive(Debug, Clone)]
62pub struct Channel {
63    pub id: ChannelId,
64    pub name: String,
65    pub direction: Direction,
66    queue: VecDeque<Message>,
67    next_sequence: u64,
68    closed: bool,
69    pub subscribers: BTreeSet<u64>,
70    /// Total messages sent (including those already received).
71    pub sent: u64,
72    /// Total messages received.
73    pub received: u64,
74}
75
76impl Channel {
77    pub fn new(id: ChannelId, name: impl Into<String>, direction: Direction) -> Self {
78        Self {
79            id,
80            name: name.into(),
81            direction,
82            queue: VecDeque::new(),
83            next_sequence: 0,
84            closed: false,
85            subscribers: BTreeSet::new(),
86            sent: 0,
87            received: 0,
88        }
89    }
90
91    /// Whether the channel accepts sends.
92    pub fn can_send(&self) -> bool {
93        self.direction.can_send() && !self.closed
94    }
95
96    /// Whether the channel has messages for a receiver.
97    pub fn can_receive(&self) -> bool {
98        self.direction.can_receive() && !self.queue.is_empty()
99    }
100
101    /// Queue a message.
102    pub fn send(&mut self, from: LaneId, payload: Value, caps: Vec<CapId>) -> Result<Message> {
103        if self.closed {
104            return Err(Error::InvalidArgument("channel is closed"));
105        }
106        if !self.direction.can_send() {
107            return Err(Error::InvalidArgument("channel is receive-only"));
108        }
109        let message = Message {
110            from,
111            payload,
112            caps,
113            sequence: self.next_sequence,
114        };
115        self.next_sequence += 1;
116        self.sent += 1;
117        self.queue.push_back(message.clone());
118        Ok(message)
119    }
120
121    /// Take the oldest message.
122    pub fn recv(&mut self) -> Result<Option<Message>> {
123        if !self.direction.can_receive() {
124            return Err(Error::InvalidArgument("channel is send-only"));
125        }
126        match self.queue.pop_front() {
127            Some(message) => {
128                self.received += 1;
129                Ok(Some(message))
130            }
131            None => Ok(None),
132        }
133    }
134
135    /// Peek at the oldest message without consuming it.
136    pub fn peek(&self) -> Option<&Message> {
137        self.queue.front()
138    }
139
140    /// Number of queued messages.
141    pub fn len(&self) -> usize {
142        self.queue.len()
143    }
144
145    pub fn is_empty(&self) -> bool {
146        self.queue.is_empty()
147    }
148
149    /// Close the channel: further sends and receives fail.
150    pub fn close(&mut self) {
151        self.closed = true;
152    }
153
154    pub fn is_closed(&self) -> bool {
155        self.closed
156    }
157}
active_rpc/channel.rs

active_rpc/
channel.rs
