Stream Transformation

const es = require('event-stream');

/**
 * Create a transform stream from an async function
 * @param {Function} asyncFunction - Function called with (data, callback)
 * @returns {TransformStream} Stream that transforms data asynchronously
 */
function map(asyncFunction);

const es = require('event-stream');

// Async transformation
const asyncTransformer = es.map(function(data, callback) {
  // Simulate async operation
  setTimeout(() => {
    const result = data.toString().toUpperCase();
    callback(null, result);
  }, 10);
});

es.readArray(['hello', 'world'])
  .pipe(asyncTransformer)
  .pipe(es.writeArray(function(err, result) {
    console.log(result); // ['HELLO', 'WORLD']
  }));

// With error handling
const errorHandler = es.map(function(data, callback) {
  if (typeof data !== 'string') {
    return callback(new Error('Expected string'));
  }
  callback(null, data.length);
});

// Filter by not calling callback
const evenFilter = es.map(function(number, callback) {
  if (number % 2 === 0) {
    callback(null, number); // Keep even numbers
  } else {
    callback(); // Drop odd numbers
  }
});

/**
 * Create a transform stream with synchronous transformation
 * @param {Function} syncFunction - Function that transforms data synchronously
 * @returns {TransformStream} Stream that transforms data synchronously
 */
function mapSync(syncFunction);

const es = require('event-stream');

// Synchronous transformation
const doubler = es.mapSync(function(data) {
  return data * 2;
});

es.readArray([1, 2, 3, 4])
  .pipe(doubler)
  .pipe(es.writeArray(function(err, result) {
    console.log(result); // [2, 4, 6, 8]
  }));

// Object transformation
const userFormatter = es.mapSync(function(user) {
  return {
    id: user.id,
    name: user.name.toUpperCase(),
    active: true
  };
});

// Return undefined to filter out data
const positiveOnly = es.mapSync(function(number) {
  return number > 0 ? number : undefined;
});

/**
 * Filter stream data based on a test function
 * @param {Function} testFunction - Function that returns boolean for each item
 * @returns {TransformStream} Stream that only emits items passing the test
 */
function filterSync(testFunction);

const es = require('event-stream');

// Filter numbers
const evenNumbers = es.filterSync(function(number) {
  return number % 2 === 0;
});

es.readArray([1, 2, 3, 4, 5, 6])
  .pipe(evenNumbers)
  .pipe(es.writeArray(function(err, result) {
    console.log(result); // [2, 4, 6]
  }));

// Filter objects
const activeUsers = es.filterSync(function(user) {
  return user.active === true;
});

const users = [
  { name: 'Alice', active: true },
  { name: 'Bob', active: false },
  { name: 'Charlie', active: true }
];

es.readArray(users)
  .pipe(activeUsers)
  .pipe(es.writeArray(function(err, result) {
    console.log(result); // [{ name: 'Alice', active: true }, { name: 'Charlie', active: true }]
  }));

/**
 * Map and flatten nested data structures
 * @param {Function} mapperFunction - Function that maps each array element
 * @returns {TransformStream} Stream that flattens mapped results
 */
function flatmapSync(mapperFunction);

const es = require('event-stream');

// Flatten arrays of numbers
const multiplier = es.flatmapSync(function(number) {
  return number * 2;
});

// Input: array of arrays, each element gets mapped
es.readArray([[1, 2], [3, 4]])
  .pipe(multiplier)
  .pipe(es.writeArray(function(err, result) {
    console.log(result); // [2, 4, 6, 8]
  }));

// Transform object properties
const propertyExtractor = es.flatmapSync(function(obj) {
  return obj.name + '_processed';
});

es.readArray([
  [{ name: 'item1' }, { name: 'item2' }],
  [{ name: 'item3' }]
])
  .pipe(propertyExtractor)
  .pipe(es.writeArray(function(err, result) {
    console.log(result); // ['item1_processed', 'item2_processed', 'item3_processed']
  }));

/**
 * Split stream data by delimiter (string or regex)
 * @param {string|RegExp} matcher - Delimiter to split on (defaults to '\n')
 * @returns {TransformStream} Stream that emits split chunks
 */
function split(matcher);

const es = require('event-stream');
const fs = require('fs');

// Split on newlines (default)
fs.createReadStream('file.txt')
  .pipe(es.split())
  .pipe(es.map(function(line, cb) {
    console.log('Line:', line);
    cb(null, line);
  }));

// Split on custom delimiter
const csvSplitter = es.split(',');
es.readArray(['a,b,c', 'd,e,f'])
  .pipe(csvSplitter)
  .pipe(es.writeArray(function(err, result) {
    console.log(result); // ['a', 'b', 'c', 'd', 'e', 'f']
  }));

// Keep line breaks using regex
fs.createReadStream('file.txt')
  .pipe(es.split(/(\r?\n)/))
  .pipe(es.map(function(chunk, cb) {
    // Chunks will include line breaks
    cb(null, chunk);
  }));

// Split on regex pattern
const wordSplitter = es.split(/\s+/);