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

hara_native/core/
async_value.rs
