Function fromObservable

packages/core/src/actors/observable.ts:118–223

(
  observableCreator: ({
    input,
    system,
    self
  }: {
    input: TInput;
    system: AnyActorSystem;
    self: ObservableActorRef<TContext>;
    emit: (emitted: TEmitted) => void;
  }) => Subscribable<TContext>
)

 * @see {@link Subscribable} interface in XState, which is based on and compatible with RxJS Observable.
 */
export function fromObservable<
  TContext,
  TInput extends NonReducibleUnknown,
  TEmitted extends EventObject = EventObject
>(
  observableCreator: ({
    input,
    system,
    self
  }: {
    input: TInput;
    system: AnyActorSystem;
    self: ObservableActorRef<TContext>;
    emit: (emitted: TEmitted) => void;
  }) => Subscribable<TContext>
): ObservableActorLogic<TContext, TInput, TEmitted> {
  // TODO: add event types
  const logic: ObservableActorLogic<TContext, TInput, TEmitted> = {
    config: observableCreator,
    transition: (snapshot, event) => {
      if (snapshot.status !== 'active') {
        return snapshot;
      }

      switch (event.type) {
        case XSTATE_OBSERVABLE_NEXT: {
          const newSnapshot = {
            ...snapshot,
            context: event.data as TContext
          };
          return newSnapshot;
        }
        case XSTATE_OBSERVABLE_ERROR:
          return {
            ...snapshot,
            status: 'error',
            error: (event as any).data,
            input: undefined,
            _subscription: undefined
          };
        case XSTATE_OBSERVABLE_COMPLETE:
          return {
            ...snapshot,
            status: 'done',
            input: undefined,
            _subscription: undefined
          };
        case XSTATE_STOP:
          snapshot._subscription!.unsubscribe();
          return {
            ...snapshot,
            status: 'stopped',
            input: undefined,
            _subscription: undefined
          };
        default:
          return snapshot;
      }