sim_lib_stream_prelude/
handle.rs

1use std::sync::{Arc, Mutex};
2
3use sim_citizen_derive::non_citizen;
4use sim_kernel::{
5    CORE_FUNCTION_CLASS_ID, ClassId, ClassRef, Cx, Error, Expr, Object, ObjectCompat, Result,
6    Symbol,
7};
8use sim_lib_stream_audio::{PcmBuffer, PcmSpec};
9use sim_lib_stream_core::{
10    MidiPacket, StreamItem, StreamMetadata, StreamPacket, StreamStats, StreamValue,
11};
12
13#[non_citizen(
14    reason = "live stream handle; reconstruct stream/Metadata and stream/Packet descriptors then open explicitly",
15    kind = "handle",
16    descriptor = "stream/Metadata"
17)]
18/// Live handle to a memory stream source, sink, or pipeline.
19///
20/// A [`StreamHandle`] is a cheap, cloneable reference (an `Arc` inner) to one
21/// of four kinds of memory endpoint: a pull source, a PCM sink, a MIDI sink, or
22/// a pipeline that connects a source to an optional sink. Reads
23/// ([`StreamHandle::next_packet`]), writes ([`StreamHandle::write_packet`]), and
24/// whole-stream runs ([`StreamHandle::run`]) are dispatched on the handle kind
25/// and fail closed when invoked on an endpoint that does not support them.
26#[derive(Clone)]
27pub struct StreamHandle {
28    inner: Arc<HandleInner>,
29}
30
31struct HandleInner {
32    metadata: StreamMetadata,
33    kind: HandleKind,
34}
35
36enum HandleKind {
37    Source {
38        stream: Arc<StreamValue>,
39    },
40    PcmSink {
41        spec: PcmSpec,
42        state: Mutex<SinkState>,
43    },
44    MidiSink {
45        tpq: u16,
46        state: Mutex<SinkState>,
47    },
48    Pipeline {
49        source: StreamHandle,
50        sink: Option<StreamHandle>,
51    },
52}
53
54/// Summary of one [`StreamHandle::run`] over a source or pipeline.
55#[derive(Clone, Debug, Default, PartialEq, Eq)]
56pub struct RunReport {
57    /// Number of packets pulled from the source.
58    pub packets: usize,
59    /// Number of packets written into the sink (zero when there is no sink).
60    pub written: usize,
61}
62
63/// Kind of pipeline stage carried by a [`StageHandle`].
64#[derive(Clone, Debug)]
65pub enum StageKind {
66    /// Pass-through stage that forwards every packet unchanged.
67    Identity,
68}
69
70#[non_citizen(
71    reason = "live stream stage handle; reconstruct stream/StageDescriptor then realize explicitly",
72    kind = "handle",
73    descriptor = "stream/StageDescriptor"
74)]
75/// Handle to a pipeline stage placed between a source and a sink.
76///
77/// In STREAM 6 the only stage is the identity stage (see
78/// [`StageHandle::identity`]); `stream/pipe` accepts identity stages and rejects
79/// any other stage kind.
80#[derive(Clone, Debug)]
81pub struct StageHandle {
82    kind: StageKind,
83}
84
85#[derive(Clone, Debug, Default)]
86struct SinkState {
87    packets: Vec<StreamPacket>,
88    stats: StreamStats,
89    closed: bool,
90}
91
92impl StreamHandle {
93    /// Builds a pull-source handle backed by the given stream value.
94    pub fn source(metadata: StreamMetadata, stream: Arc<StreamValue>) -> Self {
95        Self::new(metadata, HandleKind::Source { stream })
96    }
97
98    /// Builds a PCM sink handle that validates writes against `spec`.
99    pub fn pcm_sink(metadata: StreamMetadata, spec: PcmSpec) -> Self {
100        Self::new(
101            metadata,
102            HandleKind::PcmSink {
103                spec,
104                state: Mutex::new(SinkState::default()),
105            },
106        )
107    }
108
109    /// Builds a MIDI sink handle that validates writes against `tpq`.
110    pub fn midi_sink(metadata: StreamMetadata, tpq: u16) -> Self {
111        Self::new(
112            metadata,
113            HandleKind::MidiSink {
114                tpq,
115                state: Mutex::new(SinkState::default()),
116            },
117        )
118    }
119
120    /// Builds a pipeline handle from a source-like handle and optional sink.
121    ///
122    /// # Errors
123    ///
124    /// Returns an error when `source` is not source-like, or when `sink` is
125    /// present but is not a sink handle.
126    pub fn pipeline(source: StreamHandle, sink: Option<StreamHandle>) -> Result<Self> {
127        if !source.is_source_like() {
128            return Err(Error::Eval(
129                "stream/pipe expects a source stream first".to_owned(),
130            ));
131        }
132        if sink.as_ref().is_some_and(|handle| !handle.is_sink()) {
133            return Err(Error::Eval(
134                "stream/pipe sink position must be a sink handle".to_owned(),
135            ));
136        }
137        Ok(Self::new(
138            source.metadata().clone(),
139            HandleKind::Pipeline { source, sink },
140        ))
141    }
142
143    fn new(metadata: StreamMetadata, kind: HandleKind) -> Self {
144        Self {
145            inner: Arc::new(HandleInner { metadata, kind }),
146        }
147    }
148
149    /// Returns the stream metadata describing this handle.
150    pub fn metadata(&self) -> &StreamMetadata {
151        &self.inner.metadata
152    }
153
154    /// Returns `true` when this handle is a PCM or MIDI sink.
155    pub fn is_sink(&self) -> bool {
156        matches!(
157            self.inner.kind,
158            HandleKind::PcmSink { .. } | HandleKind::MidiSink { .. }
159        )
160    }
161
162    /// Returns `true` when this handle is a pipeline that drives a sink.
163    pub fn is_pipeline_with_sink(&self) -> bool {
164        matches!(&self.inner.kind, HandleKind::Pipeline { sink: Some(_), .. })
165    }
166
167    /// Pulls the next packet from a source or pipeline source.
168    ///
169    /// Returns `Ok(None)` once the stream is exhausted.
170    ///
171    /// # Errors
172    ///
173    /// Returns an error when called on a sink handle, or when the underlying
174    /// source fails to produce the next packet.
175    pub fn next_packet(&self) -> Result<Option<StreamItem>> {
176        match &self.inner.kind {
177            HandleKind::Source { stream } => stream.next_packet(),
178            HandleKind::Pipeline { source, .. } => source.next_packet(),
179            HandleKind::PcmSink { .. } | HandleKind::MidiSink { .. } => Err(Error::Eval(
180                "stream/next! expects a source stream handle".to_owned(),
181            )),
182        }
183    }
184
185    /// Writes one packet into a PCM or MIDI sink.
186    ///
187    /// # Errors
188    ///
189    /// Returns an error when called on a non-sink handle, when the packet media
190    /// does not match the sink, or when the packet fails sink validation.
191    pub fn write_packet(&self, packet: StreamPacket) -> Result<()> {
192        match &self.inner.kind {
193            HandleKind::PcmSink { spec, state } => {
194                let StreamPacket::Pcm(pcm) = &packet else {
195                    return Err(Error::Eval(
196                        "PCM memory sink expects PCM stream packets".to_owned(),
197                    ));
198                };
199                PcmBuffer::from_packet(*spec, pcm)?;
200                write_sink_packet(state, packet)
201            }
202            HandleKind::MidiSink { tpq, state } => {
203                let StreamPacket::Midi(midi) = &packet else {
204                    return Err(Error::Eval(
205                        "MIDI memory sink expects MIDI stream packets".to_owned(),
206                    ));
207                };
208                ensure_midi_tpq(*tpq, midi)?;
209                write_sink_packet(state, packet)
210            }
211            HandleKind::Source { .. } | HandleKind::Pipeline { .. } => Err(Error::Eval(
212                "stream/write! expects a sink stream handle".to_owned(),
213            )),
214        }
215    }
216
217    /// Drains a source or pipeline to completion, returning a [`RunReport`].
218    ///
219    /// For a pipeline with a sink, every pulled packet is written into the sink
220    /// and the sink is closed at the end.
221    ///
222    /// # Errors
223    ///
224    /// Returns an error when called on a bare sink handle, or when reading or
225    /// writing a packet fails.
226    pub fn run(&self) -> Result<RunReport> {
227        match &self.inner.kind {
228            HandleKind::Source { .. } => {
229                let mut report = RunReport::default();
230                while self.next_packet()?.is_some() {
231                    report.packets += 1;
232                }
233                Ok(report)
234            }
235            HandleKind::Pipeline { source, sink } => run_pipeline(source, sink.as_ref()),
236            HandleKind::PcmSink { .. } | HandleKind::MidiSink { .. } => Err(Error::Eval(
237                "stream/run! expects a source or pipeline handle".to_owned(),
238            )),
239        }
240    }
241
242    /// Cancels the stream, marking sources and sinks closed and cancelled.
243    ///
244    /// Cancelling a pipeline cancels its source and sink in turn.
245    ///
246    /// # Errors
247    ///
248    /// Returns an error when a sink lock is poisoned or a source cancel fails.
249    pub fn cancel(&self) -> Result<()> {
250        match &self.inner.kind {
251            HandleKind::Source { stream } => stream.cancel(),
252            HandleKind::Pipeline { source, sink } => {
253                source.cancel()?;
254                if let Some(sink) = sink {
255                    sink.cancel()?;
256                }
257                Ok(())
258            }
259            HandleKind::PcmSink { state, .. } | HandleKind::MidiSink { state, .. } => {
260                let mut state = state
261                    .lock()
262                    .map_err(|_| Error::PoisonedLock("stream sink"))?;
263                state.closed = true;
264                state.stats.closed = true;
265                state.stats.cancelled = true;
266                Ok(())
267            }
268        }
269    }
270
271    /// Returns a snapshot of the stream's runtime statistics.
272    ///
273    /// # Errors
274    ///
275    /// Returns an error when a sink lock is poisoned or a source cannot report
276    /// its statistics.
277    pub fn stats(&self) -> Result<StreamStats> {
278        match &self.inner.kind {
279            HandleKind::Source { stream } => stream.stats(),
280            HandleKind::Pipeline { source, .. } => source.stats(),
281            HandleKind::PcmSink { state, .. } | HandleKind::MidiSink { state, .. } => state
282                .lock()
283                .map_err(|_| Error::PoisonedLock("stream sink"))
284                .map(|state| state.stats.clone()),
285        }
286    }
287
288    /// Returns `true` when the stream has no more work to do.
289    ///
290    /// A pipeline is done only when both its source and its sink are done.
291    ///
292    /// # Errors
293    ///
294    /// Returns an error when a sink lock is poisoned or a source cannot report
295    /// its completion state.
296    pub fn done(&self) -> Result<bool> {
297        match &self.inner.kind {
298            HandleKind::Source { stream } => stream.is_done(),
299            HandleKind::Pipeline { source, sink } => {
300                let source_done = source.done()?;
301                let sink_done = match sink {
302                    Some(sink) => sink.done()?,
303                    None => true,
304                };
305                Ok(source_done && sink_done)
306            }
307            HandleKind::PcmSink { state, .. } | HandleKind::MidiSink { state, .. } => state
308                .lock()
309                .map_err(|_| Error::PoisonedLock("stream sink"))
310                .map(|state| state.closed),
311        }
312    }
313
314    /// Returns a copy of the packets accumulated by a sink handle.
315    ///
316    /// # Errors
317    ///
318    /// Returns an error when called on a non-sink handle or when the sink lock
319    /// is poisoned.
320    pub fn sink_packets(&self) -> Result<Vec<StreamPacket>> {
321        match &self.inner.kind {
322            HandleKind::PcmSink { state, .. } | HandleKind::MidiSink { state, .. } => state
323                .lock()
324                .map_err(|_| Error::PoisonedLock("stream sink"))
325                .map(|state| state.packets.clone()),
326            _ => Err(Error::Eval(
327                "stream/sink-packets expects a sink stream handle".to_owned(),
328            )),
329        }
330    }
331
332    /// Renders this handle as a stream graph expression.
333    ///
334    /// A pipeline renders as a `stream/pipe` call over its source and optional
335    /// sink ids; a bare source or sink renders as its id string.
336    pub fn graph_lisp_expr(&self) -> Expr {
337        match &self.inner.kind {
338            HandleKind::Pipeline { source, sink } => {
339                let mut args = vec![source.graph_lisp_atom()];
340                if let Some(sink) = sink {
341                    args.push(sink.graph_lisp_atom());
342                }
343                Expr::Call {
344                    operator: Box::new(Expr::Symbol(Symbol::qualified("stream", "pipe"))),
345                    args,
346                }
347            }
348            HandleKind::Source { .. }
349            | HandleKind::PcmSink { .. }
350            | HandleKind::MidiSink { .. } => self.graph_lisp_atom(),
351        }
352    }
353
354    fn graph_lisp_atom(&self) -> Expr {
355        Expr::String(self.metadata().id().to_string())
356    }
357
358    fn is_source_like(&self) -> bool {
359        matches!(
360            self.inner.kind,
361            HandleKind::Source { .. } | HandleKind::Pipeline { .. }
362        )
363    }
364
365    fn close_sink(&self) -> Result<()> {
366        match &self.inner.kind {
367            HandleKind::PcmSink { state, .. } | HandleKind::MidiSink { state, .. } => {
368                let mut state = state
369                    .lock()
370                    .map_err(|_| Error::PoisonedLock("stream sink"))?;
371                state.closed = true;
372                state.stats.closed = true;
373                Ok(())
374            }
375            _ => Ok(()),
376        }
377    }
378}
379
380impl StageHandle {
381    /// Builds the identity pass-through stage handle.
382    pub fn identity() -> Self {
383        Self {
384            kind: StageKind::Identity,
385        }
386    }
387
388    /// Returns `true` when this stage is the identity pass-through stage.
389    pub fn is_identity(&self) -> bool {
390        matches!(self.kind, StageKind::Identity)
391    }
392}
393
394impl Object for StreamHandle {
395    fn display(&self, _cx: &mut Cx) -> Result<String> {
396        Ok(format!("#<stream-handle {}>", self.metadata().id()))
397    }
398
399    fn as_any(&self) -> &dyn std::any::Any {
400        self
401    }
402}
403
404impl ObjectCompat for StreamHandle {
405    fn class(&self, cx: &mut Cx) -> Result<ClassRef> {
406        cx.factory()
407            .class_stub(ClassId(0), Symbol::qualified("stream", "Handle"))
408    }
409}
410
411impl Object for StageHandle {
412    fn display(&self, _cx: &mut Cx) -> Result<String> {
413        Ok("#<stream-stage identity>".to_owned())
414    }
415
416    fn as_any(&self) -> &dyn std::any::Any {
417        self
418    }
419}
420
421impl ObjectCompat for StageHandle {
422    fn class(&self, cx: &mut Cx) -> Result<ClassRef> {
423        cx.factory()
424            .class_stub(CORE_FUNCTION_CLASS_ID, Symbol::qualified("stream", "Stage"))
425    }
426}
427
428fn run_pipeline(source: &StreamHandle, sink: Option<&StreamHandle>) -> Result<RunReport> {
429    let mut report = RunReport::default();
430    while let Some(item) = source.next_packet()? {
431        report.packets += 1;
432        if let Some(sink) = sink {
433            sink.write_packet(item.packet().clone())?;
434            report.written += 1;
435        }
436    }
437    if let Some(sink) = sink {
438        sink.close_sink()?;
439    }
440    Ok(report)
441}
442
443fn write_sink_packet(state: &Mutex<SinkState>, packet: StreamPacket) -> Result<()> {
444    let mut state = state
445        .lock()
446        .map_err(|_| Error::PoisonedLock("stream sink"))?;
447    if state.closed {
448        return Err(Error::Eval(
449            "cannot write to a closed stream sink".to_owned(),
450        ));
451    }
452    state.stats.pushed += 1;
453    state.stats.accepted += 1;
454    state.packets.push(packet);
455    Ok(())
456}
457
458fn ensure_midi_tpq(expected: u16, packet: &MidiPacket) -> Result<()> {
459    if packet.tpq() != expected {
460        return Err(Error::Eval(format!(
461            "MIDI packet TPQ {} does not match sink TPQ {expected}",
462            packet.tpq()
463        )));
464    }
465    Ok(())
466}
sim_lib_stream_prelude/handle.rs

sim_lib_stream_prelude/
handle.rs
