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    pub fn can_send(self) -> bool {
44        matches!(self, Direction::Send | Direction::Both)
45    }
46
47    pub fn can_receive(self) -> bool {
48        matches!(self, Direction::Receive | Direction::Both)
49    }
50}
51
52/// One message on a channel.
53#[derive(Debug, Clone, PartialEq)]
54pub struct Message {
55    /// Sending lane.
56    pub from: LaneId,
57    /// Payload value (never secret material).
58    pub payload: Value,
59    /// Capabilities transferred with this message.
60    pub caps: Vec<CapId>,
61    /// Monotonic sequence within the channel.
62    pub sequence: u64,
63}
64
65/// A named channel.
66#[derive(Debug, Clone)]
67pub struct Channel {
68    pub id: ChannelId,
69    pub name: String,
70    pub direction: Direction,
71    queue: VecDeque<Message>,
72    next_sequence: u64,
73    closed: bool,
74    pub subscribers: BTreeSet<u64>,
75    /// Total messages sent (including those already received).
76    pub sent: u64,
77    /// Total messages received.
78    pub received: u64,
79}
80
81impl Channel {
82    pub fn new(id: ChannelId, name: impl Into<String>, direction: Direction) -> Self {
83        Self {
84            id,
85            name: name.into(),
86            direction,
87            queue: VecDeque::new(),
88            next_sequence: 0,
89            closed: false,
90            subscribers: BTreeSet::new(),
91            sent: 0,
92            received: 0,
93        }
94    }
95
96    /// Whether the channel accepts sends.
97    pub fn can_send(&self) -> bool {
98        self.direction.can_send() && !self.closed
99    }
100
101    /// Whether the channel has messages for a receiver.
102    pub fn can_receive(&self) -> bool {
103        self.direction.can_receive() && !self.queue.is_empty()
104    }
105
106    /// Queue a message.
107    pub fn send(&mut self, from: LaneId, payload: Value, caps: Vec<CapId>) -> Result<Message> {
108        if self.closed {
109            return Err(Error::InvalidArgument("channel is closed"));
110        }
111        if !self.direction.can_send() {
112            return Err(Error::InvalidArgument("channel is receive-only"));
113        }
114        let message = Message {
115            from,
116            payload,
117            caps,
118            sequence: self.next_sequence,
119        };
120        self.next_sequence += 1;
121        self.sent += 1;
122        self.queue.push_back(message.clone());
123        Ok(message)
124    }
125
126    /// Take the oldest message.
127    pub fn recv(&mut self) -> Result<Option<Message>> {
128        if !self.direction.can_receive() {
129            return Err(Error::InvalidArgument("channel is send-only"));
130        }
131        match self.queue.pop_front() {
132            Some(message) => {
133                self.received += 1;
134                Ok(Some(message))
135            }
136            None => Ok(None),
137        }
138    }
139
140    /// Peek at the oldest message without consuming it.
141    pub fn peek(&self) -> Option<&Message> {
142        self.queue.front()
143    }
144
145    /// Number of queued messages.
146    pub fn len(&self) -> usize {
147        self.queue.len()
148    }
149
150    pub fn is_empty(&self) -> bool {
151        self.queue.is_empty()
152    }
153
154    /// Close the channel: further sends and receives fail.
155    pub fn close(&mut self) {
156        self.closed = true;
157    }
158
159    pub fn is_closed(&self) -> bool {
160        self.closed
161    }
162}
active_rpc/channel.rs

active_rpc/
channel.rs
