ordinary_app_event_family_reader.bend
bendvy-ecs-core@0.1.0.1 · 118 lines · 0 definitions
1import Base
2import bendvy-ecs-core@0.1.0.1/world.bend as W
3import bendvy-ecs-core@0.1.0.1/system.bend as Sys
4import bendvy-ecs-core@0.1.0.1/event_runtime.bend as Ev
5import bendvy-ecs-core@0.1.0.1/event_stream.bend as Stream
6import bendvy-ecs-core@0.1.0.1/event_family_routing.bend as Route
7import bendvy-ecs-core@0.1.0.1/ordinary_event_declaration.bend as Event
8import bendvy-ecs-core@0.1.0.1/ordinary_app_event.bend as OrdinaryEvent
9import bendvy-ecs-core@0.1.0.1/ordinary_app_registration_fields.bend as Records
10import bendvy-ecs-core@0.1.0.1/ordinary_query_declaration.bend as Query
11import bendvy-ecs-core@0.1.0.1/ordinary_app_observation_data.bend as DTO
12import bendvy-ecs-core@0.1.0.1/ordinary_app_composer.bend as Build
13import bendvy-ecs-core@0.1.0.1/ordinary_app_schema.bend as SchemaApp
14import bendvy-ecs-core@0.1.0.1/ordinary_app_event_family_runtime.bend as App
15import bendvy-ecs-core@0.1.0.1/ordinary_app_records.bend as Leaf
16import bendvy-ecs-core@0.1.0.1/ordinary_app_recipe.bend as Tree
17import bendvy-ecs-core@0.1.0.1/ordinary_schema.bend as Schema
18# One retained operational declaration and the actual returned reader/Registry.
19# Debug folds only registration fields and never reads or advances the Reader.
20# Registered App reader for one typed Data event stream E in family F.
21# Retain the actual declaration, reader and Registry returned after each run;
22# registration metadata alone cannot replace their runtime authority. Different
23# readers of the same stream keep independent positions. F may also contain
24# affine owners, but this reader's event payload E is Data, not an owned Type log.
25type Owner<S:Data,E:Data,Reader:Type,Registry:Type> is Type:
26 Owner{declaration:Event.Declaration<S,E>,reader:Reader,registry:Registry}
27# Advanced App metadata projection: return this owner with its registration
28# fields and event-read clause. The reported Registry cursor is not a read of
29# the event stream's position. Debug collection does not deliver, advance or
30# trim events; the actual reader and Registry are returned inside the owner.
31def fields(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~Reader:Type,~A:Type,~Error:Data,~Out:Type,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,owner:Owner<S,E,Reader,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>) -> Owner<S,E,Reader,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>> & List<&2,Records.Fields<S>>:
32 match owner:
33 case Owner{Event.Declaration{Event.Domain{+key},+mode},reader,Sys.Registry{+namespace,+id,+name,+access,+cursor}}:(Owner{Event.Declaration{Event.Domain{key},mode},reader,Sys.Registry{namespace,id,name,access,cursor}},[Records.Fields{namespace,id,name,access,cursor,key,[Query.Clause{key,OrdinaryEvent.query_mode(mode)}],Records.EventKind{}}])
34# Advanced App observer returning the same World with an empty Unit observation.
35# Event inventory comes from registration metadata; this function does not read
36# payloads or advance a stream cursor merely to display the App.
37def observe(~S:Data,~C:Type,~R:Type,~P:Data,world:W.World<S,C,R,P>) -> W.World<S,C,R,P> & DTO.Observation<Unit>:(world,DTO.Observation{Unit{},[]})
38# Reader registration result. Registered returns the runtime plus its actual
39# reader/Registry owner; RegistrationFailed returns the runtime without a new
40# owner. Thread the returned runtime on either branch instead of retaining an
41# old World or copying the family owners.
42type Registration<S:Data,C:Type,R:Type,P:Data,E:Data,F:Type,A:Type,Error:Data,Out:Type,-take:F -> F & Stream.Stream<S,E>,-put:F -> Stream.Stream<S,E> -> F,-runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>> is Type:
43 Registered{runtime:Route.Runtime<S,C,R,P,F>,owner:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>}
44 RegistrationFailed{runtime:Route.Runtime<S,C,R,P,F>}
45def registered(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,declaration:Event.Declaration<S,E>,result:Route.Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>) -> Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
46 match result:
47 case Route.RegistrationFailed{runtime}:RegistrationFailed{runtime}
48 case Route.Registered{runtime,reader,registry}:Registered{runtime,Owner{declaration,reader,registry}}
49def declared(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,family:F,stream:Stream.Stream<S,E>,tick:Nat,name:String,+declaration:Event.Declaration<S,E>,named:W.World<S,C,R,P> & U32) -> Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
50 registered(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,declaration,Route.registration_named(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,family,stream,tick,declaration,name,named))
51def taken(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,world:W.World<S,C,R,P>,tick:Nat,name:String,returned:F & Stream.Stream<S,E>) -> Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
52 match returned:
53 case (family,+stream):declared(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,family,stream,tick,name,Stream.read_declaration(~S,~E,stream),W.namespace(~S,~C,~R,~P,world))
54# Register a reader against the stream selected by the closed take/put route.
55# Derives read authority from that stream's operational declaration and uses the
56# ordinary Registry registration rules. It allocates a reader identity without
57# delivering events. Refusal, including exhausted reader IDs or failed Registry
58# registration, returns RegistrationFailed with the returned runtime.
59# The runner receives event-runtime.Input{values, lagged, args}. This entry is
60# for Data streams; affine payload publication/reading uses the separate
61# ordinary-owned-event-* APIs. Qualified family setup and retention/refusal
62# scenarios are linked from docs/parity/typed-data-events.md.
63def register(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,runtime:Route.Runtime<S,C,R,P,F>,name:String) -> Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
64 match runtime:
65 case Route.Runtime{world,family,tick}:taken(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,world,tick,name,take(family))
66# Advanced type alias for the retained reader and Registry for this exact
67# stream route and runner. Use it in App owner aliases, not as a numeric ID token.
68def OwnerType(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>) -> Type:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>
69# App type after installing this family reader alongside prior owners.
70# Keeps the schema, runtime family and combined metadata observer; this leaf adds
71# Unit to observation rows rather than copying event payloads into debug output.
72# Use install to perform registration and construct the extended application.
73def Application(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~Owners:Type,~Rows:Data,~prior:W.World<S,C,R,P> -> W.World<S,C,R,P> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>) -> Type:App.Application<S,C,R,P,F,Owners & Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,DTO.Product<Rows,Unit>,Tree.observe_pair(~W.World<S,C,R,P>,~Rows,~Unit,~prior,~observe(~S,~C,~R,~P)),Leaf.fold_pair(~S,~Owners,~Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,~fold,~fields(~S,~C,~R,~P,~E,~Route.Reader<S,E,F,take,put>,~A,~Error,~Out,~runner))>
74# Advanced composition entry for an already-registered reader owner.
75# Consumes the composer and owner, returning their combined owners/observer.
76# Does not register a second reader or deliver events. Ordinary setup uses install.
77def append(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~Owners:Type,~Rows:Data,~prior:W.World<S,C,R,P> -> W.World<S,C,R,P> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>,owner:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,value:Build.Registered<S,W.World<S,C,R,P>,Owners,Rows,prior,fold>) -> Build.Registered<S,W.World<S,C,R,P>,Owners & Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,DTO.Product<Rows,Unit>,Tree.observe_pair(~W.World<S,C,R,P>,~Rows,~Unit,~prior,~observe(~S,~C,~R,~P)),Leaf.fold_pair(~S,~Owners,~Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,~fold,~fields(~S,~C,~R,~P,~E,~Route.Reader<S,E,F,take,put>,~A,~Error,~Out,~runner))>:
78 match value:
79 case Build.Registered{world,owners,_,debug}:Build.appended(~S,~W.World<S,C,R,P>,~Owners,~Rows,~prior,~fold,~Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,~Unit,~observe(~S,~C,~R,~P),~fields(~S,~C,~R,~P,~E,~Route.Reader<S,E,F,take,put>,~A,~Error,~Out,~runner),world,owners,owner,debug)
80def appended(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~Owners:Type,~Rows:Data,~prior:W.World<S,C,R,P> -> W.World<S,C,R,P> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>,schema:Schema.Schema<S,C,R>,family:F,tick:Nat,owner:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,value:Build.Registered<S,W.World<S,C,R,P>,Owners,Rows,prior,fold>) -> Application(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold):
81 App.attached(~S,~C,~R,~P,~F,~(Owners & Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>),~DTO.Product<Rows,Unit>,~Tree.observe_pair(~W.World<S,C,R,P>,~Rows,~Unit,~prior,~observe(~S,~C,~R,~P)),~Leaf.fold_pair(~S,~Owners,~Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,~fold,~fields(~S,~C,~R,~P,~E,~Route.Reader<S,E,F,take,put>,~A,~Error,~Out,~runner)),family,tick,SchemaApp.attached(~S,~C,~R,~P,~(Owners & Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>),~DTO.Product<Rows,Unit>,~Tree.observe_pair(~W.World<S,C,R,P>,~Rows,~Unit,~prior,~observe(~S,~C,~R,~P)),~Leaf.fold_pair(~S,~Owners,~Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,~fold,~fields(~S,~C,~R,~P,~E,~Route.Reader<S,E,F,take,put>,~A,~Error,~Out,~runner)),schema,append(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold,owner,value)))
82def installed(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~Owners:Type,~Rows:Data,~prior:W.World<S,C,R,P> -> W.World<S,C,R,P> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>,~Output:Type,done:Application(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold) -> Output,failed:App.Application<S,C,R,P,F,Owners,Rows,prior,fold> -> Output,schema:Schema.Schema<S,C,R>,owners:Owners,recipe:Tree.Recipe<W.World<S,C,R,P>,Rows,prior>,debug:Bool,result:Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>) -> Output:
83 match result:
84 case RegistrationFailed{runtime}:failed(App.Application{schema,runtime,owners,recipe,debug})
85 case Registered{Route.Runtime{world,family,tick},owner}:done(appended(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold,schema,family,tick,owner,Build.Registered{world,owners,recipe,debug}))
86# Register and append a typed Data-family reader to an existing family App.
87# Consumes value and calls done with the extended App on success, or failed
88# with the returned previous App on refusal. Schema, family, prior owners and
89# debug setting are threaded through both branches. name follows Registry rules.
90# This is setup only: it neither runs the reader nor advances its delivery cursor.
91def install(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~Owners:Type,~Rows:Data,~prior:W.World<S,C,R,P> -> W.World<S,C,R,P> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>,~Output:Type,done:Application(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold) -> Output,failed:App.Application<S,C,R,P,F,Owners,Rows,prior,fold> -> Output,value:App.Application<S,C,R,P,F,Owners,Rows,prior,fold>,name:String) -> Output:
92 match value:
93 case App.Application{schema,runtime,owners,recipe,debug}:installed(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold,~Output,done,failed,schema,owners,recipe,debug,register(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,runtime,name))
94# Returned runtime and reader owner together with event-runtime.Completion.
95# Succeeded carries Out, Failed carries the runner's error, and Rejected returns
96# Args when execution authority is refused. Always retain the returned runtime
97# and owner; failure does not mean the input owners can be reused separately.
98type RunResult<S:Data,C:Type,R:Type,P:Data,E:Data,F:Type,A:Type,Error:Data,Out:Type,-take:F -> F & Stream.Stream<S,E>,-put:F -> Stream.Stream<S,E> -> F,-runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>> is Type:
99 Ran{runtime:Route.Runtime<S,C,R,P,F>,owner:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,completion:Ev.Completion<Error,Out,A>}
100def ran(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,declaration:Event.Declaration<S,E>,result:Route.RunResult<S,C,R,P,E,F,A,Error,Out,take,put,runner>) -> RunResult<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
101 match result:
102 case Route.Ran{runtime,reader,registry,completion}:Ran{runtime,Owner{declaration,reader,registry},completion}
103# Run this registered reader against its typed family stream.
104# The runner receives retained unread values in publication order, a lagged flag
105# and args. On Sys.Succeeded the reader advances and synchronize threads committed
106# World/family publication into the returned runtime. On Sys.Failed its stream
107# position does not advance; the runner's returned World is retained. This wrapper
108# does not itself undo arbitrary callback or host-IO effects.
109# A routing/registration rejection returns Rejected{args} without consuming the
110# read position. The completion and actual reader/Registry owners are returned
111# on every branch. Stream retention/capacity still bound later availability;
112# lagged reports dropped unread history, so failure is not an unlimited retry
113# retention promise. Supply the matching family synchronization route; merely
114# observing fields/metadata is not a run. See event-family-routing and the
115# qualified scenarios in docs/parity/typed-data-events.md.
116def run(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~synchronize:W.World<S,C,R,P> -> F -> Nat -> W.World<S,C,R,P> & F,owner:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,runtime:Route.Runtime<S,C,R,P,F>,args:A) -> RunResult<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
117 match owner:
118 case Owner{declaration,reader,registry}:ran(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,declaration,Route.run(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~synchronize,runtime,reader,registry,args))