Late-subscribing only works if early subscribing occurred

[bjnsn]

When a remembered source receives a value, that value is not emitted to late subscribers unless that source has already been subscribed to.

Perhaps this is the intended behavior, but to me this was unexpected. See: https://stackblitz.com/edit/js-t8rvxy?file=index.js

import remember from "callbag-remember";
import { pipe, forEach } from "callbag-basics";
import fromFunction from "callbag-from-function";

const { source, emitter } = fromFunction();

const rem$ = remember(source);

// Late subscription does not receive the value unless early subscription enabled
/*
pipe(
  rem$,
  forEach(value => {
    console.log("early-subscribe", value);
  })
);
*/

emitter(111);

setTimeout(() => {
  pipe(
    rem$,
    forEach(value => {
      console.log("late-subscribe", value);
    })
  );
}, 1000);

[Andarist]

Could you compare the current behavior with Rx's .shareReplay(1) and Cycle's .remember()? We don't know if the input stream is cold or hot and I assume a cold one - but maybe remember should automatically create a hot observable? From what I remember Cycle defaults to hot observables while most things in Rx are cold - so there still might be behavioral differences between both.

[bjnsn]

Here it is in RX - behavior is the same as in callbag-remember:

https://stackblitz.com/edit/typescript-nj75gu

import { Subject } from "rxjs";
import { shareReplay } from "rxjs/operators";

const subj = new Subject<number>();

const rem$ = subj.pipe(shareReplay(1));

// Late subscription does not receive the value unless early subscription enabled
// rem$.subscribe(v => console.log("early", v));

subj.next(111);

rem$.subscribe(v => console.log("late", v));

Here it is in XStream - behavior differs from callbag-remember:

https://stackblitz.com/edit/typescript-bsgrmr

import xs from "xstream";

const rem$ = xs
  .fromPromise(
    new Promise<number>(resolve => {
      setTimeout(() => resolve(111), 100);
    })
  )
  .remember();

// Late subscription receives the value regardless of early subscription
// rem$.subscribe({next: (v) => console.log('early', v)});

setTimeout(() => {
  rem$.subscribe({ next: v => console.log("late", v) });
}, 500);

[Andarist]

@staltz what's the general guidance for things like this in Callbags? Should most Callbags default to being cold? If 2 variants of a similar operator are needed what should one do? Maybe just use a casting~ operator that takes a cold source and makes it hot?

[staltz]

@Andarist Sorry it took a while to answer.

I think the behavior for most Callbags should be cold by default, like RxJS does, not like xstream does. This is simply because cold is less opinionated: you can always get a cold source and make it hot, but the opposite is not true (for hot to cold). xstream is hot by default, and while it makes some use cases simple and easy, there are certain use cases that are weird or just broken (mathematical reasoning stops applying).

Since this operator was named remember in resemblance to xstream (I'm assuming), I'd recommend taking a look how RxJS does it. It indeed has a separate operator exclusively for making a cold source become hot, it's called multicast. On top of that it has extra operators that enable different kinds of hot behaviors.

In short:

• Subject: inherits Observable and Observer, acts as a "GroupedObserver" that can have child observers, this is what makes multicast possible
• multicast: puts a Subject in the pipeline, effectively multicasting the source but without exposing the naked Subject, but exposes some special API to allow you to control when the Subject is plugged/unplugged from the source
• refCount: does automatic plugging/unplugging (see above) when the first observer arrives and when the last observer leaves
• share: multicast(new Subject()) + refCount
• shareReplay: multicast(new ReplaySubject()) + refCount
• shareBehavior: multicast(new BehaviorSubject()) + refCount

So I would understand callbag-remember to be similar to shareBehavior, because that's what I modeled xstream's remember after. Given that callbag is a highly modular ecosystem, I'd recommend splitting up the functionalities of multicast, refcounting, and then allowing to bundle them into packages such as callbag-share-replay etc.

[Andarist]

No worries - that was not a late answer by any means!

Thank you for the detailed answer. In the past, I've actually wanted to break up things into multicasting-related but never got to it and I won't be able to find the time for it. It would be very nice to have but from this operator's perspective, this would also be an implementation detail. So I think we should be able to adjust the behavior to conform with xstream's remember and maybe swap out the implementation in the future when things get more modular.

@bjnsn would you like to work on the required adjustments?

[bjnsn]

I'll look into it! I'm relatively new to RX & Callbags so I'm not 100% sure I understand all the nuances - I'll definitely rely on you validate.

A few questions:

• I'm a bit unclear about what is being encompassed by the term hot. I know that it means multicast - but in this context it appears to also mean that the callbag is immediately active (before any subscriptions). That, as I understand it, is the distinction between the current implementation of callbag-remember and XStream remember.
• As I understand it the distinction between shareReplay and shareBehavior is that the later accepts an initial seed (I ask because I'm not seeing shareBehavior in the most recent RXJS docs).

The roadmap would be to split functionality into new packages callbag-multicast and callbag-refcount which could then be composed with existing packages callbag-subject and callbag-behavior-subject (and potentially another new package callbag-replay-subject)

[Andarist]

I'm a bit unclear about what is being encompassed by the term hot. I know that it means multicast - but in this context it appears to also mean that the callbag is immediately active (before any subscriptions). That, as I understand it, is the distinction between the current implementation of callbag-remember and XStream remember.

Yep, exactly. Consider a simple example:

const timer$ = interval(1000)

This is a cold observable because that interval operator doesn't actually do anything until it's being subscribed. If it would start a timer immediately then we would consider this a hot observable. With cold observables, you always subscribe to a "template" of some behavior so:

• your subscription always starts from a "zero" point of that behavior - in this case, you'd always get 0 as the first received value (unless something in the operator sequence changes the behavior)
• if you subscribe to that with two separate listeners then each one of them will actually kick off a separate interval (unless something changes that in the sequence 😉 )

As I understand it the distinction between shareReplay and shareBehavior is that the later accepts an initial seed (I ask because I'm not seeing shareBehavior in the most recent RXJS docs).

I'm actually not super-proficient in those more complicated RxJS operators - so I often recheck the docs for those kinds of things. As you have noticed - some of those don't exist anymore in Rx so it's a little bit more convoluted for me to validate but you are probably correct. Rx has some excellent tests for their operators - although you need to understand how they are written first - and those always serve as docs for me when it comes to some more delicate behaviors that I want to understand/replicate. I was able to dig up some old tests for those two operators - you can check them here:
ReactiveX/rxjs@d0adf57

The roadmap would be to split functionality into new packages callbag-multicast and callbag-refcount which could then be composed with existing packages callbag-subject and callbag-behavior-subject (and potentially another new package callbag-replay-subject)

Yep, that would be possible and would be a great addition to the callbag family.