push-take-until/index.js

/**
 * push-take-until - Take elements until signal stream emits
 *
 * Uses canonical push-stream pattern:
 * - This is a sink that wraps a signal stream
 * - When signal ends, the take-until ends
 * - Pass through all values until signal fires
 *
 * API:
 * import takeUntil from 'push-take-until'
 *
 * const untilClick = takeUntil(clickStream)
 * source.pipe(untilClick).pipe(sink)
 */

export function takeUntil(signalStream) {
  let source = null;
  let sink = null;
  let isEnded = false;
  let signalEnded = false;

  const stream = {
    get paused() { return false; },
    get ended() { return isEnded; },

    pipe(s) {
      if (isEnded) {
        s.end();
        return s;
      }
      sink = s;
      try { s.source = stream; } catch (e) {}
      return s;
    },

    write(data) {
      if (isEnded || signalEnded) return;
      if (sink && !sink.paused) {
        sink.write(data);
      }
    },

    end(err) {
      if (isEnded) return;
      isEnded = true;
      if (source && source.end) source.end(err);
      if (sink) sink.end(err);
    },

    abort(err) {
      if (isEnded) return;
      isEnded = true;
      if (source && source.abort) source.abort(err);
      if (sink) sink.end(err);
    },

    resume() {
      if (source && source.resume) source.resume();
    },

    get source() { return source; },
    set source(s) { source = s; }
  };

  signalStream.pipe({
    write() {},
    end() {
      if (signalEnded) return;
      signalEnded = true;

      if (!isEnded) {
        isEnded = true;
        if (source && source.end) source.end();
        if (sink) sink.end();
      }
    },
    abort(err) {
      if (signalEnded) return;
      signalEnded = true;

      if (!isEnded) {
        isEnded = true;
        if (source && source.abort) source.abort(err);
        if (sink) sink.end(err);
      }
    },
    get paused() { return false; },
    get ended() { return signalEnded; }
  });

  return stream;
}

export default takeUntil;