[v5] Explicit spawn

packages/core/src/ObservableActorRef.ts

@@ -1,18 +1,22 @@
-import { EventObject, ActorRef, Behavior, SCXML, ActorContext } from './types';
+import {
+  EventObject,
+  ActorRef,
+  Behavior,
+  SCXML,
+  ActorContext,
+  Observer
+} from './types';
 import { symbolObservable, toSCXMLEvent } from './utils';
 import { Mailbox } from './Mailbox';
 import { LifecycleSignal, startSignal, stopSignal } from './actors';
 
-const nullSubscription = {
-  unsubscribe: () => void 0
-};
-
 export class ObservableActorRef<TEvent extends EventObject, TSnapshot>
   implements ActorRef<TEvent, TSnapshot> {
   private current: TSnapshot;
   public deferred = true;
   private context: ActorContext<TEvent, TSnapshot>;
   private mailbox = new Mailbox(this._process.bind(this));
+  private _observers = new Set<Observer<TSnapshot>>();
 
   constructor(
     private behavior: Behavior<TEvent, TSnapshot>,
@@ -21,7 +25,7 @@ export class ObservableActorRef<TEvent extends EventObject, TSnapshot>
     this.context = {
       self: this,
       name: this.name,
-      observers: new Set(),
+      observers: this._observers,
       _event: toSCXMLEvent({ type: 'xstate.init' }) as SCXML.Event<TEvent> // TODO: fix
     };
     this.current = this.behavior.initialState;
@@ -39,7 +43,13 @@ export class ObservableActorRef<TEvent extends EventObject, TSnapshot>
     return this;
   }
   public subscribe(observer) {
-    return this.behavior.subscribe?.(observer) ?? nullSubscription;
+    this._observers.add(observer);
+
+    return {
+      unsubscribe: () => {
+        this._observers.delete(observer);
+      }
+    };
   }
   public send(event) {
     this.receive(event);

packages/core/src/actors.ts

@@ -91,32 +91,19 @@ export function fromCallback<TEvent extends EventObject>(
 ): Behavior<TEvent, undefined> {
   let canceled = false;
   const receivers = new Set<(e: EventObject) => void>();
-  const observers: Set<Observer<TODO>> = new Set();
   let dispose;
 
-  const sendNext = (event: TODO) => {
-    observers.forEach((o) => o.next?.(event));
-  };
-
-  const sendError = (event: TODO) => {
-    observers.forEach((o) => o.error?.(event));
-  };
-
-  const sendComplete = () => {
-    observers.forEach((o) => o.complete?.());
-  };
-
   const behavior: Behavior<TEvent, undefined> = {
-    transition: (_, event, actorContext) => {
-      const { _parent: parent } = actorContext.self;
+    transition: (_, event, { self, observers }) => {
+      const { _parent: parent } = self;
 
       if (event.type === startSignalType) {
         const sender: Sender<TEvent> = (e) => {
           if (canceled) {
             return;
           }
 
-          parent?.send(toSCXMLEvent(e, { origin: actorContext.self }));
+          parent?.send(toSCXMLEvent(e, { origin: self }));
         };
 
         const receiver: Receiver<TEvent> = (newListener) => {
@@ -128,12 +115,14 @@ export function fromCallback<TEvent extends EventObject>(
         if (isPromiseLike(dispose)) {
           dispose.then(
             (resolved) => {
-              sendNext(resolved);
-              sendComplete();
+              observers.forEach((o) => o.next?.(resolved));
+              observers.forEach((o) => o.complete?.());
+
               canceled = true;
             },
             (errorData) => {
-              sendError(errorData);
+              observers.forEach((o) => o.error?.(errorData));
+
               canceled = true;
             }
           );
@@ -158,15 +147,6 @@ export function fromCallback<TEvent extends EventObject>(
 
       return undefined;
     },
-    subscribe: (observer) => {
-      observers.add(observer);
-
-      return {
-        unsubscribe: () => {
-          observers.delete(observer);
-        }
-      };
-    },
     initialState: undefined
   };
 
@@ -177,10 +157,9 @@ export function fromPromise<T, TEvent extends EventObject>(
   lazyPromise: Lazy<PromiseLike<T>>
 ): Behavior<{ type: string }, T | undefined> {
   let canceled = false;
-  const observers: Set<Observer<T>> = new Set();
 
   const behavior: Behavior<TEvent, T | undefined> = {
-    transition: (_, event) => {
+    transition: (_, event, { observers }) => {
       switch (event.type) {
         case startSignalType:
           const resolvedPromise = Promise.resolve(lazyPromise());
@@ -213,15 +192,6 @@ export function fromPromise<T, TEvent extends EventObject>(
           return undefined;
       }
     },
-    subscribe: (observer) => {
-      observers.add(observer);
-
-      return {
-        unsubscribe: () => {
-          observers.delete(observer);
-        }
-      };
-    },
     initialState: undefined
   };
 
@@ -233,31 +203,20 @@ export function fromObservable<T, TEvent extends EventObject>(
 ): Behavior<TEvent, T | undefined> {
   let subscription: Subscription | undefined;
   let observable: Subscribable<T> | undefined;
-  const observers: Set<Observer<TODO>> = new Set();
-  const sendNext = (event: TODO) => {
-    observers.forEach((o) => o.next?.(event));
-  };
-
-  const sendError = (event: TODO) => {
-    observers.forEach((o) => o.error?.(event));
-  };
-  const sendComplete = () => {
-    observers.forEach((o) => o.complete?.());
-  };
 
   const behavior: Behavior<TEvent, T | undefined> = {
-    transition: (_, event) => {
+    transition: (_, event, { observers }) => {
       if (event.type === startSignalType) {
         observable = lazyObservable();
         subscription = observable.subscribe({
           next: (value) => {
-            sendNext(value);
+            observers.forEach((o) => o.next?.(value));
           },
           error: (err) => {
-            sendError(err);
+            observers.forEach((o) => o.error?.(err));
           },
           complete: () => {
-            sendComplete();
+            observers.forEach((o) => o.complete?.());
           }
         });
       } else if (event.type === stopSignalType) {
@@ -266,15 +225,6 @@ export function fromObservable<T, TEvent extends EventObject>(
 
       return undefined;
     },
-    subscribe: (observer) => {
-      observers.add(observer);
-
-      return {
-        unsubscribe: () => {
-          observers.delete(observer);
-        }
-      };
-    },
     initialState: undefined
   };
 
@@ -294,30 +244,22 @@ export function fromEventObservable<T extends EventObject>(
 ): Behavior<EventObject, T | undefined> {
   let subscription: Subscription | undefined;
   let observable: Subscribable<T> | undefined;
-  const observers: Set<Observer<TODO>> = new Set();
-
-  const sendError = (event: TODO) => {
-    observers.forEach((o) => o.error?.(event));
-  };
-  const sendComplete = () => {
-    observers.forEach((o) => o.complete?.());
-  };
 
   const behavior: Behavior<EventObject, T | undefined> = {
-    transition: (_, event, actorContext) => {
-      const { _parent: parent } = actorContext.self;
+    transition: (_, event, { self, observers }) => {
+      const { _parent: parent } = self;
 
       if (event.type === startSignalType) {
         observable = lazyObservable();
         subscription = observable.subscribe({
           next: (value) => {
-            parent?.send(toSCXMLEvent(value, { origin: actorContext.self }));
+            parent?.send(toSCXMLEvent(value, { origin: self }));
           },
           error: (err) => {
-            sendError(err);
+            observers.forEach((o) => o.error?.(err));
           },
           complete: () => {
-            sendComplete();
+            observers.forEach((o) => o.complete?.());
           }
         });
       } else if (event.type === stopSignalType) {
@@ -326,15 +268,6 @@ export function fromEventObservable<T extends EventObject>(
 
       return undefined;
     },
-    subscribe: (observer) => {
-      observers.add(observer);
-
-      return {
-        unsubscribe: () => {
-          observers.delete(observer);
-        }
-      };
-    },
     initialState: undefined
   };
 
@@ -405,9 +338,6 @@ export function fromMachine<TMachine extends AnyStateMachine>(
       service?.send(_event);
       return state;
     },
-    subscribe: (observer) => {
-      return service?.subscribe(observer);
-    },
     get initialState() {
       // TODO: recheck if this caching is needed, write a test for its importance or remove the caching
       if (initialState) {

packages/core/src/types.ts

@@ -1867,7 +1867,6 @@ export interface Behavior<TEvent extends EventObject, TSnapshot = any> {
   ) => TSnapshot;
   initialState: TSnapshot;
   start?: (actorCtx: ActorContext<TEvent, TSnapshot>) => TSnapshot;
-  subscribe?: (observer: Observer<TSnapshot>) => Subscription | undefined;
 }
 
 export type EmittedFrom<T> = ReturnTypeOrValue<T> extends infer R