sim_lib_stream_host/
queue.rs

1//! Cloneable bounded queue handle passed to host callbacks.
2
3use std::sync::Arc;
4
5use sim_kernel::{Error, Result, Symbol};
6use sim_lib_stream_core::{
7    PushResult, StreamDirection, StreamInspectorSnapshot, StreamItem, StreamMedia, StreamPacket,
8    StreamStats, StreamValue, TransportProfile,
9};
10
11/// Cloneable handle host callbacks use to enqueue packets into a stream.
12///
13/// Wraps a push [`StreamValue`] and enforces that every enqueued packet matches
14/// the device media. The handle is cheap to clone and callbacks enqueue through
15/// non-blocking calls.
16#[derive(Clone)]
17pub struct HostCallbackQueue {
18    media: StreamMedia,
19    direction: StreamDirection,
20    stream: Arc<StreamValue>,
21}
22
23impl HostCallbackQueue {
24    /// Wraps a push stream, capturing its media for callback validation.
25    pub fn new(stream: Arc<StreamValue>) -> Self {
26        Self {
27            media: stream.metadata().media(),
28            direction: stream.metadata().direction(),
29            stream,
30        }
31    }
32
33    /// Returns a shared handle to the underlying stream value.
34    pub fn stream(&self) -> Arc<StreamValue> {
35        Arc::clone(&self.stream)
36    }
37
38    /// Enqueues a packet from a host callback after checking its media.
39    pub fn callback_packet(&self, packet: StreamPacket) -> Result<PushResult> {
40        self.ensure_callback_input()?;
41        self.ensure_media(&packet)?;
42        self.stream.push_packet(StreamItem::new(packet))
43    }
44
45    /// Enqueues a prebuilt stream item from a host callback after checking its
46    /// media.
47    pub fn callback_item(&self, item: StreamItem) -> Result<PushResult> {
48        self.ensure_callback_input()?;
49        self.ensure_media(item.packet())?;
50        self.stream.push_packet(item)
51    }
52
53    /// Removes up to `limit` buffered items from the stream.
54    pub fn drain(&self, limit: usize) -> Result<Vec<StreamItem>> {
55        self.stream.take_packets(limit)
56    }
57
58    /// Closes the push side of the stream.
59    pub fn close(&self) -> Result<()> {
60        self.stream.close_push()
61    }
62
63    /// Cancels the stream and drops buffered packets.
64    pub fn cancel(&self) -> Result<()> {
65        self.stream.cancel()
66    }
67
68    /// Returns the current stream statistics.
69    pub fn stats(&self) -> Result<StreamStats> {
70        self.stream.stats()
71    }
72
73    /// Builds an inspector snapshot for the stream on the given route and
74    /// transport profile.
75    pub fn inspector(
76        &self,
77        route: Symbol,
78        profile: &TransportProfile,
79        recent_diagnostics: Vec<Symbol>,
80    ) -> Result<StreamInspectorSnapshot> {
81        StreamInspectorSnapshot::from_stream_value(&self.stream, route, profile, recent_diagnostics)
82    }
83
84    fn ensure_media(&self, packet: &StreamPacket) -> Result<()> {
85        let packet_media = match packet {
86            StreamPacket::Pcm(_) => StreamMedia::Pcm,
87            StreamPacket::Midi(_) => StreamMedia::Midi,
88            StreamPacket::Diagnostic(_) => StreamMedia::Diagnostic,
89            StreamPacket::Data(_) => StreamMedia::Data,
90        };
91        if packet_media == self.media {
92            Ok(())
93        } else {
94            Err(Error::TypeMismatch {
95                expected: "host callback packet matching device media",
96                found: "host callback packet for another media",
97            })
98        }
99    }
100
101    fn ensure_callback_input(&self) -> Result<()> {
102        if self.direction == StreamDirection::Sink {
103            return Err(Error::Eval(
104                "host callbacks cannot push into a sink stream".to_owned(),
105            ));
106        }
107        Ok(())
108    }
109}
sim_lib_stream_host/queue.rs

sim_lib_stream_host/
queue.rs
