Stream Transform user handler function

The handler function is the responsible for handling all the record transformation. It is the responsibility of the developer to write it. It can clone or modify the input records, skip them and emit multiple records.

Synchronous versus asynchronous execution

The mode is defined by the return value or the signature of transformation function:

• synchronous mode
  The handler is run synchronously when it declares only one argument and when its return value is a not promise.
• asynchronous mode with a returned promise
  The handler is run asynchronously when it declares only one argument and when its return value is a promise.
• asynchronous mode with a callback
  The handler is run asynchronously until a callback is called when it declare two arguments, the data to transform and the callback to be called once the transformed data is ready.

Using a callback presents the advantage that more than one record may be emitted per transform callback.

Defining synchronous transformations

In the synchronous example, the transformation function is run synchronously because it only declares one argument, the data to be transformed. It is expected to return the transformed data or to throw an error.

import { transform } from 'stream-transform';

transform([
  ['1','2','3','4'],
  ['a','b','c','d']
], function(data){
  data.push(data.shift());
  return data.join(',')+'\n';
})
  .pipe(process.stdout);

// Output:
// 2,3,4,1
// b,c,d,a

Defining asynchronous transformations with a promise

In the promise example, the transformation function is run asynchronously because it only declares one argument and because the return value is a promise with the transformed record.

import { transform } from "stream-transform";

transform(
  [
    ["a", "b", "c", "d"],
    ["1", "2", "3", "4"],
  ],
  function (data) {
    return new Promise((resolve) => {
      setImmediate(function () {
        data.push(data.shift());
        resolve(data.join(",") + "\n");
      });
    });
  },
  {
    parallel: 20,
  }
).pipe(process.stdout);

// Output:
// b,c,d,a
// 2,3,4,1

Defining asynchronous transformations

In the callback example, the transformation callback declares two arguments, the data to transform and the callback to call once the data is ready. The transformation callback is executed concurrently with a maximum of 20 parallel executions.

import { transform } from 'stream-transform';

transform([
  ['1','2','3','4'],
  ['a','b','c','d']
], function(data, callback){
  setImmediate(function(){
    data.push(data.shift());
    callback(null, data.join(',')+'\n');
  });
}, {
  parallel: 20
})
  .pipe(process.stdout);

// Output:
// 2,3,4,1
// b,c,d,a

Altering or cloning the provided data

The data received inside the transformation function is the original data and is not modified nor cloned. If you wish to alter the original data, it is your responsibility to send a new data in your transformation function instead of the original modified data.

Skipping records

Skipping records is easily achieved by returning null in synchronous mode or passing null to the callback handler in asynchronous mode.

Creating multiple records

Generating multiple records is only supported in asynchronous mode when calling a callback by providing n-arguments after the error argument instead of simply one.