[Feature] Observables as Invocable

[ksaldana1]

Bug or feature request?

Feature

Description:

Right now invocable services supports Promises, Callbacks, and other Machines. I think it would be awesome if we could support Observables first class as well. I think there's a whole bunch of really cool things this could open up, as it would be a good place to put a lot of the trickier things in a nice pluggable/testable way (explicit cancellation without self-transitions, anything with time, retries, etc).

We will need to think about whether we support arbitrary Observables, or if we just focus on RxJS. I think the addition of RxJS as an effects system will make some things in this library slightly more approachable for people familiar with redux-like approaches.

(Feature) Potential implementation:

I'm just going to steal David's words here, since it's pretty much exactly what I want:

[michaelsbradleyjr]

I'd like to recommend a possible signature:

(ctx, e) => {...; return [childMessages$, (parentMessages$) => {...}];}

It's then possible to differentiate simplex and duplex services:

simplex — parent sends messages to child. Example: child represents a logger for parent; just as console.log never returns meaningful values, the child never sends messages back to parent, and therefore child$ is null.

[null, (parentMessages$) => {...}]

simplex — child sends messages to parent. Example: child represents a web server with server-sent events and parent is a client of that server; such events are one-way from server to client, therefore the right-hand side is null.

[childMessages$, null]

duplex — child sends messages to parent, and parent sends message to child. Example: child represents a websocket server or a webrtc peer and there is bidirectional communication between child and parent.

[childMessages$, (parentMessages$) => {...}]

---

Expanding on the signature, in the duplex case...

import { Subject } from "rxjs";

(ctx, e) => {
  const childMessages$ = new Subject();

  return [childMessages$, (parentMessages$) => {
    /* communicate to parent via childMessages$ .next(), .error(), .complete() */
  }];
}

---

I think it would be interesting if childMessages$ and/or parentMessages$ could be observable or async iterable. While it would be easy to detect which protocol the child presents to parent (Symbol.observable vs. Symbol.asyncIterator), there would have to be some service configuration that indicates to xstate which protocol the child wants presented by parent.

However, since it's possible to easily convert between observable and async iterable using IxJS, it's probably simpler for xstate to just work with observables and users can convert with Ix as needed.

[ksaldana1]

It's then possible to differentiate simplex and duplex services

Can you explain further what you mean here? Do you just mean this signature gives us the ability to represent these different types of services?

A more simplified signature that could achieve all 3 types of services you mentioned:

type InvokeObservableFn<TEvents, TContext> = (ctx$: Observable<TContext>, event$: Observable<TEvents>): Observable<TEvents | never>

The behavior then would be as described by David in the picture (XState subs and sends emitted events to parent). David suggested making a differentiation on whether the return value of the InvokeObservableFn is a Subject, but I'm not sure that's 100% necessary (except for more consistent API, no new property, etc).

import {
  tap,
  ignoreElements,
  distinctUntilChanged,
  mapTo,
  mergeMap,
  delay
} from "rxjs/operators";
import { of } from "rxjs";

const simplexLogger = (_ctx$, event$) =>
  event$.pipe(
    tap(console.log),
    ignoreElements()
  );
const contextChangeLogger$ = (ctx$, _event$) =>
  ctx$.pipe(
    distinctUntilChanged(),
    tap(console.log),
    ignoreElements()
  );

const simplexChild = (_ctx$, _event$) =>
  fakeServer$.pipe(mapTo({ type: "SOME_EVENT" }));

const duplexSyncServerCtx$ = (ctx$, _event$) =>
  ctx$.pipe(mergeMap(ctx => fakeServer$.pipe(mapTo({ type: "SEND_TO_PARENT" }))));

const fakeServer$ = of({ type: "SERVER_EVENT" }).pipe(delay(2000));

At this point everything becomes an RxJS problem.... which may not be what everyone wants. I am definitely shaped with my positive experience with redux-observable, and I have a lot of confidence in my ability to model complex scenarios in this model. I personally would start pulling lots of things being handled by the xstate interpreter/runtime to RxJS (time, merging, cancellation, etc). Totally personal preference though—just giving my two cents. I may not have captured all your requirements (especially with the duplex—though I don't see what's stopping you from subscribing to a piped subject and cleaning up in a finalize operator in the body of the function), so let me know... been a bit since I've worked with RxJS so may be a bit rusty 😃

[michaelsbradleyjr]

It's then possible to differentiate simplex and duplex services

I didn't put that well. I do think it's possible to achieve the same (or at least very similar) result by looking at the type of the return value of (ctx, e) => {...}.

I think the tuple approach I outlined will be more clear, though, in a plain JS context and possibly even in a TS context. It feels more declarative to me, though I'm probably biased since I came up with it and have been conducting thought experiments with that shape. 😄

Consider that in a plain JS context, using the return value approach, it wouldn't be easy at runtime to recognize the "simplex — child sends messages to parent" case (short of toString()'ing the (ctx, e) => {...} function and checking the parameters substring is actually "(ctx)" or "()", which is probably not a good approach to adopt), and that could lead to a parentMessages$ object (however it's named) being instantiated unnecessarily.

In the return value approach, I also wonder if "Subject detection" (at compile or runtime) is a bit too wedded to RxJS per se. In my tuple approach, though Rx Subject would be a very practical choice for childMessages$, the signature is quite agnostic re: RxJS-the-library. One or both members of the tuple need only present the Symbol.observable protocol, else be null.

---

I'm rather new to xstate... or at least to actually digging into it. I read Harel's STATEMATE book and some other papers several years ago and was deeply impressed, even took a stab at writing a statecharts pseudo-scxml interpreter in JS, but never got very far because of time constraints. The point being that I may be misunderstanding some aspects of how services work in xstate 4.x, and I apologize if I'm way off base.

For example, in the signature I proposed, the relationship of e in (ctx, e) => {...} is rather under-defined in relation to the (parentMessages$) => {...} part.

It may be that, if we sub e$ for parentMessages$, the generalized forms are:

duplex

(ctx) => {const childMessages$ = ...; return [childMessages$, (e$) => {...}];}

simplex

(ctx) => {const childMessages$ = ...; return [childMessages$, null];}

(ctx) => [null, (e$) => {...}]

It would be up to the interpreter to recognize that the (ctx[, e]) function returned a tuple/array and call the non-null RHS with e$.

---

Notwithstanding what I wrote about IxJS in #435 (comment), I still wonder if there are, at runtime with respect to statechart interpreter behavior, irreducible differences between push and pull interfaces, i.e. between child/parent presenting observable vs. asyncIterator protocol to parent/child.

[ksaldana1]

I am most definitely biased towards my experience with redux-observable, TypeScript and RxJS—so I don't want to act like my proposal is coming from some perfectly rational place either. In redux-observable they have a type signature of:

(action$: Observable<Actions>, state$: Observable<State>, dependencies: Dependencies): Observable<Actions>;

That's kinda what I'm shooting for here, except we have event$ and context$ streams.

In the return value approach, I also wonder if "Subject detection" (at compile or runtime) is a bit too wedded to RxJS per se

David mentioned subject detection in his original proposal, but I don't think that's necessary with what I'm suggesting. Making the arguments of the function themselves observable makes the subject detection unnecessary—xstate would just need to keep internal observables for context and events and listen to the return values of the service observables to feed back into the parent machines. XState is "indirectly" passing onNext to those services by maintaining these observables consumed by services.

I definitely am with you that this might be a too RxJS-centric solution. I also am so deep into the TS ecosystem that I've pretty much lost empathy for JS API design... so definitely good to check me on that. My initial gut reaction to your API is that it's unfamiliar to me, so my natural reaction is going to be hesitation. Do you have any examples of other libraries using tuple-like approaches with observables? Particularly in some bidi streaming case?

[davidkpiano]

I'll be honest, this is a good place to start:

invoke: {
  src: (ctx, e) => {
    const mousemove$ = fromEvent(e.target, 'mousemove');

    return mousemove$;
  }
}

Let's keep it simple for now, and discuss duplex/bidi streams after it goes in?

With the current XState, you can already do bidirectional within a callback, and the code needed to do so is pretty non-intimidating and terse:

invoke: {
  src: (ctx, e) => (cb, receive) => {
    const event$ = fromEventPattern(receive);

    const sub = event$
      .pipe(debounceTime(1000), map(format))
      .subscribe(cb);

    return sub.unsubscribe;
  }
}

[ksaldana1]

I am totally good with the API you initially proposed. It serves my 90% use case off the top of my head—leaning too hard into Observables everywhere is very opinionated and I don't expect other people to want that as much as I do. We can always iterate as we figure out the remaining 10%.

[davidkpiano]

Yep, that last 10% would just be removing the (cb, receive) part for duplex Observables.

[michaelsbradleyjr]

Since xstate's support for promises cleanly maps resolve and reject to onDone and onError, it would be really nice if there was similar support for observers' next, complete, and error (relative to child observable) to map to onNext, onDone, and onError in the parent machine.

While the callback pattern is quite flexible, it also seems a bit more ad hoc. The familiar NodeJS pattern of (err, value) => {...} could be adopted and then truthy 1st argument passed to callback could trigger onError, like promise's reject. But how to signal "done"?

Changing the callback API that's in place now probably isn't a good idea, I'm just working through some ideas. Also, the callback is more like a listener that's used with an event emitter's .on(<event>, <callback>) and those callbacks aren't error-first. I realize you can setup whatever on: {} is needed to support signaling of errors and completion via callback, but that's what I meant by ad hoc.

[davidkpiano]

Since xstate's support for promises cleanly maps resolve and reject to onDone and onError, it would be really nice if there was similar support for observers' next, complete, and error (relative to child observable) to map to onNext, onDone, and onError in the parent machine.

• For onNext it's expected that invoked observables emit events, otherwise we'd have to make ad-hoc events. These are treated like any other event.
• onError works as expected with observables.
• onDone works as expected with completed observables, as well.

Closing this as observables are now in 4.6 and invocable 🎉 We can open a new issue for duplex observables if needed.

[ksaldana1]

Thanks, @davidkpiano! Very excited about this recent release!