Why the different Dispose Strategies in the Slice Sinks?

[semmel]

I think #428 might be related to this behaviour.

Curious Behaviour

slice, take, takeWhile (i.e. SliceSink and TakeWhileSink resp.) employ a SettableDisposable and dispose the source stream themselves if the stop condition is met. They also forward the Disposable to the run subscriber, to allow it to dispose the source on its own criteria.

On the other hand skipAfter performs no disposing of the source itself, but only forwards the source's Disposable to the run subscriber, whose responsibility it then is to call Disposable.dispose() on error or on end.

I could not trigger an unexpected behaviour from that. However it might be an issue in some edge cases?

Example

The example formulates a simple stream A -- X |, first using take, second using skipAfter to end.
Since SettableDisposable guards against double invocation, the source is disposed twice in the skipAfter implementation.

const 
  {runEffects, tap, merge, take, at, skipAfter, newStream} = require('@most/core'),
  {currentTime, newDefaultScheduler} = require('@most/scheduler'),
  sched = newDefaultScheduler();

// creates a simple one-valued stream which never ends
// and logs dispose() calls to the console
const valueAtNeverEnd = delay => value => newStream(
  (sink, scheduler) => {
    const publishVal = val => {
      sink.event(currentTime(scheduler), val);
    };
    const timer = setTimeout(publishVal, delay, value);
    return { dispose: () => { 
	  console.log(`disposing ${value} at ${delay} ms...`);
	  clearTimeout(timer); 
	} };
  }
);

const source = valueAtNeverEnd(500)("X");

// A -- X - 'disposing X at 500ms'
runEffects(tap(
  console.log, 
  merge(
    take(1, source), 
    at(100, "A")
  )
), sched);

// A -- X - 'disposing X at 500ms' - 'disposing X at 500ms'
runEffects(tap(
  console.log, 
  merge(
    skipAfter(v => v === "X", source), 
    at(100, "A")
  )
), sched);

So, why the different source dispose strategies in the slice operators?

[TylorS]

I'll come back later with something more detailed if no one else gets to it first, but off the top of my head, it's primarily to do with circularity. The SliceSink needs to be able to dispose of the source Stream when its bounds are met, but the source Stream needs the SliceSink to create its own Disposable, and SettableDisposable works to facilitate this circularity.

The differences between slice and skipAfter are semantic related, slice/skip/take manages when to accept or stop accepting events, and it's the best use of resources to dispose of the source stream after meeting the "take" boundary than to simply ignore the events. However that's exactly what skipAfter is intended to do, not dispose of the stream after meeting its own "take" boundary

[TylorS]

I found the original issue where skipAfter was introduced to most.js v1, there might be some useful context in there - cujojs/most#427

[semmel]

Ok, I see, so takeWhile(p) !≅ skipAfter(!p).
I mean takeWhile and skipAfter differ in more than just the negated predicate and in/exclusion of the last passed event.
Indeed it would not make sense to offer two functions for such a very similar purpose.

Perhaps one could add a little hint in the documentation? Because looking at the marble diagrams and the current description, takeWhile and skipAfter seem pretty similar.

I still have no idea why the difference though. The source stream gets disposed of in both cases at the same time. When skipAfter injects an end in the run sink, that in turn disposes the forwarded source stream. take injects the end in the run sink too, but disposes the source itself, any further dispose of the run sink is not propagated further up.

Can you construct an example where that extended source stream lifetime of skipAfter(p) manifests in a different stream outcome compared to takeWhile(!p)?

[TylorS]

Actually I think I'm a bit wrong about the semantic difference of things from my earlier post. After getting a chance to look at the code again, it's about inclusive vs exclusive of the matching predicat function. Imagine we have some source stream that iterates from 0 to Infinity

const source: Stream<number> = ... // 0-Inifinity over some time period

const takeUntilItems: number[] = []
pipe(
  source,
  M.takeUntil(x => x === 100),
  M.tap(x => takeUntilItems.push(x))
  _ => M.runEffects(_, scheduler),
)

const skipAfterItems: number[] = []
pipe(
  source,
  M.skipAfter(x => x === 100),
  M.tap(x => skipAfterItems.push(x))
  _ => M.runEffects(_, scheduler),
)

console.log(takeUntilItems) // [0..99]
console.log(skipAfterItems) // [0..100]