harn-vm 0.10.169

Async bytecode virtual machine for the Harn programming language
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

use crate::stdlib::macros::harn_builtin;
use crate::value::{VmError, VmValue};
use crate::vm::Vm;

pub(crate) fn register_channel_builtins(vm: &mut Vm) {
    crate::stdlib::macros::register_module_builtins(vm, module_path!());
}

#[harn_builtin(
    exposure = "runtime_internal",
    effects = [],
    sig = "emit_channel(name: string, payload: any, options?: dict) -> dict",
    kind = "async",
    category = "channels"
)]
async fn emit_channel_impl(
    ctx: crate::vm::AsyncBuiltinCtx,
    args: Vec<VmValue>,
) -> Result<VmValue, VmError> {
    crate::channels::emit_channel_from_vm(Some(&ctx), args).await
}

#[harn_builtin(
    exposure = "runtime_internal",
    effects = [],
    sig = "channel_events(name: string, options?: dict) -> list",
    kind = "async",
    category = "channels"
)]
async fn channel_events_impl(
    ctx: crate::vm::AsyncBuiltinCtx,
    args: Vec<VmValue>,
) -> Result<VmValue, VmError> {
    crate::channels::channel_events_from_vm(Some(&ctx), args).await
}

#[harn_builtin(
    exposure = "runtime_internal",
    effects = [],
    sig = "channel_subscribe(name: string, options?: dict) -> stream",
    kind = "async",
    category = "channels"
)]
async fn channel_subscribe_impl(
    ctx: crate::vm::AsyncBuiltinCtx,
    args: Vec<VmValue>,
) -> Result<VmValue, VmError> {
    crate::channels::channel_subscribe_from_vm(Some(&ctx), args).await
}

#[harn_builtin(
    exposure = "runtime_internal",
    effects = [],
    sig = "channel_consumer_cursor(name: string, consumer_id: string, options?: dict) -> int?",
    kind = "async",
    category = "channels"
)]
async fn channel_consumer_cursor_impl(
    ctx: crate::vm::AsyncBuiltinCtx,
    args: Vec<VmValue>,
) -> Result<VmValue, VmError> {
    crate::channels::channel_consumer_cursor_from_vm(Some(&ctx), args).await
}

#[harn_builtin(
    exposure = "runtime_internal",
    effects = [],
    sig = "channel_ack(name: string, consumer_id: string, cursor: int, options?: dict) -> dict",
    kind = "async",
    category = "channels"
)]
async fn channel_ack_impl(
    ctx: crate::vm::AsyncBuiltinCtx,
    args: Vec<VmValue>,
) -> Result<VmValue, VmError> {
    crate::channels::channel_ack_from_vm(Some(&ctx), args).await
}

#[harn_builtin(
    exposure = "runtime_internal",
    effects = [],
    sig = "flush_trigger_aggregations() -> nil",
    kind = "async",
    category = "channels"
)]
async fn flush_trigger_aggregations_impl(
    ctx: crate::vm::AsyncBuiltinCtx,
    _args: Vec<VmValue>,
) -> Result<VmValue, VmError> {
    // CH-04 (#1875): explicit flush so tests can deterministically exercise
    // window-expire semantics together with `advance_time(ms)`.
    crate::channels::flush_expired_aggregations_inner(Some(&ctx)).await;
    Ok(VmValue::Nil)
}