reader_domains.bend
bendvy-ecs-core@0.1.0.1 · 78 lines · 0 definitions
1import Base
2import world.bend as W
3import event_runtime.bend as Ev
4
5# Affine transport owner for two independently retained Data streams. Gameplay
6# does not receive this owner. The closed classifier belongs to schema setup.
7type Events<S: Data,E: Data> is Type:
8 Events{namespace: U32,batches: List<&2,Ev.Batch<E>>,positions: List<&2,Ev.Position>,nextReader: U32,tick: Nat,frameStart: Nat,boundary: Nat,capacity: Nat,dropped: Nat,values: List<&2,E>}
9type Detached<S: Data,C: Type,R: Type,E: Data> is Type:
10 Detached{world: W.World<S,C,R,E>,events: Events<S,E>}
11def choose(~E: Data,selected: Bool,+value: E,+events: List<&2,E>,+removals: List<&2,E>) -> List<&2,E> & List<&2,E>:
12 match selected:
13 case True{}: (value <> events,removals)
14 case False{}: (events,value <> removals)
15def split_reverse(~E: Data,pair: List<&2,E> & List<&2,E>) -> List<&2,E> & List<&2,E>:
16 (events,removals) = pair
17 (List.reverse(&2,E,events),List.reverse(&2,E,removals))
18def split_loop(~E: Data,~is_event: E -> Bool,values: List<&2,E>,acc: List<&2,E> & List<&2,E>) -> List<&2,E> & List<&2,E>:
19 match values:
20 case []: split_reverse(~E,acc)
21 case +value <> rest:
22 (events,removals) = acc
23 split_loop(~E,~is_event,rest,choose(~E,is_event(value),value,events,removals))
24def split(~E: Data,~is_event: E -> Bool,values: List<&2,E>) -> List<&2,E> & List<&2,E>:
25 split_loop(~E,~is_event,values,([],[]))
26def detached_values(~S: Data,~C: Type,~R: Type,~E: Data,+namespace: U32,+batches: List<&2,Ev.Batch<E>>,+positions: List<&2,Ev.Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat,+removals: List<&2,E>,observed: W.World<S,C,R,E> & List<&2,E>) -> Detached<S,C,R,E>:
27 (world,+values) = observed
28 Detached{W.events_replace(~S,~C,~R,~E,world,removals),Events{namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped,values}}
29def detach(~S: Data,~C: Type,~R: Type,~E: Data,runtime: Ev.Runtime<S,C,R,E>,+removals: List<&2,E>) -> Detached<S,C,R,E>:
30 match runtime:
31 case Ev.Runtime{world,+ns,+batches,+positions,+next,+tick,+start,+boundary,+capacity,+dropped}: detached_values(~S,~C,~R,~E,ns,batches,positions,next,tick,start,boundary,capacity,dropped,removals,W.events_view(~S,~C,~R,~E,world))
32def attach(~S: Data,~C: Type,~R: Type,~E: Data,events: Events<S,E>,world: W.World<S,C,R,E>) -> Ev.Runtime<S,C,R,E>:
33 match events:
34 case Events{ns,batches,positions,next,tick,start,boundary,capacity,dropped,+values}: Ev.Runtime{W.events_replace(~S,~C,~R,~E,world,values),ns,batches,positions,next,tick,start,boundary,capacity,dropped}
35# Absorb newly committed ordinary events from a removal-domain World through
36# the real Event runtime publication synchronization. Deferred barriers use
37# this same operation; rollback leaves no publication delta to transport.
38def collected(~S: Data,~C: Type,~R: Type,~E: Data,world: W.World<S,C,R,E>,events: Events<S,E>,partition: List<&2,E> & List<&2,E>) -> Detached<S,C,R,E>:
39 match events partition:
40 case Events{+ns,+batches,+positions,+next,+tick,+start,+boundary,+capacity,+dropped,+old}, (+fresh,+removals):
41 +all = Ev.append(~E,old,fresh)
42 detach(~S,~C,~R,~E,Ev.sync_values(~S,~C,~R,~E,(W.events_replace(~S,~C,~R,~E,world,all),all),ns,batches,positions,next,1n+tick,start,boundary,capacity,dropped),removals)
43def collect_view(~S: Data,~C: Type,~R: Type,~E: Data,~is_event: E -> Bool,events: Events<S,E>,observed: W.World<S,C,R,E> & List<&2,E>) -> Detached<S,C,R,E>:
44 (world,+values) = observed
45 collected(~S,~C,~R,~E,world,events,split(~E,~is_event,values))
46def collect(~S: Data,~C: Type,~R: Type,~E: Data,~is_event: E -> Bool,events: Events<S,E>,world: W.World<S,C,R,E>) -> Detached<S,C,R,E>:
47 collect_view(~S,~C,~R,~E,~is_event,events,W.events_view(~S,~C,~R,~E,world))
48
49# Reusable pre-dispatch seam. The setup callback receives the actual separated
50# event metadata and complete World owners; no affine owner is copied.
51def driven(~S: Data,~C: Type,~R: Type,~E: Data,~H: Type,~Out: Type,~body: Events<S,E> -> H -> W.World<S,C,R,E> -> Out,owners: H,result: Detached<S,C,R,E>) -> Out:
52 match result:
53 case Detached{world,events}: body(events,owners,world)
54def drive(~S: Data,~C: Type,~R: Type,~E: Data,~H: Type,~Out: Type,~is_event: E -> Bool,~body: Events<S,E> -> H -> W.World<S,C,R,E> -> Out,events: Events<S,E>,owners: H,world: W.World<S,C,R,E>) -> Out:
55 driven(~S,~C,~R,~E,~H,~Out,~body,owners,collect(~S,~C,~R,~E,~is_event,events,world))
56
57# Event cursors are publication ticks, not list offsets. Removing notice-only
58# batches therefore retains every real cursor/tick and the original dropped
59# boundary. Ev.run does not trim; retention is applied only by Ev.frame.
60def event_batch(~E: Data,~is_event: E -> Bool,+tick: Nat,filtered: List<&2,E> & List<&2,E>,rest: List<&2,Ev.Batch<E>>) -> List<&2,Ev.Batch<E>>:
61 match filtered:
62 case ([],_): rest
63 case (items,_): Ev.Batch{tick,items} <> rest
64def event_batches_loop(~E: Data,~is_event: E -> Bool,batches: List<&2,Ev.Batch<E>>,acc: List<&2,Ev.Batch<E>>) -> List<&2,Ev.Batch<E>>:
65 match batches:
66 case []: List.reverse(&2,Ev.Batch<E>,acc)
67 case Ev.Batch{tick,items} <> rest: event_batches_loop(~E,~is_event,rest,event_batch(~E,~is_event,tick,split(~E,~is_event,items),acc))
68def event_batches(~E: Data,~is_event: E -> Bool,batches: List<&2,Ev.Batch<E>>) -> List<&2,Ev.Batch<E>>:
69 event_batches_loop(~E,~is_event,batches,[])
70def recovered(~S: Data,~C: Type,~R: Type,~E: Data,~is_event: E -> Bool,world: W.World<S,C,R,E>,+namespace: U32,+batches: List<&2,Ev.Batch<E>>,+positions: List<&2,Ev.Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat,+old_removals: List<&2,E>,partition: List<&2,E> & List<&2,E>) -> Detached<S,C,R,E>:
71 match partition:
72 case (+values,+fresh_removals): Detached{W.events_replace(~S,~C,~R,~E,world,Ev.append(~E,old_removals,fresh_removals)),Events{namespace,event_batches(~E,~is_event,batches),positions,nextReader,tick,frameStart,boundary,capacity,dropped,values}}
73def recovery_view(~S: Data,~C: Type,~R: Type,~E: Data,~is_event: E -> Bool,+namespace: U32,+batches: List<&2,Ev.Batch<E>>,+positions: List<&2,Ev.Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat,+old_removals: List<&2,E>,observed: W.World<S,C,R,E> & List<&2,E>) -> Detached<S,C,R,E>:
74 (world,values) = observed
75 recovered(~S,~C,~R,~E,~is_event,world,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped,old_removals,split(~E,~is_event,values))
76def recover(~S: Data,~C: Type,~R: Type,~E: Data,~is_event: E -> Bool,runtime: Ev.Runtime<S,C,R,E>,+old_removals: List<&2,E>) -> Detached<S,C,R,E>:
77 match runtime:
78 case Ev.Runtime{world,+namespace,+batches,+positions,+nextReader,+tick,+frameStart,+boundary,+capacity,+dropped}: recovery_view(~S,~C,~R,~E,~is_event,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped,old_removals,W.events_view(~S,~C,~R,~E,world))