active_rpc/
channel.rs

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

active_rpc/
channel.rs
