fix order of execution

2023-05-17 16:28:32 -04:00
parent 3e769c0fa5
commit 636ba279c8
8 changed files with 274 additions and 119 deletions
@@ -10,7 +10,7 @@ export interface EventReactionCouplings_T<C> {
  eventName: string;
  reactions: Array<Reaction_T<C>>;
 };
-export type Reaction_T<C> = SideEffect_T<C> | ContextMutation_T<C> | Goto_T;
+export type Reaction_T<C> = SideEffect_T<C> | ContextMutation_T<C> | Peering_T<C,unknown> | Goto_T;
 export interface SideEffect_T<C> {
  type: 'SideEffect';
  fn: SideEffectFunction_T<C>;
@@ -21,6 +21,12 @@ export interface ContextMutation_T<C> {
  fn: ContextMutationFunction_T<C>;
 };
 export type ContextMutationFunction_T<C> = (ctx:C,e:Event_T,self:Interpreter_T<C>)=>C;
+export interface Peering_T<C,C_Peer> {
+  type: 'Peering';
+  name: string;
+  peerCreationFunction: PeerCreationFunction_T<C,C_Peer>
+};
+export type PeerCreationFunction_T<C,C_Peer> = (ctx:C,e:Event_T,self:Interpreter_T<C>) => Interpreter_T<C_Peer>;
 export interface Goto_T {
  type: 'Goto';
  targetStateName: string;
@@ -32,16 +38,14 @@ export const On = function<C>(eventName:string, ...reactions:Array<Reaction_T<C>
 export const SideEffect = function<C>(fn:SideEffectFunction_T<C>) : SideEffect_T<C>{ return {type:'SideEffect', fn}; };
 export const Goto = function(targetStateName:string) : Goto_T { return {type:'Goto', targetStateName} };
 export const Context = function<C>(fn:ContextMutationFunction_T<C>) : ContextMutation_T<C> { return {type:'ContextMutation', fn} };
-
-
-
+export const Peer = function<C,C_Peer>(name:string, peerCreationFunction:PeerCreationFunction_T<C,C_Peer>) : Peering_T<C,C_Peer>{  return {type:'Peering', name, peerCreationFunction};  }


 export interface Interpreter_T<C> {
  machine: Machine_T<C>;
  state: string;
  context: C;
-  peers: Record<string, Interpreter_T<unknown> | Array<Interpreter_T<unknown>>>;
+  peers: Record<string, Interpreter_T<unknown>>;
  peerSubscriptionIds: Map<Interpreter_T<unknown>,string>;
  eventQueue:Array<Event_T>;
  subscriptionsToEvents: Record<string, EventsSubscriptionCallbackFunction_T<C>>; // called upon every event
@@ -64,7 +68,7 @@ export interface Interpreter_T<C> {
 export function Interpreter<C>(machine:Machine_T<C>, initialContext:any, initialStateName?:string) : Interpreter_T<C>{
  if(typeof initialStateName === 'undefined'){ initialStateName = machine.states[0].name; }
  //@ts-expect-error
-  const interpreter : Interpreter_T<C> = {machine, state: initialStateName, context:initialContext, eventQueue:[], isTransitioning:false, subscriptionsToEvents: {}, subscriptionsToState: {}, subscriptionsToSettledState: {}, isPaused: true};
+  const interpreter : Interpreter_T<C> = {machine, state: initialStateName, context:initialContext, eventQueue:[], isTransitioning:false, peers:{}, peerSubscriptionIds:new Map(), subscriptionsToEvents: {}, subscriptionsToState: {}, subscriptionsToSettledState: {}, isPaused: true};
  interpreter.start = ()=>{ start(interpreter); return interpreter; }
  send(interpreter, ['entry', null] );
  return interpreter;
@@ -125,20 +129,13 @@ function processNextEvent<C>(interpreter:Interpreter_T<C>){
    const reactions = eventReactionCouplings
      .map((eventReactionCoupling)=>eventReactionCoupling.reactions)
      .flat();
-    const {sideEffects, contextMutations, goto_} = categorizeReactions(reactions);
+    const {sideEffects, contextMutations, peerings, goto_} = categorizeReactions(reactions);
    // save the current context, before it's mutated, so as to pass it to sideEffects below:
    const originalContext = interpreter.context;
    // must process contextMutations in-series:
    contextMutations.forEach((contextMutation)=>{
      interpreter.context = contextMutation.fn(interpreter.context, event, interpreter);
    });
-    // run subscription-to-events callbacks (can be in parallel), since an event just happened:
-    Object.values(interpreter.subscriptionsToEvents).forEach((callbackFunction)=>{ callbackFunction(event, interpreter); });
-    // can process sideEffects in parallel (though we currently don't due to the overhead of doing so in Node.js):
-    // they're processed *after* the context changes, since that's what most sideEffects would be interested in; but nevertheless the original context is passed in case this sideEffect needs it:
-    sideEffects.forEach((sideEffect)=>{
-      sideEffect.fn(interpreter.context, event, interpreter, originalContext);
-    });
    // processing of `goto` must be last:
    if(goto_ !== null){
      send(interpreter, ['exit', null]);
@@ -147,12 +144,23 @@ function processNextEvent<C>(interpreter:Interpreter_T<C>){
      Object.values(interpreter.subscriptionsToState).forEach((callbackFunction)=>{ callbackFunction(event, interpreter); });
      send(interpreter, ['entry', null]);
    }
+    // now that "internal" stuff has been run, we can run "external" stuff:
+    // process peerings (possibly in parallel):
+    peerings.forEach((peering)=>{ addPeer(interpreter, peering.name, peering.peerCreationFunction(interpreter.context, event, interpreter)); });
+    // run subscription-to-events callbacks (can be in parallel), since an event just happened:
+    Object.values(interpreter.subscriptionsToEvents).forEach((callbackFunction)=>{ callbackFunction(event, interpreter); });
+    // can process sideEffects in parallel (though we currently don't due to the overhead of doing so in Node.js):
+    // they're processed *after* the context changes, since that's what most sideEffects would be interested in; but nevertheless the original context is passed in case this sideEffect needs it:
+    sideEffects.forEach((sideEffect)=>{
+      sideEffect.fn(interpreter.context, event, interpreter, originalContext);
+    });
  }
 }
-function categorizeReactions<C>(reactions:Array<Reaction_T<C>>) : {sideEffects:Array<SideEffect_T<C>>, contextMutations:Array<ContextMutation_T<C>>, goto_:Goto_T|null}{
+function categorizeReactions<C>(reactions:Array<Reaction_T<C>>) : {sideEffects:Array<SideEffect_T<C>>, contextMutations:Array<ContextMutation_T<C>>, peerings:Array<Peering_T<C,unknown>>, goto_:Goto_T|null}{
  let 
    sideEffects:Array<SideEffect_T<C>> = [], 
    contextMutations:Array<ContextMutation_T<C>> = [], 
+    peerings:Array<Peering_T<C,unknown>> = [],
    goto_:Goto_T|null = null;
  reactions.forEach((reaction)=>{
    if(reaction.type === 'SideEffect'){
@@ -161,11 +169,14 @@ function categorizeReactions<C>(reactions:Array<Reaction_T<C>>) : {sideEffects:A
    else if(reaction.type === 'ContextMutation'){
      contextMutations.push(reaction);
    }
+    else if(reaction.type === 'Peering'){
+      peerings.push(reaction);
+    }
    else if(reaction.type === 'Goto'){
      goto_ = reaction;
    }
  });
-  return {sideEffects, contextMutations, goto_};
+  return {sideEffects, contextMutations, peerings, goto_};
 }

 export type EventsSubscriptionCallbackFunction_T<C> = (e:Event_T, self:Interpreter_T<C>)=>void;
@@ -195,13 +206,15 @@ export function unsubscribe<C>(interpreter:Interpreter_T<C>, subscriptionId:stri
  delete interpreter.subscriptionsToEvents[subscriptionId.toString()];
 }

-export function addPeer<C, C_Peer>(self:Interpreter_T<C>, peer:Interpreter_T<C_Peer>, name:string){
+export function addPeer<C, C_Peer>(self:Interpreter_T<C>, name:string, peer:Interpreter_T<C_Peer>){
  self.peers[name] = peer;
+  subscribeToEvents(peer, (e, peer)=>{
+    // this `if` prevents infinite loops due to mutually-subscribed peers (cyclical dependencies):
+    if(self.isTransitioning === false){
+      send(self, [name+'.'+e[0], e[1]]);
+    }
+  });
 }
-export function addPeers(){}
-
-export const Spawn = function(){};
-export const Unspawn = function(){};

 /*
 export function useMachine(machine, options){
@@ -43,4 +43,4 @@ const machine =
    ),
  );

-const actor = Interpreter<C>(machine, {context:{}});
+const actor = Interpreter<C>(machine, {context:{}}).start();
@@ -1,34 +1,44 @@
-import { Machine, State, On, SideEffect, Goto, Spawn, Unspawn, Interpreter, Interpreter_T, send, Event_T, Context, SideEffectFunction_T } from '../index';
+import { Machine, State, On, SideEffect, Goto, Interpreter, Interpreter_T, send, Event_T, Context, SideEffectFunction_T, Peer, PeerCreationFunction_T, ContextMutationFunction_T } from '../index';

 const wait = (ms:number)=>new Promise((resolve)=>{ setTimeout(()=>{ resolve(1); }, ms); });
-const makeRequest : SideEffectFunction_T<Cc> = (ctx,e,self)=>{ send(ctx.serverActor, ['received-request',self]); };
-const sendResponse : SideEffectFunction_T<Cs> = (ctx,e,self)=>{ send(ctx.clientActor, ['received-response',self]); };
-const startTimer : SideEffectFunction_T<Cs> = async (ctx,e,self)=>{ await wait(1500); send(self, ['timer-finished',null]); }
-const log : SideEffectFunction_T<Cc|Cs> = (ctx, e, self)=>{ console.log(self.state, ctx); };
+const makeRequest : SideEffectFunction_T<Cc> = (ctx,e,self)=>{ send(self.peers.server, ['received-request',self]); };
+const sendResponse : SideEffectFunction_T<Cs> = (ctx,e,self)=>{ send(ctx.client, ['received-response',self]); };
+const startTimer : SideEffectFunction_T<Cs> = async (ctx,e,self)=>{ await wait(1500); console.log('   timer actually finished'); send(self, ['timer-finished',null]); }
+const log = (namespace:string)=>(ctx, e, self)=>{ console.log(namespace, self.state, e[0]); };
+const logClientStats : SideEffectFunction_T<Cc> = (ctx,e,self)=>{ console.log('client', ctx.requestsMade, ctx.responsesReceived); }
+const logServerStats : SideEffectFunction_T<Cs> = (ctx,e,self)=>{ console.log('server', ctx.requestsReceived, ctx.responsesSent); }
+const logEventQueue = (namespace:string)=>(ctx,e,self)=>{ console.log(namespace+'.eventQueue', [e[0]], self.eventQueue.map(([eventName])=>eventName)); }
+const saveClient : ContextMutationFunction_T<Cs> = (ctx, e, self)=>({...ctx, client:e[1]});
+const createServer : PeerCreationFunction_T<Cc,Cs> = (ctx, e, self)=>Interpreter(server,{requestsReceived:0, responsesSent:0}).start();

 type Cc = {
  requestsMade: number;
  responsesReceived: number;
-  serverActor: Interpreter_T<Cs>;
 };
 type Cs = {
-  clientActor: Interpreter_T<Cc>
+  client: Interpreter_T<Cc>;
+  requestsReceived:number;
+  responsesSent:number;
 };

 const client = 
  Machine<Cc>(
+    State('initializing',
+      On('entry',
+        Peer('server', createServer),
+        //SideEffect(log('client')),
+        Goto('idle'),
+      )
+    ),
    State<Cc>('idle',
      On<Cc>('entry',
-        SideEffect(log),
-      ),
-      On('server-created',
-        SideEffect((_ctx,[_eventName,serverActor],self)=>{ self.context.serverActor=serverActor; }),
+        //SideEffect(log('client')),
        Goto('making-request')
      )
    ),
    State<Cc>('making-request',
      On<Cc>('entry', 
-        SideEffect(log),
+        //SideEffect(log('client')),
        SideEffect(makeRequest),
        Context<Cc>((ctx)=>({...ctx, requestsMade: ctx.requestsMade+1})),
        Goto('awaiting-response')
@@ -36,11 +46,11 @@ const client =
    ),
    State<Cc>('awaiting-response',
      On<Cc>('entry',
-        SideEffect(log),
+        //SideEffect(log('client')),
      ),
      On<Cc>('received-response', 
-        SideEffect(log),
        Context<Cc>((ctx)=>({...ctx, responsesReceived: ctx.responsesReceived+1})),
+        //SideEffect(log('client')),
        Goto('making-request')
      ),
    ),
@@ -50,25 +60,38 @@ const server =
  Machine<Cs>(
    State<Cs>('awaiting-request',
      On<Cs>('entry',
-        SideEffect(log),
+        //SideEffect(log('server')),
+        Context<Cs>((ctx)=>({...ctx, requestsReceived: ctx.requestsReceived+1})),
      ),
      On('received-request', 
-        SideEffect((_ctx,[_eventName,clientActor],self)=>{ self.context.clientActor=clientActor; }),
+        //SideEffect(log('server')),
+        Context<Cs>(saveClient),
        Goto('sending-response')
      ),
    ),
    State<Cs>('sending-response',
      On<Cs>('entry', 
-        SideEffect(log),
+        //SideEffect(log('server')),
        SideEffect(startTimer)
      ),
      On('timer-finished',
+        //SideEffect(log('server')),
+        SideEffect(logServerStats),
        SideEffect(sendResponse),
-        Goto('awaiting-request')
-      )
+        Context<Cs>((ctx)=>({...ctx, responsesSent: ctx.responsesSent+1})),
+        Goto('awaiting-request') // for some reason, at this point there's a "received-request" waiting in the eventQueue, which gets processed before the "exit" then "entry" that get appended to the queue due to this Goto, which makes the Interpreter come right back to this State
+        /*
+          Server gets timer-finished, which sends response to client. 
+
+          But client, at the time, is not transitioning, so it immediately begins 
+          processing that event. The problem is that one of the sideeffects involved 
+          in processing that event is to send another request to the server, 
+          which hasn't yet even queued `exit`-then-`entry` events for its next state!
+          
+          So we have to ensure they get queued first, before processing the client.
+        */
+      ),
    ),
  );

-const clientActor = Interpreter(client, {context:{requestsMade:0, responsesReceived:0}});
-const serverActor = Interpreter(server, {context:{}});
-send(clientActor, ['server-created', serverActor]);
+const clientActor = Interpreter(client, {requestsMade:0, responsesReceived:0}).start();