fidius_guest/
stream_marker.rs

1// Copyright 2026 Colliery, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! The `fidius::Stream<T>` server-streaming return marker (FIDIUS-I-0026, D4).
16
17use core::marker::PhantomData;
18
19/// Marker type a plugin interface uses to declare a **server-streaming** method:
20///
21/// ```ignore
22/// #[fidius::plugin_interface(version = 1, buffer = PluginAllocated)]
23/// pub trait Source: Send + Sync {
24///     fn read(&self, config: String) -> fidius::Stream<Row>;
25/// }
26/// ```
27///
28/// `#[plugin_interface]` recognises a return type whose final path segment is
29/// `Stream<T>` and, for that method:
30///
31/// 1. folds a `!stream` marker into the interface hash (so a streaming method
32///    can never be confused with a unary `-> T` method of the same name/args —
33///    a producer/consumer mismatch is rejected at load), and
34/// 2. (Phase 1+) generates a host-side client method returning the runtime
35///    pull handle (`fidius_host::ChunkStream`).
36///
37/// The marker carries no data — the *runtime* representation of a stream is the
38/// host-side `ChunkStream`, not this type. It exists so the interface trait
39/// type-checks and so the macro has an explicit, unambiguous thing to detect
40/// (rather than guessing from `impl Stream`). `T` is the per-item type and
41/// follows the same wire/`WitType` rules as a unary return.
42///
43/// Argument-position `Stream<T>` (client-streaming / bidirectional) is rejected
44/// in v1.
45///
46/// ## Two forms (FIDIUS-I-0026)
47///
48/// - **Marker form** ([`Stream::new`]): no items. Used purely to *declare* a
49///   streaming method in an interface trait — the Python path (Phase 1) never
50///   iterates it in Rust (its generator is bridged to a `ChunkStream`
51///   host-side).
52/// - **Iterator-backed form** ([`Stream::from_iter`]): a Rust **WASM** guest
53///   (Phase 2) returns real items. The macro-generated component resource pulls
54///   them one at a time via [`Stream::next_item`] to satisfy the WIT contract:
55///   ```wit
56///   resource <m>-stream { next: func() -> result<option<T>, plugin-error>; }
57///   <m>: func(args) -> own<<m>-stream>;
58///   ```
59///   `some(item)` = item, `none` = clean end, `err` = mid-stream error;
60///   dropping the resource runs the guest dtor = cancel (design decision D3).
61pub struct Stream<T> {
62    /// `None` = the pure marker form; `Some` = an iterator-backed producer.
63    iter: Option<Box<dyn Iterator<Item = T> + Send>>,
64    _marker: PhantomData<fn() -> T>,
65}
66
67impl<T> Stream<T> {
68    /// The marker form — declares a streaming method without producing items.
69    /// The interface/Python path uses this; it is never iterated in Rust.
70    pub fn new() -> Self {
71        Self {
72            iter: None,
73            _marker: PhantomData,
74        }
75    }
76
77    /// Build a stream from any iterator — how a Rust WASM guest produces its
78    /// items. The iterator must be `Send + 'static` so the host can drive the
79    /// component resource across its pump thread.
80    #[allow(clippy::should_implement_trait)] // inherent ctor; can't add Send+'static via FromIterator
81    pub fn from_iter<I>(items: I) -> Self
82    where
83        I: IntoIterator<Item = T>,
84        I::IntoIter: Send + 'static,
85    {
86        Self {
87            iter: Some(Box::new(items.into_iter())),
88            _marker: PhantomData,
89        }
90    }
91
92    /// Advance the underlying iterator by one item. Driven by the
93    /// macro-generated component resource's `next()` (WS.2). Returns `None` for
94    /// the marker form and at end of iteration.
95    pub fn next_item(&mut self) -> Option<T> {
96        self.iter.as_mut().and_then(|it| it.next())
97    }
98}
99
100impl<T> Default for Stream<T> {
101    fn default() -> Self {
102        Self::new()
103    }
104}
105
106#[cfg(test)]
107mod tests {
108    use super::*;
109
110    #[test]
111    fn from_iter_yields_then_none() {
112        let mut s = Stream::from_iter(vec![1u64, 2, 3]);
113        assert_eq!(s.next_item(), Some(1));
114        assert_eq!(s.next_item(), Some(2));
115        assert_eq!(s.next_item(), Some(3));
116        assert_eq!(s.next_item(), None);
117        assert_eq!(s.next_item(), None);
118    }
119
120    #[test]
121    fn from_iter_accepts_a_range() {
122        let s = Stream::from_iter(0u64..3);
123        let got: Vec<u64> = collect(s);
124        assert_eq!(got, vec![0, 1, 2]);
125    }
126
127    #[test]
128    fn marker_form_is_empty() {
129        let mut s: Stream<u64> = Stream::new();
130        assert_eq!(s.next_item(), None);
131        let mut d: Stream<u64> = Stream::default();
132        assert_eq!(d.next_item(), None);
133    }
134
135    fn collect<T>(mut s: Stream<T>) -> Vec<T> {
136        let mut out = Vec::new();
137        while let Some(x) = s.next_item() {
138            out.push(x);
139        }
140        out
141    }
142}
fidius_guest/stream_marker.rs

fidius_guest/
stream_marker.rs
