execution_stream_channel.bend

bendvy-ecs-core@0.1.0.3 · 68 lines · 20 definitions

1import Base
2import bendvy-ecs-core@0.1.0.3/machine_stream_queue.bend as Queue
3import bendvy-ecs-core@0.1.0.3/change_tick.bend as Tick
4
5# Shared message storage. Sequence numbers are independent of World ticks.
6# Bevy code: https://github.com/bevyengine/bevy/blob/ad678262ce53b5d142fe49ee5e08caff6f00ab60/crates/bevy_ecs/src/message/message_cursor.rs#L54-L63
7# Behavior: each reader tracks message count, independently of change ticks.
8# Bevy code: https://github.com/bevyengine/bevy/blob/ad678262ce53b5d142fe49ee5e08caff6f00ab60/crates/bevy_ecs/src/message/messages.rs#L188-L201
9# Behavior: every update removes the older generation and rotates the retained one.
10type Store<-E:Data> is Data:
11  Store{queue:Queue.Queue<E>,olderThrough:Nat,maintenance:Maybe<&2,Tick.Tick>}
12type Reading<E:Data> is Data:
13  Read{cursor:Nat,events:List<&2,E>,missed:Nat,tick:Tick.Tick}
14  Refused{}
15def empty_generic(-E:Data) -> Store<E>:Store{Queue.empty_generic(E),0n,None{}}
16def publish_generic(-E:Data,store:Store<E>,sequence:Nat,event:E) -> Store<E>:
17  match store:
18    case Store{queue,olderThrough,maintenance}:Store{Queue.publish_generic(E,queue,sequence,[event]),olderThrough,maintenance}
19def push_generic(-E:Data,+sequence:Nat,store:Store<E>,event:E) -> Nat & Store<E>:
20  (1n+sequence,publish_generic(E,store,1n+sequence,event))
21def maintained_generic(-E:Data,+sequence:Nat,tick:Tick.Tick,olderThrough:Nat,queue:Queue.Queue<E>) -> Store<E>:
22  match queue:
23    case Queue.Queue{front,back,size,dropped,+batches}:Store{Queue.past_loop_generic(E,1n+batches,Queue.Queue{front,back,size,dropped,batches},olderThrough),sequence,Some{tick}}
24def maintain_generic(-E:Data,sequence:Nat,tick:Tick.Tick,store:Store<E>) -> Store<E>:
25  match store:
26    case Store{queue,olderThrough,_}:maintained_generic(E,sequence,tick,olderThrough,queue)
27def missed_generic(-E:Data,queue:Queue.Queue<E>,cursor:Nat) -> Nat:
28  match queue:
29    case Queue.Queue{_,_,_,dropped,_}:Nat.sub(dropped,cursor)
30def reading_generic(-E:Data,+sequence:Nat,tick:Tick.Tick,+cursor:Nat,store:Store<E>) -> Store<E> & Reading<E>:
31  match store:
32    case Store{+queue,olderThrough,maintenance}:(Store{queue,olderThrough,maintenance},Read{sequence,Queue.read_generic(E,queue,cursor),missed_generic(E,queue,cursor),tick})
33def checked_generic(-E:Data,allowed:Bool,sequence:Nat,tick:Tick.Tick,cursor:Nat,store:Store<E>) -> Store<E> & Reading<E>:
34  match allowed:
35    case False{}:(store,Refused{})
36    case True{}:reading_generic(E,sequence,tick,cursor,store)
37def read_generic(-E:Data,namespace:Maybe<&2,U32>,actual:U32,sequence:Nat,tick:Tick.Tick,cursor:Nat,store:Store<E>) -> Store<E> & Reading<E>:
38  match namespace:
39    case None{}:(store,Refused{})
40    case Some{namespace}:checked_generic(E,U32.is_eq(namespace,actual),sequence,tick,cursor,store)
41
42# Preserve the public template API over the same erased generic channel policy.
43def empty(~E:Data) -> Store<E>:
44  empty_generic(E)
45
46def publish(~E:Data,store:Store<E>,sequence:Nat,event:E) -> Store<E>:
47  publish_generic(E,store,sequence,event)
48
49def push(~E:Data,+sequence:Nat,store:Store<E>,event:E) -> Nat & Store<E>:
50  push_generic(E,sequence,store,event)
51
52def maintained(~E:Data,+sequence:Nat,tick:Tick.Tick,olderThrough:Nat,queue:Queue.Queue<E>) -> Store<E>:
53  maintained_generic(E,sequence,tick,olderThrough,queue)
54
55def maintain(~E:Data,sequence:Nat,tick:Tick.Tick,store:Store<E>) -> Store<E>:
56  maintain_generic(E,sequence,tick,store)
57
58def missed(~E:Data,queue:Queue.Queue<E>,cursor:Nat) -> Nat:
59  missed_generic(E,queue,cursor)
60
61def reading(~E:Data,+sequence:Nat,tick:Tick.Tick,+cursor:Nat,store:Store<E>) -> Store<E> & Reading<E>:
62  reading_generic(E,sequence,tick,cursor,store)
63
64def checked(~E:Data,allowed:Bool,sequence:Nat,tick:Tick.Tick,cursor:Nat,store:Store<E>) -> Store<E> & Reading<E>:
65  checked_generic(E,allowed,sequence,tick,cursor,store)
66
67def read(~E:Data,namespace:Maybe<&2,U32>,actual:U32,sequence:Nat,tick:Tick.Tick,cursor:Nat,store:Store<E>) -> Store<E> & Reading<E>:
68  read_generic(E,namespace,actual,sequence,tick,cursor,store)