/** * 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;