nodejs-streams-operators.mjs

// @erickwendel_
import { Readable } from 'node:stream'
import { pipeline } from 'node:stream/promises'

const take = (limit) => async function* (source) {
  let count = 0;
  for await (const item of source) {
    if (count++ >= limit) break
    yield item
  }
}

const filter = (fn) => async function* (source) {
  for await (const item of source)
    if (fn(item)) yield item
}

const map = (fn) => async function* (source) {
  for await (const item of source) yield fn(item)
}

const log = _ => async function* (source) {
  for await (const item of source) console.log(item)
}


await pipeline(
  Readable.from(['🤯', '😬', '😂', '😳', '😳']),
  take(3),
  filter(item => item !== '😬'),
  map(item => 'emoji: '.concat(item)),
  log(),
)
/*
emoji: 🤯
emoji: 😂
*/

console.log()

const readable = 
  Readable.from(['🤯', '😬', '😂', '😳', '😳'])
    .take(3)
    .filter(item => item !== '😬')
    .map(item => 'emoji: '.concat(item))

await pipeline(
  readable,
  log()
)
/*
emoji: 🤯
emoji: 😂
*/