node async Writable or Transform stream

stream-examples.ts

/* eslint-disable @typescript-eslint/no-misused-promises */
import { Readable, Transform, Writable } from 'node:stream'
import { pipeline } from 'node:stream/promises'

export const asyncHandler = (handle: (value: any) => Promise<void>) => {
  return new Writable({
    objectMode: true,
    write: async (value, _, next) => {
      try {
        await handle(value)
        next()
      } catch (error) {
        next(error)
      }
    },
  })
}

export const asyncTransform = (convert: (value: any) => Promise<any>) => {
  return new Transform({
    objectMode: true,
    transform: async (value, _, callback) => {
      try {
        callback(null, await convert(value))
      } catch (error) {
        callback(error)
      }
    },
  })
}

const run = async () => {
  const readable = Readable.from(['foo', 'bar', 'baz'])
  try {
    await pipeline(
      readable,
      asyncTransform(async (value) => String(value).toUpperCase()),
      asyncHandler(async (value) => console.log(value))
    )
  } catch (error) {
    console.error(error)
  }
}

void run()