How to fix Error: Transform stream callback not called (transform not pushed data) in Node.js

const { Transform } = require('stream');

const transformStream = new Transform({
  transform(chunk, encoding, callback) {
    try {
      const result = processData(chunk);
      // Option 1: Pass data directly to callback
      callback(null, result);
    } catch (err) {
      // Always pass errors to callback
      callback(err);
    }
  }
});

// Usage
sourceStream.pipe(transformStream).pipe(destinationStream);

const { Transform } = require('stream');

const transformStream = new Transform({
  async transform(chunk, encoding, callback) {
    try {
      // Async operation
      const result = await fetchAndProcess(chunk);
      callback(null, result);
    } catch (err) {
      // CRITICAL: Pass error to callback
      callback(err);
    }
  }
});

// Example with Promise
const transformStream2 = new Transform({
  transform(chunk, encoding, callback) {
    asyncOperation(chunk)
      .then(result => {
        callback(null, result); // Call on success
      })
      .catch(err => {
        callback(err); // Call with error
      });
  }
});

const { Transform } = require('stream');

// Pattern 1: Pass data to callback (recommended)
const transform1 = new Transform({
  transform(chunk, encoding, callback) {
    const result = processData(chunk);
    callback(null, result); // Data goes directly to callback
  }
});

// Pattern 2: Use this.push() then call callback
const transform2 = new Transform({
  transform(chunk, encoding, callback) {
    const result = processData(chunk);
    this.push(result);
    callback(); // Signal chunk processing is complete
  }
});

// Pattern 3: Multiple pushes with one callback
const transform3 = new Transform({
  transform(chunk, encoding, callback) {
    // Can push multiple times
    this.push('prefix_' + chunk);
    this.push('_suffix');
    // But callback is called once
    callback();
  }
});

// Pattern 4: Filter out data (no push, but still callback)
const filterTransform = new Transform({
  transform(chunk, encoding, callback) {
    if (shouldKeep(chunk)) {
      callback(null, chunk);
    } else {
      callback(); // Don't output this chunk
    }
  }
});

const { Transform } = require('stream');

// WRONG: Throwing an error breaks the stream
const badTransform = new Transform({
  transform(chunk, encoding, callback) {
    if (chunk.length === 0) {
      throw new Error('Empty chunk'); // Bad: stream breaks
    }
    callback(null, chunk.toUpperCase());
  }
});

// CORRECT: Pass errors to callback
const goodTransform = new Transform({
  transform(chunk, encoding, callback) {
    if (chunk.length === 0) {
      // Good: error is handled by callback
      return callback(new Error('Empty chunk'));
    }
    callback(null, chunk.toUpperCase());
  }
});

// CORRECT: Use try-catch to capture and pass errors
const safeTransform = new Transform({
  transform(chunk, encoding, callback) {
    try {
      const result = riskyOperation(chunk);
      callback(null, result);
    } catch (err) {
      callback(err); // Pass the error
    }
  }
});

// Listen for error events
goodTransform.on('error', (err) => {
  console.error('Transform error:', err.message);
});

const { Transform } = require('stream');

// WRONG: 'this' binding is lost in callback
const badTransform = new Transform({
  transform(chunk, encoding, callback) {
    setTimeout(() => {
      // 'this' is undefined here
      this.push(processData(chunk)); // Error: this is undefined
      callback();
    }, 100);
  }
});

// CORRECT: Preserve 'this' with arrow functions or bind()
const goodTransform = new Transform({
  transform(chunk, encoding, callback) {
    // Arrow function preserves 'this'
    setTimeout(() => {
      this.push(processData(chunk)); // 'this' is the Transform
      callback();
    }, 100);
  }
});

// CORRECT: Use bind() with regular functions
const altTransform = new Transform({
  transform: function(chunk, encoding, callback) {
    setTimeout(function() {
      this.push(processData(chunk));
      callback();
    }.bind(this), 100);
  }
});

// CORRECT: Store reference outside callback
const storeRefTransform = new Transform({
  transform(chunk, encoding, callback) {
    const self = this;
    setTimeout(() => {
      self.push(processData(chunk));
      callback();
    }, 100);
  }
});

const { Transform, pipeline } = require('stream');

const myTransform = new Transform({
  timeout: 5000, // Optional: add timeout
  transform(chunk, encoding, callback) {
    try {
      const result = processData(chunk);
      callback(null, result);
    } catch (err) {
      callback(err);
    }
  }
});

// Add error listeners
myTransform.on('error', (err) => {
  console.error('Transform error:', err);
});

// Use pipeline for proper error handling
const { pipeline } = require('stream/promises');

async function processStream() {
  try {
    await pipeline(
      sourceStream,
      myTransform,
      destinationStream
    );
    console.log('Transform completed successfully');
  } catch (err) {
    console.error('Pipeline failed:', err);
    // All streams are cleaned up automatically
  }
}

// With timeout detection
const transformWithTimeout = new Transform({
  transform(chunk, encoding, callback) {
    const timeout = setTimeout(() => {
      callback(new Error('Transform operation timed out'));
    }, 5000);

    try {
      const result = processData(chunk);
      clearTimeout(timeout);
      callback(null, result);
    } catch (err) {
      clearTimeout(timeout);
      callback(err);
    }
  }
});

processStream();