Q25: Build a Custom Stream Transform Pipeline Component

Create a data stream processor that processes stream chunks sequentially through composed async transform middleware, allowing chunks to be modified or filtered out.

Transform Stream Pipeline

class DataPipeline {
  constructor() {
    this.transforms = [];
  }

  use(transformFn) {
    this.transforms.push(transformFn);
    return this;
  }

  async execute(sourceDataStream) {
    const results = [];
    for (let chunk of sourceDataStream) {
      let currentData = chunk;
      let skipped = false;
      
      for (let transform of this.transforms) {
        currentData = await transform(currentData);
        if (currentData === null || currentData === undefined) {
          skipped = true;
          break;
        }
      }
      
      if (!skipped) {
        results.push(currentData);
      }
    }
    return results;
  }
}

const pipeline = new DataPipeline()
  .use(async x => x * 2)
  .use(async x => x > 5 ? x : null);

pipeline.execute([1, 2, 4]).then(console.log); // [8]

💡 Applies sequential transforms to each chunk. Returning null inside any transform halts the middleware chain and drops the chunk.

Key Takeaways

Pipes individual stream chunks through sequentially composed transformation middleware.
Supports filtering chunks from the stream when transform functions return null or undefined.

Knowledge Check

What happens when a transformation function in the pipeline returns null for a specific chunk?

Q25: Build a Custom Stream Transform Pipeline Component

STAY UPDATED WITH BYTES!