How to fix Error: Duplex stream ended unexpectedly (both readable and writable failed) in Node.js

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

const transformStream = new Transform({
  transform(chunk, encoding, callback) {
    try {
      // Transform logic here
      const result = processData(chunk);
      callback(null, result);
    } catch (err) {
      callback(err);
    }
  }
});

// Critical: listen for errors
transformStream.on('error', (err) => {
  console.error('Transform stream error:', err);
  // Handle error gracefully
});

transformStream.on('end', () => {
  console.log('Stream ended successfully');
});

transformStream.on('close', () => {
  console.log('Stream closed');
});

const { pipeline } = require('stream/promises');
const fs = require('fs');
const zlib = require('zlib');

// Bad: No error handling
// readStream.pipe(gzip).pipe(writeStream);

// Good: Automatic error handling
async function processFiles() {
  try {
    await pipeline(
      fs.createReadStream('input.txt'),
      zlib.createGzip(),
      fs.createWriteStream('output.txt.gz')
    );
    console.log('Pipeline completed successfully');
  } catch (err) {
    console.error('Pipeline failed:', err);
    // All streams are properly cleaned up
  }
}

processFiles();

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

const duplexStream = new Duplex({
  write(chunk, encoding, callback) {
    // Process write
    this.processWrite(chunk);
    callback();
  },
  read(size) {
    // Provide data for reading
    this.push(this.getData());
  }
});

// Handle backpressure when writing
function writeData(stream, data) {
  const canContinue = stream.write(data);

  if (!canContinue) {
    console.log('Backpressure detected, waiting for drain...');
    stream.once('drain', () => {
      console.log('Stream drained, can continue writing');
    });
  }
}

// Check if stream is writable before writing
if (duplexStream.writable) {
  writeData(duplexStream, 'some data');
}

function safeWrite(stream, data) {
  // Check if stream is still writable
  if (!stream.writable || stream.destroyed) {
    console.error('Stream is not writable');
    return false;
  }

  // Check if stream has ended
  if (stream.writableEnded) {
    console.error('Stream has already ended');
    return false;
  }

  return stream.write(data);
}

function safeRead(stream) {
  // Check if stream is still readable
  if (!stream.readable || stream.destroyed) {
    console.error('Stream is not readable');
    return null;
  }

  // Check if stream has ended
  if (stream.readableEnded) {
    console.error('Stream has already ended');
    return null;
  }

  return stream.read();
}

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

class CustomDuplex extends Duplex {
  constructor(options) {
    super(options);
    this.resources = [];
  }

  _construct(callback) {
    // Initialize resources
    this.setupResources();
    callback();
  }

  _destroy(error, callback) {
    // Clean up resources
    console.log('Cleaning up duplex stream resources');
    this.resources.forEach(resource => resource.close());
    this.resources = [];

    // Call callback with error if present
    callback(error);
  }

  write(chunk, encoding, callback) {
    // Check state before writing
    if (this.destroyed) {
      const err = new Error('Cannot write to destroyed stream');
      if (callback) callback(err);
      return false;
    }
    return super.write(chunk, encoding, callback);
  }
}

// Usage with proper error handling
const stream = new CustomDuplex();

stream.on('error', (err) => {
  console.error('Stream error:', err);
  stream.destroy(err); // Ensure cleanup happens
});

// Graceful shutdown
process.on('SIGTERM', () => {
  stream.destroy(); // Clean shutdown
});