Confusion about spec

[RylinMirabel]

The callbag spec dictates that "termination should not be mutual". In the implementation of callbag-take, on line 15, when sink(t, d) is called and taken === max, if sink terminates its source (callback-take) immediately by calling talkback(2), mutual termination will occur after sink(2) is called on line 17. Does the callbag spec expect some degree of asynchronity (setTimeout, process.nextTick)?

[franciscotln]

take is an operator, meaning that it takes a number max and returns a function, which takes a callbag (the original source) and then returns a new function (a new callbag, this one will be the source in the sink's eyes).
The sink (like forEach) will subscribe to the takeSource and the take operator will do its magic, meaning that it will subscribe to the original source and act as a proxy.
When the new callbag source returned by take reaches the number max of emitted items it sends a 2 to the sink but the sink will never send a 2 upwards to terminate it.
That's the meaning of it :-)
from here we can see:

const take = max => source => function takeSource(start, sink) {
  if (start !== 0) return;
  let taken = 0;
  let sourceTalkback;
  function talkback(t, d) {
    if (taken < max) sourceTalkback(t, d);
  }
  source(0, (t, d) => {
    if (t === 0) {
      sourceTalkback = d;
      sink(0, talkback);
    } else if (t === 1) {
      if (taken < max) {
        taken++;
        sink(t, d);
        if (taken === max) {
          sink(2);
          sourceTalkback(2);
        }
      }
    } else {
      sink(t, d);
    }
  });
};

module.exports = take;

ex.:

pipe(
  fromIter('abcdef'),
  take(2),
  forEach(console.log) . // a - b|
);

[RylinMirabel]

Why is it the case that the sink must not terminate the source? Here is a case where the implementation of callbag-take leads to a double termination

[mperktold]

I agree with @RylinMirabel.

Maybe this situation rarely occurs in practice, but it is a bug; take should only terminate the sink if it hasn't been teminated yet itself.

[franciscotln]

Ok using take(2) followed by another take(2) isn't something that I considered because it's not a realistic case 😄 but in any case, this could be solved like so:
https://repl.it/repls/UntrueBisqueMice

adding a flag end: boolean

[RylinMirabel]

That still results in mutual termination between the two take callbags, although the downstream take callbag hides that from its sink. See the last two lines printed by the logger in this repl. To prevent mutual termination (which can lead to out-of-spec behaviour if unlike take, the downstream callbag does not disconnect its source from its sink upon termination), we need to check the end flag before sending any end signal (to either the source or the sink), and we need to set the end flag upon receiving any end signal. See this repl for a possible fix.

[StreetStrider]

it's not a realistic case

It can occur due to some composition.

[staltz]

I followed this thread and had my own thoughts, but you all figured it out, and I agree it's a bug in callbag-take, for which a PR has solved. The spec is quite clear that termination shouldn't be mutual, so I don't think we need this issue open. :)