add support for proxying events from webworker to actors in different threads

Signed-off-by: aabidsofi19 <[email protected]>

Author: aabidsofi19

src/actors/utils.ts

@@ -3,6 +3,7 @@
 
 import { AnyActorRef, AnyEventObject, assign, enqueueActions, sendTo } from 'xstate';
 import { AnyActorSystem } from 'xstate/dist/declarations/src/system';
+import { workerEvents } from './worker/events';
 
 type ContextWithReturnAddress = { returnAddress: AnyActorRef };
 
@@ -39,8 +40,28 @@ export const forwardToActors = (actorSystemIds: string[]) =>
 
 export const deadLetter = (event: AnyEventObject) => ({ type: 'DEAD_LETTER', event });
 
-export const reply = (eventFn: (actionArgs: any, params: any) => AnyEventObject) =>
-  sendTo(({ context }: { context: ContextWithReturnAddress }) => context.returnAddress, eventFn);
+
+
+function isInWorker() {
+  return (
+    typeof self !== 'undefined' &&         // 'self' exists
+    typeof self?.document === 'undefined'        // no Window in worker
+  );
+}
+
+export const reply = (eventFn: (actionArgs: any, params: any) => AnyEventObject) =>  enqueueActions(({ enqueue, ...actionArgs }, params) => {
+
+    if (!actionArgs.context.returnAddress) {
+        console.warn('No return address specified in context for reply action');
+        return;
+    }
+  if (isInWorker()) {
+        console.log('reply in worker - posting message to main thread', actionArgs.context.returnAddress, eventFn(actionArgs, params));
+        postMessage(workerEvents.proxyEvent(eventFn(actionArgs, params), actionArgs.context.returnAddress));
+        return;
+  }
+   enqueue.sendTo(actionArgs.context.returnAddress, eventFn);
+})
 
 export const XSTATE_DEBUG_EVENT = 'XSTATE_DEBUG_EVENT';
 

src/actors/validators/dataValidator.ts

@@ -7,7 +7,6 @@ import {
   SnapshotFrom,
   assign,
   fromPromise,
-  sendParent,
   setup
 } from 'xstate';
 import { reply } from '../utils';
@@ -55,15 +54,6 @@ interface DataValidationFailedEvent extends EventObject {
   };
 }
 
-const toSerializable = (data: unknown) => {
-    try {
-        return JSON.parse(JSON.stringify(data));
-    } catch {
-        console.warn('Data is not serializable, using toPlainObject as fallback', data);
-        return {}
-    }
-}
-
 export const dataValidatorCommands = {
   validateData: ({
     validationPayload,
@@ -196,7 +186,6 @@ export const dataValidatorMachine = setup({
           target: 'idle',
           actions: [
             'setValidationResults',
-            // send to parent
             reply(
               ({
                 context,
@@ -206,7 +195,7 @@ export const dataValidatorMachine = setup({
                 event: ValidateActorDoneEvent;
               }) =>
                 dataValidatorEvents.dataValidated({
-                  validationPayload: toSerializable(context.validationPayload) as ValidationPayload,
+                  validationPayload: context.validationPayload as ValidationPayload,
                   validationResults: event.output.validationResults
                 })
             ) as any
@@ -218,7 +207,7 @@ export const dataValidatorMachine = setup({
             reply(
               ({ context, event }: { context: ValidationMachineContext; event: ErrorActorEvent }) =>
                 dataValidatorEvents.dataValidationFailed({
-                  validationPayload: toSerializable(context.validationPayload) as ValidationPayload,
+                  validationPayload: context.validationPayload as ValidationPayload,
                   systemErrors: event.error
                 })
             ) as any,

src/actors/worker/fromWorkerfiedActor.ts

@@ -35,6 +35,9 @@ export const fromWorkerfiedActor = (
   worker: Worker
 ): WorkerActorLogic<EventObject, WorkerInput> => ({
   config: Worker,
+  
+  
+  
 
   start: (state, actorScope) => {
     const { self, system } = actorScope;
@@ -46,23 +49,39 @@ export const fromWorkerfiedActor = (
     };
 
     worker.addEventListener('message', (event) => {
-      const eventFromWorker = event.data as AnyEventObject;
-      if (eventFromWorker.type == 'STATE_SNAPSHOT') {
+      console.log('Message received from worker --> ', event,event.data.type == WORKER_EVENTS.PROXY_EVENT,WORKER_EVENTS.PROXY_EVENT);
+      const eventPayload = event.data;
+      if (eventPayload.type === WORKER_EVENTS.STATE_SNAPSHOT) {
+        const eventFromWorker = eventPayload as STATE_SNAPSHOT_EVENT;
         self.send(eventFromWorker);
         return state;
       }
 
-      if (event.type === WORKER_EVENTS.PROXY_EVENT) {
-        console.log('Proxy event received from worker', eventFromWorker);
-        const proxyEvent = event as ProxyEvent;
-        if (proxyEvent.data.to === 'parent' && self._parent) {
+      if (eventPayload.type === WORKER_EVENTS.PROXY_EVENT) {
+    
+        const proxyEvent = eventPayload as ProxyEvent;
+        const targetActorId = proxyEvent.data.to;
+        const targetEvent = proxyEvent.data.event;
+        const isToParent = targetActorId === 'parent';
+        console.log('Proxy event received from worker to', targetActorId, targetEvent, isToParent);
+
+
+        if (isToParent && self._parent) {
           console.log('Relaying to parent', proxyEvent.data);
           self._parent.send(proxyEvent.data.event);
           return state;
         }
+        if (!isToParent) {
+          const targetActor = system.get(proxyEvent.data.to);
+          console.log('Relaying to system actor', proxyEvent.data, targetActor,system,self);
+          if (targetActor){
+            targetActor.send(proxyEvent.data.event);
+          }
+          return state;
+        }
+
+
 
-        system.get(proxyEvent.data.to).send(proxyEvent.data.event);
-        return state;
       }
     });
 
@@ -71,6 +90,8 @@ export const fromWorkerfiedActor = (
   transition: (state, event, actorScope) => {
     const { self } = actorScope;
     const workerState = instanceStates.get(self);
+    console.log('fromWorkerActor transition', state, event, actorScope);
+
     if (event.type === 'xstate.stop') {
       console.log('Stopping fromWorkerActor...', state, event, actorScope);
       workerState.worker.postMessage(workerCommands.stopActor());
@@ -81,6 +102,9 @@ export const fromWorkerfiedActor = (
         error: undefined
       };
     }
+
+
+
     if (event.type == WORKER_EVENTS.STATE_SNAPSHOT) {
       const snapshot = (event as STATE_SNAPSHOT_EVENT).data.snapshot as AnyMachineSnapshot;
       return {

src/actors/worker/workerfy.ts

@@ -26,7 +26,7 @@ const ProxyActor = setup({
       on: {
         '*': {
           actions: [
-            ({ event, context }) => console.log('Proxying event', event, 'to', context.proxyToId),
+            ({ event, context }) => console.log('Proxying actor event', event, 'to', context.proxyToId),
             ({ event, context }) => postMessage(workerEvents.proxyEvent(event, context.proxyToId))
           ]
         }
@@ -56,6 +56,8 @@ export const workerfyActor = (actor: AnyActorLogic) => {
     }
   }).start();
 
+  console.log('Worker actor initialized, waiting for commands...');
+
   addEventListener('message', (event) => {
     if (event.data.type === WORKER_COMMANDS.START_ACTOR) {
       actorRef = createActor(actor, {