phantom_protocol/api/
stream.rs

1use crate::api::session::SessionCommand;
2use crate::errors::CoreError;
3use crate::transport::multiplexer::{StreamHandle, StreamMessage};
4use bytes::Bytes;
5use tokio::sync::mpsc;
6use tokio::sync::Mutex;
7
8/// A single multiplexed stream inside an established [`PhantomSession`].
9///
10/// Created by the session's stream multiplexer (one per logical stream id).
11/// Outbound data is queued to the session's data pump over the `tx` command
12/// channel (`send_reliable` / `send_unreliable`); inbound demultiplexed data
13/// arrives on `rx`. The session owns all encryption and transport — a
14/// `PhantomStream` is just the per-stream send/recv handle exposed over FFI.
15///
16/// [`PhantomSession`]: crate::api::session::PhantomSession
17#[cfg_attr(feature = "bindings", derive(uniffi::Object))]
18pub struct PhantomStream {
19    stream_id: u32,
20    /// Channel to send data to the session to be packaged and sent
21    tx: mpsc::Sender<SessionCommand>,
22    /// Receiver for incoming demultiplexed stream data
23    rx: Mutex<mpsc::Receiver<StreamMessage>>,
24}
25
26impl PhantomStream {
27    pub fn new(handle: StreamHandle, tx: mpsc::Sender<SessionCommand>) -> Self {
28        Self {
29            stream_id: handle.stream_id,
30            tx,
31            rx: Mutex::new(handle.rx),
32        }
33    }
34}
35
36#[cfg_attr(feature = "bindings", uniffi::export(async_runtime = "tokio"))]
37impl PhantomStream {
38    pub fn stream_id(&self) -> u32 {
39        self.stream_id
40    }
41
42    pub async fn send_reliable(&self, data: Vec<u8>) -> Result<(), CoreError> {
43        self.tx
44            .send(SessionCommand::SendStreamReliable {
45                stream_id: self.stream_id,
46                data: Bytes::from(data),
47            })
48            .await
49            .map_err(|_| CoreError::NetworkError("Session closed".into()))
50    }
51
52    pub async fn send_unreliable(&self, data: Vec<u8>) -> Result<(), CoreError> {
53        self.tx
54            .send(SessionCommand::SendStreamUnreliable {
55                stream_id: self.stream_id,
56                data: Bytes::from(data),
57            })
58            .await
59            .map_err(|_| CoreError::NetworkError("Session closed".into()))
60    }
61
62    pub async fn recv(&self) -> Result<Vec<u8>, CoreError> {
63        let mut rx = self.rx.lock().await;
64        loop {
65            match rx.recv().await {
66                Some(StreamMessage::Data(b)) => return Ok(b.to_vec()),
67                Some(StreamMessage::Ack(seq)) => {
68                    log::debug!(
69                        "PhantomStream {}: received ACK for seq {}",
70                        self.stream_id,
71                        seq
72                    );
73                    // This `recv()` surfaces only application data to the caller.
74                    // Reliable delivery / retransmission (the L1 RTO + SACK
75                    // fast-retransmit path) lives in the data pump and the
76                    // reliable-stream layer (`transport/stream.rs`), not here, so
77                    // a stream-level ACK is informational at this surface: log it
78                    // and keep waiting for the next data frame.
79                    continue;
80                }
81                Some(StreamMessage::Close) => {
82                    return Err(CoreError::NetworkError("Stream closed by peer".into()));
83                }
84                None => {
85                    return Err(CoreError::NetworkError("Stream closed locally".into()));
86                }
87            }
88        }
89    }
90
91    /// Close this stream; the peer will see EOF on its read half.
92    ///
93    /// Named `disconnect` rather than `close` for the same reason as
94    /// `PhantomSession::disconnect` — UniFFI's Kotlin generator emits
95    /// `AutoCloseable.close()` on every object.
96    pub async fn disconnect(&self) -> Result<(), CoreError> {
97        self.tx
98            .send(SessionCommand::CloseStream {
99                stream_id: self.stream_id,
100            })
101            .await
102            .map_err(|_| CoreError::NetworkError("Session closed".into()))
103    }
104}
phantom_protocol/api/stream.rs

phantom_protocol/api/
stream.rs
