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, StreamInspectorSnapshot, StreamItem, StreamMedia, StreamPacket, StreamStats,
8    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    stream: Arc<StreamValue>,
20}
21
22impl HostCallbackQueue {
23    /// Wraps a push stream, capturing its media for callback validation.
24    pub fn new(stream: Arc<StreamValue>) -> Self {
25        Self {
26            media: stream.metadata().media(),
27            stream,
28        }
29    }
30
31    /// Returns a shared handle to the underlying stream value.
32    pub fn stream(&self) -> Arc<StreamValue> {
33        Arc::clone(&self.stream)
34    }
35
36    /// Enqueues a packet from a host callback after checking its media.
37    pub fn callback_packet(&self, packet: StreamPacket) -> Result<PushResult> {
38        self.ensure_media(&packet)?;
39        self.stream.push_packet(StreamItem::new(packet))
40    }
41
42    /// Enqueues a prebuilt stream item from a host callback after checking its
43    /// media.
44    pub fn callback_item(&self, item: StreamItem) -> Result<PushResult> {
45        self.ensure_media(item.packet())?;
46        self.stream.push_packet(item)
47    }
48
49    /// Removes up to `limit` buffered items from the stream.
50    pub fn drain(&self, limit: usize) -> Result<Vec<StreamItem>> {
51        self.stream.take_packets(limit)
52    }
53
54    /// Closes the push side of the stream.
55    pub fn close(&self) -> Result<()> {
56        self.stream.close_push()
57    }
58
59    /// Cancels the stream and drops buffered packets.
60    pub fn cancel(&self) -> Result<()> {
61        self.stream.cancel()
62    }
63
64    /// Returns the current stream statistics.
65    pub fn stats(&self) -> Result<StreamStats> {
66        self.stream.stats()
67    }
68
69    /// Builds an inspector snapshot for the stream on the given route and
70    /// transport profile.
71    pub fn inspector(
72        &self,
73        route: Symbol,
74        profile: &TransportProfile,
75        recent_diagnostics: Vec<Symbol>,
76    ) -> Result<StreamInspectorSnapshot> {
77        StreamInspectorSnapshot::from_stream_value(&self.stream, route, profile, recent_diagnostics)
78    }
79
80    fn ensure_media(&self, packet: &StreamPacket) -> Result<()> {
81        let packet_media = match packet {
82            StreamPacket::Pcm(_) => StreamMedia::Pcm,
83            StreamPacket::Midi(_) => StreamMedia::Midi,
84            StreamPacket::Diagnostic(_) => StreamMedia::Diagnostic,
85            StreamPacket::Data(_) => StreamMedia::Data,
86        };
87        if packet_media == self.media {
88            Ok(())
89        } else {
90            Err(Error::TypeMismatch {
91                expected: "host callback packet matching device media",
92                found: "host callback packet for another media",
93            })
94        }
95    }
96}
sim_lib_stream_host/queue.rs

sim_lib_stream_host/
queue.rs
