Design problems with async stop

[staltz]

Because of issues cyclejs/cyclejs#365 and #90, I made a fix to xstream which gives special treatment to a corner case in flatten, which was released in v5.3.2. 819bc94

However, (1) it doesn't actually fix the problem, (2) it conflicts with previously desired features. It also makes new bugs appear such as #103.

(1)

Note how we added this test, which passed in v5.3.2:

    it('should restart inner stream if switching to the same inner stream', (done) => {
      const outer = fromDiagram('-A---------B----------C--------|');
      const nums = fromDiagram(  '-a-b-c-----------------------|', {
        values: {a: 1, b: 2, c: 3}
      });
      const inner = nums.fold((acc, x) => acc + x, 0);

      const stream = outer.map(() => inner).flatten();

      const expected = [0, 1, 3, 6, 0, 1, 3, 6, 0, 1, 3, 6];

      stream.addListener({
        next: (x: number) => {
          assert.equal(x, expected.shift());
        },
        error: (err: any) => done(err),
        complete: () => {
          assert.equal(expected.length, 0);
          done();
        }
      });
    });

However, this does not pass in v5.3.2:

    it('should restart inner stream if switching to the same inner stream', (done) => {
      const outer = fromDiagram('-A---------B----------C--------|');
      const nums = fromDiagram(  '-a-b-c-----------------------|', {
        values: {a: 1, b: 2, c: 3}
      });
      const inner = nums.fold((acc, x) => acc + x, 0);

-     const stream = outer.map(() => inner).flatten();
+     const stream = outer.map(() => inner.map(x => x)).flatten();

      const expected = [0, 1, 3, 6, 0, 1, 3, 6, 0, 1, 3, 6];

      stream.addListener({
        next: (x: number) => {
          assert.equal(x, expected.shift());
        },
        error: (err: any) => done(err),
        complete: () => {
          assert.equal(expected.length, 0);
          done();
        }
      });
    });

Because every time the function () => inner.map(x => x) is called, inner.map(x => x) will yield a different stream, where as () => inner would always yield inner as the same stream.

(2)

Sync start and async stop was designed to allow the inner stream to not restart if it was the same during the switch in a flatten. This was really by design, to avoid some confusion with a common pattern we had in Cycle.js, the use of RxJS connect() here: cyclejs/cyclejs@67d176e#diff-32a0a3abed94d032137ef603ee4dd261L30

So "don't restart the inner stream if it remains the same during flatten" is a feature by design.

However, to have referential transparency we want these two cases to give the same behavior:

    const inc$ = sources.DOM.select('.inc').events('click').mapTo(+1);
    const refresh$ = sources.DOM.select('.ref').events('click').startWith(0);
+   const sum$ = inc$.fold((x, y) => x + y, 0);
+   const lastSum$ = refresh$.map(_ => sum$).flatten();
-   const lastSum$ = refresh$.map(_ => inc$.fold((x, y) => x + y, 0)).flatten();
    const vdom$ = lastSum$.map(count =>
      div([

Which means we want the property "restart the inner stream if it remains the same during flatten" by design.

Which means we have a conflict, and we need to choose which of these two to do. It may mean a breaking change. I had hopes sync start and async stop would make things more intuitive but there is an obvious drawback that makes xstream less intuitive.

If you're reading this thread, please give your friendly and thoughtful opinion on this topic. This appears to be my design mistake, but I'm just a human. What's important is that I'm willing to look for a better way forward. My intent with sync start and async stop was to provide an "just works" experience for most cases, and together with Tylor we did a lot of predictions and bike-shedding, but a corner case slipped out of our sight.

For now, I'll revert the bugfix that happened in v5.3.2, so that other issues don't surface, and to keep a consistent behavior.

[staltz]

cc @TylorS @Hypnosphi @whitecolor @Widdershin @ntilwalli @paldepind

[Hypnosphi]

I liked the non-restarting behavior more. The restarting lowers the streams temperature, and referential transparency isn't really compatible with hotness

[wclr]

Is it possible to remove setTimeout (async stop) and save flatten behaviour (subscribe to new stream, then unsubscribe from old - synchronously)?

[Hypnosphi]

Btw, the same behaviour can be achieved without async stop: switching over addition and removal lines might be enough: https://github.com/staltz/xstream/blob/master/src/core.ts#L649,L650

[ntilwalli]

I personally prefer maintaining the hot/cold distinctions, but I feel like the philosophy of xstream is to prefer hot-always behavior (non-restarting) to referential transparency, so I'd vote for that.

If non-restarting wins out, what would this code print out?

import xs from 'xstream'
const a$ = xs.of(1, 2, 4).remember()
const b$ = xs.periodic(1000).map(() => a$).flatten()
b$.addListener({
  next: (x) => {bin.log(x)},
  error: () => {},
  complete: () => {}
})

[Hypnosphi]

it will be 1,2,3 each second anyway. Enforced restarting is irrelevant here, because a$ is already stopped at the moment of switching.

[staltz]

There are cases where non-restarting makes a lot of sense, and where restarting makes more sense. Also, as a reminder, non-restarting behavior can also happen in flatten even when switching to different streams.

const a$ = // ...
const b$ = // ...
a$.addListener(l1);
b$.addListener(l2);
const c$ = xs.periodic(1000).map(i => {
  if (i % 2 === 0) {
    return a$;
  } else {
    return b$;
  }
}).flatten();

Neither a$ nor b$ will restart when the switch happens in flatten, because these already have listeners, so they are executing no matter what.

One way of looking at this behavior of map+flatten is map(x => a$).flatten() means "map x to the current execution of a$". Only if there is no current execution of a$ will flatten force the (re)start of an execution of a$.

@ntilwalli I personally prefer maintaining the hot/cold distinctions, but I feel like the philosophy of xstream is to prefer hot-always behavior (non-restarting) to referential transparency, so I'd vote for that.

One way we can keep this distinction is to add xs.fromObservable so you can provide a cold Rx Observable or most stream, and then it returns a hot xstream stream. This is how we can compose restart-tricky behaviors in a ref transparent world and then convert to the hot world.

That example code from @ntilwalli reminds me of issue #91, which I'm working on in another branch and although interesting, has not much to do with this issue. I'll submit that PR soon for discussion.

@whitecolor

Is it possible to remove setTimeout (async stop) and save flatten behaviour (subscribe to new stream, then unsubscribe from old - synchronously)?

Even if we do remove async stop, we still need to choose between restarting or non-restarting semantics.

[wclr]

Even if we do remove async stop, we still need to choose between restarting or non-restarting semantics.

non-restarting would make things more other stream libs like (rx and most). non-restrting is really more flexible, but requires more understanding.

In this case non-restarting behaviour could be modeled using such tricky trick:

const a$ = // ...
const b$ = // ...
const c$ = xs.merge(a$.drop(), b$.drop(), xs.periodic(1000)).map(i => {
  if (i % 2 === 0) {
    return a$;
  } else {
    return b$;
  }
}).flatten();

where restarting/no-restarting will play else beside flatten?

[staltz]

where restarting/no-restarting will play else beside flatten?

when synchronously removing last listener and adding a new listener, but this use case is so rare that I'd say the main use case is flatten.

[geovanisouza92]

Maybe I'm too noob to opinate on this, but, does it make sense to make flatten() accept a "restart" parameter, assuming the true/false to the most common case? Maybe this could introduce a non-intuitive option and turn the things most confusing that already are...

[staltz]

Everything is possible with options, but it's important (specially for xstream) to have smart defaults and "just works" behavior.

[wclr]

Options is not needed here of course. Need to list and consider all pros and cons non-restarting/restarting brings to the table.

[staltz]

After thinking about this for a few days, maybe settling with non-restarting is better, while accepting lack of referential transparency.

restarting is incompatible with non-restarting, but restarting is also incompatible with use cases of multiple listeners, see below:

const sum$ = inc$.fold((x, y) => x + y, 0);
sum$.addListener(/* ... */);
const lastSum$ = refresh$.map(_ => sum$).flatten();
// not the same as:
const lastSum$ = refresh$.map(_ => inc$.fold((x, y) => x + y, 0)).flatten();

// because the sum$ will not restart since it always 
// has the listener on the second line.
// While we keep the guarantee that each stream has only one execution.

I think the best way to resolve this is keep non-restarting behavior consistent, and then add fromObservable to allow composing streams in the cold world of RxJS or most.js, then convert to the hot world in xstream when we want to.

Please 👍 if you agree. Comment if you don't.

[wclr]

@staltz also it would be better to get rid of setTimeout, I think it would make easier potential integration of scheduling.

[staltz]

Okay, let's close this because the conclusion is to implement #7.