hara_native/core/
async_value.rs

1fn portable_type_keyword(value: &Value) -> Result<Keyword, String> {
2    let builtin = match value {
3        Value::Nil => "Nil",
4        Value::Number(_) => "Long",
5        Value::Float(_) => "Float",
6        Value::BigInteger(_) if crate::numeric::is_long_value(value) => "Long",
7        Value::BigInteger(_) => "BigInteger",
8        Value::Character(_) => "Character",
9        Value::Regex(_) => "RegExp",
10        Value::Tagged(value) if is_uuid_tagged(value) => "UUID",
11        Value::Tagged(value) if is_reduced_value(&Value::Tagged(value.clone())) => "Reduced",
12        Value::Tagged(_) => "TaggedLiteral",
13        Value::Bool(_) => "Boolean",
14        Value::String(_) => "String",
15        Value::Keyword(_) => "Keyword",
16        Value::Symbol(_) => "Symbol",
17        Value::Pointer(_) => "Pointer",
18        Value::Function(_) => "Function",
19        Value::Bytes(_) => "Bytes",
20        Value::ByteBuffer(_) => "ByteBuffer",
21        Value::Array(_) => "Array",
22        Value::Object(_) => "Object",
23        Value::Promise(_) => "Promise",
24        Value::Atom(_) => "Atom",
25        Value::Recur(_) => "Recur",
26        Value::List(_) => "List",
27        Value::Cons(_) => "Cons",
28        Value::Queue(_) => "Queue",
29        Value::Deque(_) => "Deque",
30        Value::Tuple(_) => "Vector",
31        Value::Vector(_) => "Vector",
32        Value::MapEntry(_) => "MapEntry",
33        Value::MutableCollection(_) => "MutableCollection",
34        Value::Seq(_) => "Seq",
35        Value::Map(_) => "HashMap",
36        Value::OrderedMap(_) => "OrderedMap",
37        Value::SortedMap(_) => "SortedMap",
38        Value::Trie(_) => "Trie",
39        Value::PriorityMap(_) => "PriorityMap",
40        Value::Set(_) => "HashSet",
41        Value::OrderedSet(_) => "OrderedSet",
42        Value::SortedSet(_) => "SortedSet",
43        Value::Iterator(_) => "Iterator",
44        Value::Var(_) => "Var",
45        Value::Namespace(_) => "Namespace",
46        Value::Extension(_) => "Extension",
47        Value::StructType(_) => "StructType",
48        Value::Struct(value) => return Ok(Keyword::from(value.ty.name.replace('/', "."))),
49        Value::MutableType(_) => "MutableType",
50        Value::Mutable(value) => return Ok(Keyword::from(value.ty.name.replace('/', "."))),
51        Value::Protocol(_) => "Protocol",
52        Value::NativeType(_) => "NativeType",
53        Value::Schema(_) => "SchemaType",
54        Value::Coroutine(_) => "Coroutine",
55        Value::Stream(_) => "Stream",
56        Value::Result(_) => "Result",
57        Value::ExceptionInfo(_) => "Exception",
58    };
59    Ok(Keyword::from(format!("std.native.{builtin}")))
60}
61
62fn native_type_instance(native: &NativeType, value: &Value) -> Result<bool, String> {
63    Ok(portable_type_keyword(value)?.as_str() == native.name)
64}
65
66pub fn receiver_category(value: &Value) -> &'static str {
67    match value {
68        Value::Nil => "nil",
69        Value::Number(_) | Value::Float(_) | Value::BigInteger(_) => "number",
70        Value::Character(_) => "character",
71        Value::Regex(_) => "pattern",
72        Value::Tagged(_) => "tagged",
73        Value::Bool(_) => "boolean",
74        Value::String(_) => "string",
75        Value::Keyword(_) => "keyword",
76        Value::Symbol(_) => "symbol",
77        Value::Pointer(_) => "pointer",
78        Value::Function(_) => "function",
79        Value::Bytes(_) | Value::ByteBuffer(_) => "bytes",
80        Value::Array(_) => "array",
81        Value::Object(_) => "object",
82        Value::Promise(_) => "promise",
83        Value::Atom(_) => "atom",
84        Value::Recur(_) => "recur",
85        Value::List(_) => "list",
86        Value::Cons(_) => "cons",
87        Value::Queue(_) => "queue",
88        Value::Deque(_) => "deque",
89        Value::Tuple(_) => "vector",
90        Value::Vector(_) => "vector",
91        Value::MapEntry(_) => "map-entry",
92        Value::MutableCollection(_) => "mutable",
93        Value::Seq(_) => "seq",
94        Value::Map(_)
95        | Value::OrderedMap(_)
96        | Value::SortedMap(_)
97        | Value::Trie(_)
98        | Value::PriorityMap(_) => "map",
99        Value::Set(_) | Value::OrderedSet(_) | Value::SortedSet(_) => "set",
100        Value::Iterator(_) => "iterator",
101        Value::Var(_) => "var",
102        Value::Namespace(_) => "namespace",
103        Value::Extension(_) => "extension",
104        Value::StructType(_) => "struct-type",
105        Value::Struct(_) => "struct",
106        Value::MutableType(_) => "mutable-type",
107        Value::Mutable(_) => "mutable",
108        Value::Protocol(_) => "protocol",
109        Value::NativeType(_) => "native-type",
110        Value::Schema(_) => "schema",
111        Value::Coroutine(_) => "coroutine",
112        Value::Stream(_) => "stream",
113        Value::Result(_) => "result",
114        Value::ExceptionInfo(_) => "exception",
115    }
116}
117
118fn coroutine_status(coroutine: &Coroutine) -> Value {
119    let state = coroutine.state.borrow();
120    Value::Keyword(Keyword::from(match &*state {
121        CoroutineState::New(_) | CoroutineState::Suspended(_) => "suspended",
122        CoroutineState::Running => "running",
123        CoroutineState::Dead => "dead",
124    }))
125}
126
127fn coroutine_close(coroutine: &Coroutine) -> Result<(), String> {
128    let mut state = coroutine.state.borrow_mut();
129    match &*state {
130        CoroutineState::Dead => Ok(()),
131        CoroutineState::Running => Err("coroutine/close: cannot close a running coroutine".into()),
132        _ => {
133            *state = CoroutineState::Dead;
134            Ok(())
135        }
136    }
137}
138
139fn stream_close(stream: &RuntimeStream) -> Result<(), String> {
140    if stream.closed.replace(true) {
141        return Ok(());
142    }
143    match &stream.source {
144        RuntimeStreamSource::Coroutine { coroutine, .. } => coroutine_close(coroutine),
145        RuntimeStreamSource::Guest { close, .. } => {
146            if let Some(close) = close {
147                call_function(close, Vec::new())?;
148            }
149            Ok(())
150        }
151        RuntimeStreamSource::Host { close, .. } => close(),
152    }
153}
154
155fn stream_next(stream: &RuntimeStream) -> Value {
156    let promise = Promise::new();
157    if stream.closed.get() {
158        promise.resolve(Value::Nil);
159        return Value::Promise(promise);
160    }
161    if stream.pending.replace(true) {
162        promise.reject("stream/pending-pull: only one Stream/next may be pending");
163        return Value::Promise(promise);
164    }
165    match &stream.source {
166        RuntimeStreamSource::Coroutine {
167            coroutine,
168            initial_arguments,
169        } => {
170            let arguments = initial_arguments.borrow_mut().take().unwrap_or_default();
171            let coroutine = coroutine.clone();
172            let state = Rc::new((stream.pending.clone(), stream.closed.clone()));
173            let step = fiber::coroutine::coroutine_resume(
174                coroutine.clone(),
175                arguments,
176                Box::new(Step::Done),
177            );
178            drive_stream_step(step, coroutine, state, promise.clone());
179        }
180        RuntimeStreamSource::Guest { next, .. } => {
181            let source = match call_function(next, Vec::new()) {
182                Ok(value) => promise_from(value),
183                Err(error) => {
184                    stream.pending.set(false);
185                    promise.reject(error);
186                    return Value::Promise(promise);
187                }
188            };
189            let pending = stream.pending.clone();
190            let closed = stream.closed.clone();
191            let output = promise.clone();
192            source.on_settle(Rc::new(move |settled| {
193                pending.set(false);
194                match settled {
195                    PromiseState::Fulfilled(value) => {
196                        if matches!(value, Value::Nil) {
197                            closed.set(true);
198                        }
199                        output.resolve(value);
200                    }
201                    PromiseState::Rejected(error) => {
202                        closed.set(true);
203                        output.reject_rejection(error);
204                    }
205                    PromiseState::Pending => {}
206                };
207            }));
208            let source_poll = source.clone();
209            promise.set_poller(Rc::new(move || {
210                source_poll.state();
211            }));
212            let source_wait = source.clone();
213            promise.set_waiter(Rc::new(move || {
214                source_wait.wait_state();
215            }));
216        }
217        RuntimeStreamSource::Host { next, .. } => match next() {
218            Ok(source) => {
219                let pending = stream.pending.clone();
220                let closed = stream.closed.clone();
221                let output = promise.clone();
222                source.on_settle(Rc::new(move |settled| {
223                    pending.set(false);
224                    match settled {
225                        PromiseState::Fulfilled(value) => {
226                            if matches!(value, Value::Nil) {
227                                closed.set(true);
228                            }
229                            output.resolve(value);
230                        }
231                        PromiseState::Rejected(error) => {
232                            closed.set(true);
233                            output.reject_rejection(error);
234                        }
235                        PromiseState::Pending => {}
236                    };
237                }));
238                let source_poll = source.clone();
239                promise.set_poller(Rc::new(move || {
240                    source_poll.state();
241                }));
242                let source_wait = source.clone();
243                promise.set_waiter(Rc::new(move || {
244                    source_wait.wait_state();
245                }));
246            }
247            Err(error) => {
248                stream.pending.set(false);
249                promise.reject(error);
250            }
251        },
252    }
253    Value::Promise(promise)
254}
255
256pub(crate) fn host_stream(
257    next: Rc<dyn Fn() -> Result<Promise, String>>,
258    close: Rc<dyn Fn() -> Result<(), String>>,
259) -> Value {
260    Value::Stream(Rc::new(RuntimeStream::host(next, close)))
261}
262
263/// Pulls one item from a native Stream without exposing its representation to an embedder.
264pub fn stream_next_value(value: &Value) -> Result<Promise, String> {
265    let Value::Stream(stream) = value else {
266        return Err("stream/next expects a Stream".into());
267    };
268    let Value::Promise(promise) = stream_next(stream) else {
269        unreachable!("native Stream/next always returns a Promise")
270    };
271    Ok(promise)
272}
273
274/// Closes a native Stream owned by the current runtime worker.
275pub fn stream_close_value(value: &Value) -> Result<(), String> {
276    let Value::Stream(stream) = value else {
277        return Err("stream/close expects a Stream".into());
278    };
279    stream_close(stream)
280}
281
282pub fn stream_value(value: &Value) -> bool {
283    matches!(value, Value::Stream(_))
284}
285
286fn drive_stream_step(
287    mut step: Step,
288    coroutine: Rc<Coroutine>,
289    state: Rc<(Rc<Cell<bool>>, Rc<Cell<bool>>)>,
290    output: Promise,
291) {
292    loop {
293        match step {
294            Step::Done(result) => {
295                state.0.set(false);
296                match result {
297                    Ok(_) if matches!(*coroutine.state.borrow(), CoroutineState::Dead) => {
298                        state.1.set(true);
299                        output.resolve(Value::Nil);
300                    }
301                    Ok(Value::Nil) => {
302                        state.1.set(true);
303                        let _ = coroutine_close(&coroutine);
304                        output.reject("stream/nil-item: a stream coroutine may not yield nil");
305                    }
306                    Ok(value) => {
307                        output.resolve(value);
308                    }
309                    Err(error) => {
310                        state.1.set(true);
311                        output.reject(error);
312                    }
313                }
314                return;
315            }
316            Step::Continue(next) => step = next(),
317            Step::Wait(promise, resume) => {
318                let resume = Rc::new(RefCell::new(Some(resume)));
319                let coroutine_next = coroutine.clone();
320                let state_next = state.clone();
321                let output_next = output.clone();
322                promise.on_settle(Rc::new(move |settled| {
323                    if let Some(resume) = resume.borrow_mut().take() {
324                        drive_stream_step(
325                            resume(settled),
326                            coroutine_next.clone(),
327                            state_next.clone(),
328                            output_next.clone(),
329                        );
330                    }
331                }));
332                return;
333            }
334            Step::Yield(_, _) => {
335                state.0.set(false);
336                state.1.set(true);
337                output.reject("stream/internal: yield escaped its coroutine boundary");
338                return;
339            }
340        }
341    }
342}
343
344fn native_stream_values(operation: &str, values: Vec<Value>) -> Result<Value, String> {
345    let method = operation
346        .strip_prefix("std.native.Stream/")
347        .unwrap_or(operation);
348    match method {
349        "create" => {
350            if !(1..=2).contains(&values.len()) {
351                return Err("Stream/create expects next and optional close functions".into());
352            }
353            let Value::Function(next) = &values[0] else {
354                return Err("Stream/create expects a next function".into());
355            };
356            let close = match values.get(1) {
357                None | Some(Value::Nil) => None,
358                Some(Value::Function(close)) => Some(close.clone()),
359                Some(_) => return Err("Stream/create expects a close function or nil".into()),
360            };
361            Ok(Value::Stream(Rc::new(RuntimeStream::guest(next.clone(), close))))
362        }
363        "generate" => {
364            if values.is_empty() {
365                return Err("Stream/generate expects a function".into());
366            }
367            let body = values[0].clone();
368            if !matches!(body, Value::Function(_)) {
369                return Err("Stream/generate expects a function".into());
370            }
371            let arguments = values[1..].to_vec();
372            Ok(Value::Stream(Rc::new(RuntimeStream::new(body, arguments))))
373        }
374        "next" => {
375            if values.len() != 1 {
376                return Err("Stream/next expects one stream".into());
377            }
378            match &values[0] {
379                Value::Stream(stream) => Ok(stream_next(stream)),
380                _ => Err("Stream/next expects a stream".into()),
381            }
382        }
383        _ => Err(format!("unknown std.native.Stream operation: {method}")),
384    }
385}
386
387fn parse_forms(source: &str) -> Result<Vec<Form>, String> {
388    crate::kernel::parse_forms(source)
389}
390
391pub fn read_edn(source: &str) -> Result<Value, String> {
392    let forms = parse_forms(source).map_err(|error| format!("edn/read: {error}"))?;
393    if forms.len() != 1 {
394        return Err("edn/read expects exactly one value".into());
395    }
396    form_to_value(&forms[0]).map_err(|error| format!("edn/read: {error}"))
397}
hara_native/core/async_value.rs

hara_native/core/
async_value.rs
