event_publication.bend
bendvy-ecs-core@0.1.0.0 · 108 lines · 0 definitions
1import Base
2import world.bend as W
3import event_runtime.bend as Ev
4import transaction.bend as Tx
5import ordinary_event_declaration.bend as Event
6import event_stream.bend as Leaf
7
8# Each ordinary emit stages one singleton product record. Tx.finish preserves
9# the record list's emission order; products are never merged across emissions.
10type Publication<A:Data,B:Data> is Data:
11 Publication{first:List<&2,A>,second:List<&2,B>}
12type Streams<S:Data,A:Data,B:Data,H:Type> is Type:
13 Streams{first:Leaf.Stream<S,A>,second:Leaf.Stream<S,B>,owners:H}
14type Application<S:Data,C:Type,R:Type,A:Data,B:Data,H:Type> is Type:
15 Application{world:W.World<S,C,R,Publication<A,B>>,streams:Streams<S,A,B,H>,journal:List<&2,Ev.Batch<Publication<A,B>>>,tick:Nat,frameStart:Nat}
16def setup(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,world:W.World<S,C,R,Publication<A,B>>,first:Event.Domain<S,A>,second:Event.Domain<S,B>,owners:H,+capacity:Nat) -> Application<S,C,R,A,B,H>:
17 Application{world,Streams{Leaf.provision(~S,~A,first,capacity),Leaf.provision(~S,~B,second,capacity),owners},[],0n,0n}
18def first_take(~S:Data,~A:Data,~B:Data,~H:Type,streams:Streams<S,A,B,H>) -> Streams<S,A,B,H> & Leaf.Stream<S,A>:
19 match streams:
20 case Streams{+first,second,owners}:(Streams{first,second,owners},first)
21def first_put(~S:Data,~A:Data,~B:Data,~H:Type,streams:Streams<S,A,B,H>,first:Leaf.Stream<S,A>) -> Streams<S,A,B,H>:
22 match streams:
23 case Streams{_,second,owners}:Streams{first,second,owners}
24def second_take(~S:Data,~A:Data,~B:Data,~H:Type,streams:Streams<S,A,B,H>) -> Streams<S,A,B,H> & Leaf.Stream<S,B>:
25 match streams:
26 case Streams{first,+second,owners}:(Streams{first,second,owners},second)
27def second_put(~S:Data,~A:Data,~B:Data,~H:Type,streams:Streams<S,A,B,H>,second:Leaf.Stream<S,B>) -> Streams<S,A,B,H>:
28 match streams:
29 case Streams{first,_,owners}:Streams{first,second,owners}
30def emit_first(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,tx:Tx.Tx<W.World<S,C,R,Publication<A,B>>,Publication<A,B>>,event:A) -> Tx.Tx<W.World<S,C,R,Publication<A,B>>,Publication<A,B>>:
31 Tx.emit(~W.World<S,C,R,Publication<A,B>>,~Publication<A,B>,tx,Publication{[event],[]})
32def emit_second(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,tx:Tx.Tx<W.World<S,C,R,Publication<A,B>>,Publication<A,B>>,event:B) -> Tx.Tx<W.World<S,C,R,Publication<A,B>>,Publication<A,B>>:
33 Tx.emit(~W.World<S,C,R,Publication<A,B>>,~Publication<A,B>,tx,Publication{[],[event]})
34# Reverse-accumulate each record's values, then reverse once. Both traversals
35# decrease before the accumulator and preserve order within/across records.
36def prepend_values(~E:Data,values:List<&2,E>,reversed:List<&2,E>) -> List<&2,E>:
37 match values:
38 case []:reversed
39 case head <> rest:prepend_values(~E,rest,head <> reversed)
40def first_values_loop(~A:Data,~B:Data,items:List<&2,Publication<A,B>>,reversed:List<&2,A>) -> List<&2,A>:
41 match items:
42 case []:List.reverse(&2,A,reversed)
43 case Publication{first,_} <> rest:first_values_loop(~A,~B,rest,prepend_values(~A,first,reversed))
44def first_values(~A:Data,~B:Data,items:List<&2,Publication<A,B>>) -> List<&2,A>:first_values_loop(~A,~B,items,[])
45def second_values_loop(~A:Data,~B:Data,items:List<&2,Publication<A,B>>,reversed:List<&2,B>) -> List<&2,B>:
46 match items:
47 case []:List.reverse(&2,B,reversed)
48 case Publication{_,second} <> rest:second_values_loop(~A,~B,rest,prepend_values(~B,second,reversed))
49def second_values(~A:Data,~B:Data,items:List<&2,Publication<A,B>>) -> List<&2,B>:second_values_loop(~A,~B,items,[])
50def synced(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,observed:W.World<S,C,R,Publication<A,B>> & List<&2,Publication<A,B>>,streams:Streams<S,A,B,H>,+journal:List<&2,Ev.Batch<Publication<A,B>>>,+tick:Nat,frameStart:Nat) -> Application<S,C,R,A,B,H>:
51 match observed streams:
52 case (world,items) Streams{first,second,owners}:
53 +published=List.drop(&2,Publication<A,B>,items,Ev.size(~Publication<A,B>,journal))
54 Application{world,Streams{Leaf.append(~S,~A,first,tick,first_values(~A,~B,published)),Leaf.append(~S,~B,second,tick,second_values(~A,~B,published)),owners},Ev.sync_checked(~Publication<A,B>,Nat.is_lt(0n,Ev.count(~Publication<A,B>,published)),journal,tick,published),tick,frameStart}
55def sync(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,app:Application<S,C,R,A,B,H>) -> Application<S,C,R,A,B,H>:
56 match app:
57 case Application{world,streams,journal,+tick,frameStart}:synced(~S,~C,~R,~A,~B,~H,W.events_view(~S,~C,~R,~Publication<A,B>,world),streams,journal,1n+tick,frameStart)
58
59# Arbitrary finite products can use this same closed selection operation. The
60# generated take/put retain every other leaf and arbitrary affine owner H.
61def observed_read(~S:Data,~E:Data,~F:Type,~put:F -> Leaf.Stream<S,E> -> F,family:F,returned:Leaf.Stream<S,E> & Ev.Reading<E>) -> F & Ev.Reading<E>:
62 match returned:
63 case (stream,reading):(put(family,stream),reading)
64def observed_leaf(~S:Data,~E:Data,~F:Type,~put:F -> Leaf.Stream<S,E> -> F,returned:F & Leaf.Stream<S,E>,+tick:Nat,id:U32) -> F & Ev.Reading<E>:
65 match returned:
66 case (family,+stream):observed_read(~S,~E,~F,~put,family,Leaf.reading(~S,~E,stream,id,tick))
67def observe(~S:Data,~E:Data,~F:Type,~take:F -> F & Leaf.Stream<S,E>,~put:F -> Leaf.Stream<S,E> -> F,family:F,id:U32,tick:Nat) -> F & Ev.Reading<E>:
68 observed_leaf(~S,~E,~F,~put,take(family),tick,id)
69
70def retained_values(~E:Data,remove:Bool,values:List<&2,E>) -> List<&2,E>:
71 match remove:
72 case True{}:[]
73 case False{}:values
74def retained_record(~A:Data,~B:Data,record:Publication<A,B>,+tick:Nat,+firstDropped:Nat,+secondDropped:Nat) -> Publication<A,B>:
75 match record:
76 case Publication{first,second}:Publication{retained_values(~A,Nat.is_le(tick,firstDropped),first),retained_values(~B,Nat.is_le(tick,secondDropped),second)}
77def nonempty_record(~A:Data,~B:Data,record:Publication<A,B>) -> Bool:
78 match record:
79 case Publication{first,second}:Nat.is_lt(0n,Nat.add(Ev.count(~A,first),Ev.count(~B,second)))
80def retained_head(~A:Data,~B:Data,keep:Bool,record:Publication<A,B>,rest:List<&2,Publication<A,B>>) -> List<&2,Publication<A,B>>:
81 match keep:
82 case True{}:record <> rest
83 case False{}:rest
84def retained_join(~A:Data,~B:Data,+record:Publication<A,B>,rest:List<&2,Publication<A,B>>) -> List<&2,Publication<A,B>>:
85 retained_head(~A,~B,nonempty_record(~A,~B,record),record,rest)
86def retained_records(~A:Data,~B:Data,records:List<&2,Publication<A,B>>,+tick:Nat,+firstDropped:Nat,+secondDropped:Nat) -> List<&2,Publication<A,B>>:
87 match records:
88 case []:[]
89 case record <> rest:retained_join(~A,~B,retained_record(~A,~B,record,tick,firstDropped,secondDropped),retained_records(~A,~B,rest,tick,firstDropped,secondDropped))
90def retained_batch(~A:Data,~B:Data,keep:Bool,tick:Nat,records:List<&2,Publication<A,B>>,rest:List<&2,Ev.Batch<Publication<A,B>>>) -> List<&2,Ev.Batch<Publication<A,B>>>:
91 match keep:
92 case True{}:Ev.Batch{tick,records} <> rest
93 case False{}:rest
94def retained_batch_join(~A:Data,~B:Data,tick:Nat,+records:List<&2,Publication<A,B>>,rest:List<&2,Ev.Batch<Publication<A,B>>>) -> List<&2,Ev.Batch<Publication<A,B>>>:
95 retained_batch(~A,~B,Nat.is_lt(0n,Ev.count(~Publication<A,B>,records)),tick,records,rest)
96def retained_journal(~A:Data,~B:Data,journal:List<&2,Ev.Batch<Publication<A,B>>>,+firstDropped:Nat,+secondDropped:Nat) -> List<&2,Ev.Batch<Publication<A,B>>>:
97 match journal:
98 case []:[]
99 case Ev.Batch{+tick,records} <> rest:retained_batch_join(~A,~B,tick,retained_records(~A,~B,records,tick,firstDropped,secondDropped),retained_journal(~A,~B,rest,firstDropped,secondDropped))
100def framed_journal(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,world:W.World<S,C,R,Publication<A,B>>,streams:Streams<S,A,B,H>,+tick:Nat,+journal:List<&2,Ev.Batch<Publication<A,B>>>) -> Application<S,C,R,A,B,H>:
101 Application{W.events_replace(~S,~C,~R,~Publication<A,B>,world,Ev.values(~Publication<A,B>,journal)),streams,journal,tick,tick}
102def framed_streams(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,world:W.World<S,C,R,Publication<A,B>>,streams:Streams<S,A,B,H>,tick:Nat,journal:List<&2,Ev.Batch<Publication<A,B>>>) -> Application<S,C,R,A,B,H>:
103 match streams:
104 case Streams{+first,+second,owners}:
105 framed_journal(~S,~C,~R,~A,~B,~H,world,Streams{first,second,owners},tick,retained_journal(~A,~B,journal,Leaf.dropped(~S,~A,first),Leaf.dropped(~S,~B,second)))
106def frame(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,app:Application<S,C,R,A,B,H>) -> Application<S,C,R,A,B,H>:
107 match app:
108 case Application{world,Streams{first,second,owners},journal,+tick,_}:framed_streams(~S,~C,~R,~A,~B,~H,world,Streams{Leaf.frame(~S,~A,first,tick),Leaf.frame(~S,~B,second,tick),owners},tick,journal)