1 import Base 2 import ./owned_event_log.bend as Log 3 4 type Family<-S:Data,-P:Type,-key:String> is Type: 5 Family{identity:U32,open:Bool,log:Log.Log

,previous:List<&2,Nat>,current:List<&2,Nat>,nextReader:U32,active:List<&2,U32>,droppedThrough:Nat} 6 type Reader<-S:Data,-P:Type,-key:String> is Type: 7 Reader{identity:U32,id:U32,cursor:Nat} 8 type Creation<-S:Data,-P:Type,-key:String> is Type: 9 Created{family:Family} 10 CreationRefused{} 11 type Start is Data: 12 Retained{} 13 Future{} 14 type Refusal is Data: 15 Closed{} 16 ForeignLog{} 17 Inactive{} 18 ReaderLimit{} 19 MissingPayload{} 20 type Publication<-P:Type> is Type: 21 Published{id:Nat} 22 PublicationRefused{payload:P} 23 type Registration<-S:Data,-P:Type,-key:String> is Type: 24 Registered{reader:Reader} 25 RegistrationRefused{reason:Refusal} 26 type Reading<-S:Data,-P:Type,-key:String,V:Data> is Type: 27 Item{reader:Reader,id:Nat,value:V,missed:Nat} 28 Empty{reader:Reader,missed:Nat} 29 ReadRefused{reader:Reader,reason:Refusal} 30 type Removal<-S:Data,-P:Type,-key:String> is Type: 31 Removed{reader:Reader} 32 RemovalRefused{reader:Reader,reason:Refusal} 33 type Snapshot is Data: 34 Snapshot{identity:U32,open:Bool,next:Nat,previous:List<&2,Nat>,current:List<&2,Nat>,nextReader:U32,active:List<&2,U32>,droppedThrough:Nat} 35 36 def created(~S:Data,~P:Type,~key:String,+identity:U32,valid:Bool) -> Creation: 37 match valid: 38 case False{}:CreationRefused{} 39 case True{}:Created{Family{identity,True{},Log.empty(~P),[],[],1,[],0n}} 40 def allocated(~S:Data,~P:Type,~key:String,+identity:U32) -> Creation: 41 created(~S,~P,~key,identity,Bool.and(U32.is_lt(0,identity),U32.is_lt(identity,4294967295))) 42 def has(+ids:List<&2,U32>,+wanted:U32) -> Bool: 43 match ids: 44 case []:False{} 45 case id <> rest:Bool.or(U32.is_eq(id,wanted),has(rest,wanted)) 46 def without_selected(equal:Bool,id:U32,rest:List<&2,U32>) -> List<&2,U32>: 47 match equal: 48 case True{}:rest 49 case False{}:id <> rest 50 def without(ids:List<&2,U32>,+wanted:U32) -> List<&2,U32>: 51 match ids: 52 case []:[] 53 case +id <> rest:without_selected(U32.is_eq(id,wanted),id,without(rest,wanted)) 54 def published(~S:Data,~P:Type,~key:String,identity:U32,previous:List<&2,Nat>,current:List<&2,Nat>,nextReader:U32,active:List<&2,U32>,dropped:Nat,result:Log.Log

& Nat) -> Family & Publication

: 55 match result: 56 case (log,+id):(Family{identity,True{},log,previous,List.append(&2,Nat,current,[id]),nextReader,active,dropped},Published{id}) 57 # Publication updates the owned log immediately, as Bevy Messages.write does. 58 # https://github.com/bevyengine/bevy/blob/ad678262ce53b5d142fe49ee5e08caff6f00ab60/crates/bevy_ecs/src/message/messages.rs#L120-L143 59 # A later handled body error does not stage or undo this already-authored send. 60 def publish(~S:Data,~P:Type,~key:String,family:Family,payload:P) -> Family & Publication

: 61 match family: 62 case Family{identity,False{},log,previous,current,nextReader,active,dropped}:(Family{identity,False{},log,previous,current,nextReader,active,dropped},PublicationRefused{payload}) 63 case Family{identity,True{},log,previous,current,nextReader,active,dropped}:published(~S,~P,~key,identity,previous,current,nextReader,active,dropped,Log.append(~P,log,payload)) 64 def starting(start:Start,next:Nat) -> Nat: 65 match start: 66 case Retained{}:0n 67 case Future{}:Nat.sub(next,1n) 68 def register_checked(~S:Data,~P:Type,~key:String,start:Start,available:Bool,family:Family) -> Family & Registration: 69 match available family: 70 case False{} family:(family,RegistrationRefused{ReaderLimit{}}) 71 case True{} Family{+identity,open,Log.Log{+next,rows},previous,current,+nextReader,active,dropped}:(Family{identity,open,Log.Log{next,rows},previous,current,U32.add(nextReader,1),List.append(&2,U32,active,[nextReader]),dropped},Registered{Reader{identity,nextReader,starting(start,next)}}) 72 def register(~S:Data,~P:Type,~key:String,start:Start,family:Family) -> Family & Registration: 73 match family: 74 case Family{identity,False{},log,previous,current,nextReader,active,dropped}:(Family{identity,False{},log,previous,current,nextReader,active,dropped},RegistrationRefused{Closed{}}) 75 case Family{identity,True{},log,previous,current,+nextReader,active,dropped}:register_checked(~S,~P,~key,start,U32.is_lt(nextReader,4294967295),Family{identity,True{},log,previous,current,nextReader,active,dropped}) 76 def admission_identity(same:Bool,active:Bool) -> Maybe<&2,Refusal>: 77 match same active: 78 case False{} _:Some{ForeignLog{}} 79 case True{} False{}:Some{Inactive{}} 80 case True{} True{}:None{} 81 def admission(open:Bool,identity:U32,actual:U32,id:U32,active:List<&2,U32>) -> Maybe<&2,Refusal>: 82 match open: 83 case False{}:Some{Closed{}} 84 case True{}:admission_identity(U32.is_eq(identity,actual),has(active,id)) 85 def remove_admitted(~S:Data,~P:Type,~key:String,refusal:Maybe<&2,Refusal>,family:Family,reader:Reader) -> Family & Removal: 86 match refusal family reader: 87 case Some{reason} family reader:(family,RemovalRefused{reader,reason}) 88 case None{} Family{identity,open,log,previous,current,nextReader,active,dropped} Reader{actual,+id,cursor}:(Family{identity,open,log,previous,current,nextReader,without(active,id),dropped},Removed{Reader{actual,id,cursor}}) 89 def remove(~S:Data,~P:Type,~key:String,family:Family,reader:Reader) -> Family & Removal: 90 match family reader: 91 case Family{+identity,+open,log,previous,current,nextReader,+active,dropped} Reader{+actual,+id,cursor}:remove_admitted(~S,~P,~key,admission(open,identity,actual,id,active),Family{identity,open,log,previous,current,nextReader,active,dropped},Reader{actual,id,cursor}) 92 def highest(ids:List<&2,Nat>,+initial:Nat) -> Nat: 93 match ids: 94 case []:initial 95 case id <> rest:highest(rest,Nat.max(initial,id)) 96 def updated(~S:Data,~P:Type,~key:String,identity:U32,current:List<&2,Nat>,nextReader:U32,active:List<&2,U32>,dropped:Nat,result:Log.Log

& List>) -> Family & List>: 97 match result: 98 case (log,retired):(Family{identity,True{},log,current,[],nextReader,active,dropped},retired) 99 # Messages.update_drain returns the previous generation's actual payload owners. 100 # https://github.com/bevyengine/bevy/blob/ad678262ce53b5d142fe49ee5e08caff6f00ab60/crates/bevy_ecs/src/message/messages.rs#L187-L218 101 # Bend returns retained rows explicitly instead of claiming an automatic IO finalizer. 102 def update(~S:Data,~P:Type,~key:String,family:Family) -> Family & List>: 103 match family: 104 case Family{identity,False{},log,previous,current,nextReader,active,dropped}:(Family{identity,False{},log,previous,current,nextReader,active,dropped},[]) 105 case Family{identity,True{},log,previous,+current,nextReader,active,dropped}:updated(~S,~P,~key,identity,current,nextReader,active,highest(previous,dropped),Log.retain(~P,log,current)) 106 def shutdown_returned(~S:Data,~P:Type,~key:String,identity:U32,nextReader:U32,dropped:Nat,result:Log.Log

& List>) -> Family & List>: 107 match result: 108 case (log,rows):(Family{identity,False{},log,[],[],nextReader,[],dropped},rows) 109 def shutdown(~S:Data,~P:Type,~key:String,family:Family) -> Family & List>: 110 match family: 111 case Family{identity,_,log,previous,current,nextReader,_,dropped}:shutdown_returned(~S,~P,~key,identity,nextReader,highest(List.append(&2,Nat,previous,current),dropped),Log.drain(~P,log)) 112 def snapshot(~S:Data,~P:Type,~key:String,family:Family) -> Family & Snapshot: 113 match family: 114 case Family{+identity,+open,Log.Log{+next,rows},+previous,+current,+nextReader,+active,+dropped}:(Family{identity,open,Log.Log{next,rows},previous,current,nextReader,active,dropped},Snapshot{identity,open,next,previous,current,nextReader,active,dropped})