Complex Event Stream Filtering via Async Generator Control Flow

async function* filterStream(source, predicate) {
  for await (const item of source) {
    if (await predicate(item)) {
      yield item;
    }
  }
}
 
const numbers = async function* () { yield 5; yield 12; yield 8; }();
const evenStream = filterStream(numbers, async (x) => x % 2 === 0);
 
for await (const val of evenStream) {
  console.log(val);
}