1	import Base
2	import ./machine_stream_queue.bend as Queue
3	import ./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.
10	type Store<-E:Data> is Data:
11	  Store{queue:Queue.Queue<E>,olderThrough:Nat,maintenance:Maybe<&2,Tick.Tick>}
12	type Reading<E:Data> is Data:
13	  Read{cursor:Nat,events:List<&2,E>,missed:Nat,tick:Tick.Tick}
14	  Refused{}
15	def empty_generic(-E:Data) -> Store<E>:Store{Queue.empty_generic(E),0n,None{}}
16	def 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}
19	def push_generic(-E:Data,+sequence:Nat,store:Store<E>,event:E) -> Nat & Store<E>:
20	  (1n+sequence,publish_generic(E,store,1n+sequence,event))
21	def 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}}
24	def 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)
27	def missed_generic(-E:Data,queue:Queue.Queue<E>,cursor:Nat) -> Nat:
28	  match queue:
29	    case Queue.Queue{_,_,_,dropped,_}:Nat.sub(dropped,cursor)
30	def 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})
33	def 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)
37	def 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.
43	def empty(~E:Data) -> Store<E>:
44	  empty_generic(E)
45	
46	def publish(~E:Data,store:Store<E>,sequence:Nat,event:E) -> Store<E>:
47	  publish_generic(E,store,sequence,event)
48	
49	def push(~E:Data,+sequence:Nat,store:Store<E>,event:E) -> Nat & Store<E>:
50	  push_generic(E,sequence,store,event)
51	
52	def 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	
55	def maintain(~E:Data,sequence:Nat,tick:Tick.Tick,store:Store<E>) -> Store<E>:
56	  maintain_generic(E,sequence,tick,store)
57	
58	def missed(~E:Data,queue:Queue.Queue<E>,cursor:Nat) -> Nat:
59	  missed_generic(E,queue,cursor)
60	
61	def 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	
64	def 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	
67	def 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)