Listening to transitions of invoked/spawned services

[geekox86]

Greetings all,

I have a generic machine that I invoke/spawn and I want to react to some of its transitions in its parent machine. I cannot find a mechanism to achieve this without forcing the generic machine to send events to its parent on each transition which is tedious and couples it a bit with its parent.

Do I have other/simpler ways?
Sort of like onTransition but used inside the parent machine.

[korniychuk]

The question is still relevant.

I would like to add one sub-question:
Maybe it's prevented on purpose and listening to child actors breaks some design pattern that XState authors laid down?

I've implemented a tricky example of how to make it work, but it looks like shit. I wouldn't do it in production code.

import { ActorRefFrom, AnyActorRef, assertEvent, assign, enqueueActions, fromCallback, raise, setup } from 'xstate';
import { BehaviorSubject, combineLatest, from } from 'rxjs';
import * as $ from 'rxjs/operators';

// This is a shitty workaround to listen the state of the spawned actors

export const randomCounterMachine = setup({
  types: {} as {
    context: {
      count: number;
    };
    events:
      | { type: 'NEXT' }
      ;
  },
  actors: {
  },
  actions: {
    increment: assign({
      count: ({ context: { count } }) => count + 1,
    }),
  },
  guards: {
  },
}).createMachine({
  context: ({ self }) => {
    console.log('Counter [%s] is instantiated', self.id);
    return {
      count: 0,
    };
  },
  entry: [
    ({ context, self }) => console.log('Counter [%s] entry, count: %s', self.id, context.count),
    raise({ type: 'NEXT' }, {
      delay: () => Math.random() * 1000,
    }),
  ],
  on: {
    NEXT: {
      target: '.', // without .target re-enter doesn't work!
      actions: 'increment',
      reenter: true,
    },
  },
});

export type RandomCounterActorRef = ActorRefFrom<typeof randomCounterMachine>;

interface ChildSpawnedEvent {
  type: 'CHILD_SPAWNED';
  systemId: string;
}

export interface ChildRemovedEvent {
  type: 'CHILD_REMOVED';
  systemId: string;
}

export const parentMachine = setup({
  types: {} as {
    context: {
      childIds: string[];
      count: number;
    };
    events:
      | { type: 'COUNT_UPDATE'; value: number }
      | { type: 'ADD_CHILD' }
      | { type: 'REMOVE_CHILD'; systemId: string }
      ;
  },
  actors: {
    child: randomCounterMachine,
    childListener: fromCallback<ChildSpawnedEvent | ChildRemovedEvent>(({ self, receive, sendBack }) => {
      const counterRefs$ = new BehaviorSubject<[systemId: string, ref: RandomCounterActorRef][]>([]);

      const totalCount$ = counterRefs$.pipe(
        $.tap(counters => console.log('counters list updated$', counters.map(([systemId]) => systemId))),
        $.map(counters =>
          counters.map(([, counter]) =>
            from(counter).pipe(
              $.startWith(counter.getSnapshot()),
            ),
          ),
        ),
        $.switchMap(countersWithInitialSnapshot => combineLatest(countersWithInitialSnapshot)),
        $.map(snapshots => snapshots.reduce((acc, snapshot) => acc + snapshot.context.count, 0)),
      );

      const sub = totalCount$.subscribe(count => sendBack({ type: 'COUNT_UPDATE', value: count }));

      receive((event) => {
        if (event.type === 'CHILD_SPAWNED') {
          queueMicrotask(() => { // simultaneously doesn't work
            const ref = self.system.get(event.systemId);
            if (!ref) {
              console.error(`Child not found: ${ event.systemId }`);
              return;
            }

            counterRefs$.next([...counterRefs$.getValue(), [event.systemId, ref]]);
          });
        } else if (event.type === 'CHILD_REMOVED') {
          counterRefs$.next(
            counterRefs$.getValue().filter(([systemId]) => systemId !== event.systemId),
          );
        }

      });

      return () => sub.unsubscribe();
    }),
  },
  actions: {
    addChild: enqueueActions(({ enqueue, self, context, system }) => {
      const systemId = 'system_child_' + context.childIds.length;
      enqueue.assign({
        childIds: [...context.childIds, systemId],
      });
      enqueue.spawnChild('child', {
        systemId,
      });

      queueMicrotask(() => { // simultaneously doesn't work
        console.log('Child spawned', systemId, system.get(systemId));
      });

      const totalCountListenerRef = system.get('totalCountListener') as AnyActorRef;
      if (!totalCountListenerRef) throw new Error(`Listener not found: totalCountListener`);

      totalCountListenerRef.send({ type: 'CHILD_SPAWNED', systemId });
    }),
    removeChild: enqueueActions(({ enqueue, system, context }, childSystemId: string) => {
      enqueue.assign({
        childIds: context.childIds.filter(id => id !== childSystemId),
      });

      const ref = system.get(childSystemId);
      if (!ref) throw new Error(`Child not found: ${ childSystemId }`);
      enqueue.stopChild(ref);

      console.log('Child removed', childSystemId);

      const totalCountListenerRef = system.get('totalCountListener') as AnyActorRef;
      if (!totalCountListenerRef) throw new Error(`Listener not found: totalCountListener`);

      totalCountListenerRef.send({ type: 'CHILD_REMOVED', systemId: childSystemId });
    }),
    setTotalCount: assign({
      count: ({ event }) => {
        assertEvent(event, 'COUNT_UPDATE');
        return event.value;
      },
    }),
  },
  guards: {
  },
}).createMachine({
  context: ({ spawn, self }) => {
    return {
      childIds: [],
      count: 0,
    };
  },
  invoke: [
    {
      src: 'childListener',
      systemId: 'totalCountListener',
    }
  ],
  on: {
    ADD_CHILD: {
      actions: 'addChild',
    },
    REMOVE_CHILD: {
      actions: {
        type: 'removeChild',
        params: ({ event }) => event.systemId
      },
    },
    COUNT_UPDATE: {
      actions: 'setTotalCount',
    },
  }
});

export type ParentActorRef = ActorRefFrom<typeof parentMachine>;

Angular component:

@Component({
  selector: 'app-listen-spawned',
  standalone: true,
  imports: [CommonModule],
  template: `
    Total: {{ total$ | async }}
    <br>
    <button (click)="add()">Add</button>
    <div *ngFor="let systemId of counterIds$ | async">
      {{ systemId }} |
      <button (click)="remove(systemId)">Remove</button>
    </div>
  `,
  styles: `
    :host {
      display: block;
    }
  `,
  changeDetection: ChangeDetectionStrategy.OnPush,
})
export class ListenSpawnedComponent implements OnInit {
  readonly actor = createActor(parentMachine);
  readonly state$ = ngActorState$(() => this.actor);

  readonly total$ = this.state$.pipe(
    $.map((state) => state.context.count),
  );
  readonly counterIds$ = this.state$.pipe(
    $.map((state) => state.context.childIds),
  );

  public ngOnInit(): void {
    this.actor.start();
  }

  public add(): void {
    this.actor.send({ type: 'ADD_CHILD' });
  }

  public remove(systemId: string): void {
    this.actor.send({ type: 'REMOVE_CHILD', systemId });
  }

}

Demo:

Screenshot.2024-03-22.at.15.55.15.mp4

[korniychuk]

Another implementation (better, without queueMicrotask):

import { ActorRefFrom, assertEvent, assign, EventObject, fromCallback, setup, stopChild } from 'xstate';
import { combineLatest, from, fromEventPattern, merge, Observable, of } from 'rxjs';
import { RandomCounterActorRef, randomCounterMachine } from './random-counter.machine';
import * as $ from 'rxjs/operators';

interface CounterSpawnedEvent extends EventObject {
  type: 'COUNTER_SPAWNED';
  ref: RandomCounterActorRef;
}

export const parentMachine = setup({
  types: {} as {
    context: {
      ids: string[];
      total: number,
    };
    events:
      | { type: 'ADD_COUNTER' }
      | { type: 'REMOVE_COUNTER'; id: string }
      | { type: 'TOTAL_UPDATED'; value: number }
      ;
  },
  actors: {
    randomCounter: randomCounterMachine,
    totalCounter: fromCallback<CounterSpawnedEvent>(({ receive, sendBack }) => {
      const spawnedCounter$ = fromEventPattern<RandomCounterActorRef>((handler) => {
        receive((event) => {
          if (event.type === 'COUNTER_SPAWNED') {
            console.log('COUNTER_SPAWNED received', event.ref.id);
            handler(event.ref);
          } else {
            console.log('totalCounter: Unknown event type', event.type);
          }
        })
      });

      const activeCounters$ = spawnedCounter$.pipe(
        $.mergeMap((actor): Observable<{ type: 'add' | 'remove', actor: RandomCounterActorRef }> => merge(
          of({ type: 'add', actor } as const),
          from(actor).pipe(
            $.last(),
            $.defaultIfEmpty(undefined),
            $.map(() => ({ type: 'remove', actor } as const)),
          ),
        )),
        $.scan((acc: RandomCounterActorRef[], curr: { type: 'add' | 'remove', actor: RandomCounterActorRef }) =>
            curr.type === 'add'
            ? [...acc, curr.actor]
            : acc.filter(actor => actor !== curr.actor)
          ,
          [],
        ),
      );

      const total$ = activeCounters$.pipe(
        $.map(counters =>
          counters.map((counter) =>
            from(counter).pipe(
              $.startWith(counter.getSnapshot()),
            ),
          ),
        ),
        $.switchMap(countersWithInitialSnapshot => combineLatest(countersWithInitialSnapshot)),
        $.map(snapshots => snapshots.reduce((acc, snapshot) => acc + snapshot.context.count, 0)),
      );

      const sub = total$.subscribe(value => sendBack({ type: 'TOTAL_UPDATED', value }));

      return () => sub.unsubscribe();
    }),
  },
  actions: {
    setTotalCount: assign({
      total: ({ event }) => {
        assertEvent(event, 'TOTAL_UPDATED');
        return event.value;
      },
    }),
    addCounter: assign({
      ids: ({ spawn, context, event, self, system }) => {
        const id = `counter-${ context.ids.length }`;
        const ref = spawn('randomCounter', { id });

        const aggregator = system.get('aggregator');
        if (!aggregator) {
          throw new Error(`Aggregator not found in system`);
        }
        aggregator.send({ type: 'COUNTER_SPAWNED', ref } satisfies CounterSpawnedEvent);

        return context.ids.concat(id);
      },
    }),
    stopCounter: stopChild(({ event }) => {
      assertEvent(event, 'REMOVE_COUNTER');
      console.log('Stopping counter [id: %s]', event.id);
      return event.id;
    }),
    delistCounter: assign({
      ids: ({ context, event }) => {
        assertEvent(event, 'REMOVE_COUNTER');
        console.log('Removing counter [id: %s] from list', event.id);
        return context.ids.filter(id => id !== event.id);
      },
    }),
  },
  guards: {
  },
}).createMachine({
  context: () => ({
    ids: [],
    total: 0,
  }),
  invoke: {
    src: 'totalCounter',
    systemId: 'aggregator',
  },
  on: {
    REMOVE_COUNTER: {
      actions: ['stopCounter', 'delistCounter'],
    },
    ADD_COUNTER: {
      actions: 'addCounter'
    },
    TOTAL_UPDATED: {
      actions: 'setTotalCount',
    }
  }
});

export type ParentActorRef = ActorRefFrom<typeof parentMachine>;