stream/iter: abort signal causes broadcast() and share() consumers to resolve EOF instead of rejecting #63357

import { broadcast, share } from 'node:stream/iter';

const outcome = (p) => p.then(
  (value) => ({ status: 'resolved', value }),
  (err) => ({ status: 'rejected', name: err.name, message: err.message }),
);

// broadcast({ signal })
{
  const ac = new AbortController();
  const { broadcast: bc } = broadcast({ signal: ac.signal });
  const pending = outcome(bc.push()[Symbol.asyncIterator]().next());

  ac.abort();

  console.log('broadcast:', await pending);
}

// share(source, { signal })
{
  const ac = new AbortController();
  const enc = new TextEncoder();

  async function* source() {
    yield [enc.encode('a')];
    yield [enc.encode('b')];
  }

  const shared = share(source(), {
    highWaterMark: 1,
    backpressure: 'block',
    signal: ac.signal,
  });

  const fast = shared.pull()[Symbol.asyncIterator]();
  shared.pull(); // Lagging consumer keeps the one-slot buffer full.

  await fast.next();
  const pending = outcome(fast.next());

  ac.abort();

  console.log('share:', await pending);
}