How to model a looped/recursive/circular observable?

[nandin-borjigin]

All the observables under discussion are hot observables.

                    ╭─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─╮
                    │                                                       │
                    ▼                                                       │
a$ ──── [ pipe(withLatestFrom(x$), ...) ] ────────┬─────── [ pipe(...) ] ───┴────▶ x$
                                                  |
                                                  ╰──────────────────────────────▶ m$

For a simple loop, I could use scan operator. But as demonstrated in the graph, I need to extract an intermediate observable m$, and I don't think it's feasible with scan operator.

I can think of a simple way using Subjects but it requires explicit subscription handling which doesn't seem to be an elegant and idiomatic way of using Rx.

function system(a$) {
  const x$ = new Subject()
  const m$ = a$.pipe(
    withLatestFrom(x$),
    ...,
    share()
  )
  m$.pipe(...).subscribe(x$) // <--- explicit subscription while we only want to describe a relationship graph between observables

  return {
    m$,
    x$
  }
}

[voliva]

You can use tap instead. It's still not elegant, but you will avoid that explicit subscription.

Also, I would keep the subject as something internal from your function. And don't forget to .startWith() your withLatestFrom, otherwise it will not get the first few elements through:

function system(a$) {
  const xSubject = new Subject()
  const m$ = a$.pipe(
    withLatestFrom(xSubject.pipe(startWith(...))),
    ...,
    share() // share or shareReplay? you'll need to think about this too...
  );
  const x$ = m$.pipe(..., tap(r => xSubject.next(r)))

  return {
    m$,
    x$
  }
}