machine_stream.bend
bendvy-ecs-core@0.1.0.0 · 99 lines · 0 definitions
1import Base
2import machine.bend as M
3import event_runtime.bend as Ev
4import machine_stream_queue.bend as Q
5
6# FIFO metadata: front oldest-first, back newest-first; size counts events.
7type Stream<V: Data> is Data:
8 Stream{queue: Q.Queue<M.Transition<V>>,positions: List<&2,Ev.Position>,frameStart: Nat}
9type Reader<S: Data,M: Data> is Type:
10 Reader{namespace: U32,id: U32}
11
12def empty(~V: Data) -> Stream<V>: Stream{Q.empty(~M.Transition<V>),[],0n}
13def publish(~V: Data,stream: Stream<V>,tick: Nat,event: M.Transition<V>) -> Stream<V>:
14 match stream:
15 case Stream{queue,positions,frameStart}: Stream{Q.publish(~M.Transition<V>,queue,tick,[event]),positions,frameStart}
16def activate(~V: Data,stream: Stream<V>,id: U32,tick: Nat) -> Stream<V>:
17 match stream:
18 case Stream{queue,positions,frameStart}: Stream{queue,Ev.activate(positions,id,tick),frameStart}
19def complete(~V: Data,stream: Stream<V>,id: U32,tick: Nat,success: Bool) -> Stream<V>:
20 match stream success:
21 case Stream{queue,positions,frameStart} True{}: Stream{queue,Ev.update(positions,id,tick,False{}),frameStart}
22 case stream False{}: stream
23# A condition-false system never executes its reader parameter.
24def skip(~V: Data,stream: Stream<V>,id: U32,tick: Nat) -> Stream<V>:
25 stream
26
27# Bevy insert_state replaces messages while preserving reader positions and loss.
28def initial_queue(~V:Data,queue:Q.Queue<M.Transition<V>>,tick:Nat,value:V) -> Q.Queue<M.Transition<V>>:
29 match queue:
30 case Q.Queue{_,_,_,dropped,_}:Q.Queue{[],[Q.Batch{tick,1n,[M.Initial{value}]}],1n,dropped,1n}
31def initial(~V:Data,stream:Stream<V>,tick:Nat,value:V) -> Stream<V>:
32 match stream:
33 case Stream{queue,positions,frameStart}:Stream{initial_queue(~V,queue,tick,value),positions,frameStart}
34
35def observe_position(~V: Data,position: Maybe<Ev.Position>,batches: List<&2,Ev.Batch<M.Transition<V>>>,dropped: Nat,id: U32,tick: Nat) -> Ev.Reading<M.Transition<V>>:
36 match position:
37 case None{}: Ev.reading(~M.Transition<V>,Ev.Position{id,0n,tick},batches,dropped)
38 case Some{position}: Ev.reading(~M.Transition<V>,position,batches,dropped)
39
40def observe_registered(~V: Data,position: Ev.Position,+queue: Q.Queue<M.Transition<V>>) -> Ev.Reading<M.Transition<V>>:
41 match position queue:
42 case Ev.Position{_,+cursor,registered} Q.Queue{+front,+back,+size,+dropped,+batches}: Ev.Reading{Q.read(~M.Transition<V>,Q.Queue{front,back,size,dropped,batches},cursor),Nat.is_lt(Nat.max(cursor,registered),dropped)}
43def observe_queue_position(~V: Data,position: Maybe<Ev.Position>,queue: Q.Queue<M.Transition<V>>,id: U32,tick: Nat) -> Ev.Reading<M.Transition<V>>:
44 match position:
45 case None{}: observe_registered(~V,Ev.Position{id,0n,tick},queue)
46 case Some{position}: observe_registered(~V,position,queue)
47def observe(~V: Data,stream: Stream<V>,+id: U32,tick: Nat) -> Ev.Reading<M.Transition<V>>:
48 match stream:
49 case Stream{queue,positions,_}: observe_queue_position(~V,Ev.lookup(positions,id),queue,id,tick)
50def batches_into_queue(~V:Data,batches:List<&2,Ev.Batch<M.Transition<V>>>,queue:Q.Queue<M.Transition<V>>) -> Q.Queue<M.Transition<V>>:
51 match batches:
52 case []:queue
53 case Ev.Batch{tick,+items} <> rest:batches_into_queue(~V,rest,Q.publish_nonempty(~M.Transition<V>,queue,tick,items,Q.count_loop(~M.Transition<V>,items,0n)))
54def trimmed(~V:Data,frameStart:Nat,positions:List<&2,Ev.Position>,result:Ev.Trimmed<M.Transition<V>>) -> Stream<V>:
55 match result:
56 case Ev.Trimmed{batches,dropped}:Stream{batches_into_queue(~V,batches,Q.Queue{[],[],0n,dropped,0n}),positions,frameStart}
57
58def frame(~V: Data,stream: Stream<V>,tick: Nat) -> Stream<V>:
59 match stream:
60 case Stream{queue,+positions,frameStart}: Stream{Q.trim(~M.Transition<V>,queue,Ev.hold_boundary(positions,frameStart),65536n),positions,tick}
61def dispose(~V: Data,stream: Stream<V>,id: U32) -> Stream<V>:
62 match stream:
63 case Stream{queue,positions,frameStart}: Stream{queue,Ev.update(positions,id,0n,True{}),frameStart}
64
65# Canonical snapshot includes every retained batch/payload, without changing cursor.
66def batches_loop(~V: Data,batches: List<&2,Q.Batch<M.Transition<V>>>,acc: List<&2,Ev.Batch<M.Transition<V>>>) -> List<&2,Ev.Batch<M.Transition<V>>>:
67 match batches:
68 case []: List.reverse(&2,Ev.Batch<M.Transition<V>>,acc)
69 case Q.Batch{tick,_,items} <> rest: batches_loop(~V,rest,Ev.Batch{tick,items} <> acc)
70def snapshot_queue(~V: Data,queue: Q.Queue<M.Transition<V>>,positions: List<&2,Ev.Position>,frameStart: Nat) -> List<&2,Ev.Batch<M.Transition<V>>> & (List<&2,Ev.Position> & (Nat & Nat)):
71 match queue:
72 case Q.Queue{front,back,_,dropped,_}: (batches_loop(~V,Ev.append(~Q.Batch<M.Transition<V>>,front,List.reverse(&2,Q.Batch<M.Transition<V>>,back)),[]),(positions,(dropped,frameStart)))
73def snapshot(~V: Data,stream: Stream<V>) -> List<&2,Ev.Batch<M.Transition<V>>> & (List<&2,Ev.Position> & (Nat & Nat)):
74 match stream:
75 case Stream{queue,positions,frameStart}: snapshot_queue(~V,queue,positions,frameStart)
76
77# Engine entry points consume and return the real affine reader capability.
78# A foreign-world refusal preserves both owners and never activates a cursor.
79type OpenResult<S: Data,M: Data,V: Data> is Type:
80 Opened{reader: Reader<S,M>,stream: Stream<V>,reading: Ev.Reading<M.Transition<V>>}
81 ReaderRefused{reader: Reader<S,M>,stream: Stream<V>}
82
83def open_checked(~S: Data,~M: Data,~V: Data,allowed: Bool,reader: Reader<S,M>,+stream: Stream<V>,+tick: Nat) -> OpenResult<S,M,V>:
84 match allowed reader:
85 case False{}, reader: ReaderRefused{reader,stream}
86 case True{}, Reader{namespace,+id}:
87 Opened{Reader{namespace,id},activate(~V,stream,id,tick),observe(~V,stream,id,tick)}
88
89def open(~S: Data,~M: Data,~V: Data,reader: Reader<S,M>,stream: Stream<V>,namespace: U32,tick: Nat) -> OpenResult<S,M,V>:
90 match reader:
91 case Reader{+owner,id}: open_checked(~S,~M,~V,U32.is_eq(owner,namespace),Reader{owner,id},stream,tick)
92
93def finish(~S: Data,~M: Data,~V: Data,reader: Reader<S,M>,stream: Stream<V>,tick: Nat,success: Bool) -> Reader<S,M> & Stream<V>:
94 match reader:
95 case Reader{namespace,+id}: (Reader{namespace,id},complete(~V,stream,id,tick,success))
96
97def skip_owned(~S: Data,~M: Data,~V: Data,reader: Reader<S,M>,stream: Stream<V>,tick: Nat) -> Reader<S,M> & Stream<V>:
98 match reader:
99 case Reader{namespace,+id}: (Reader{namespace,id},skip(~V,stream,id,tick))