Node.js ReadStream Reference

const fs = require('fs');
const path = require('path');

// Create a sample file for the example
const sampleFile = path.join(__dirname, 'readstream-example.txt');
fs.writeFileSync(sampleFile, 'This is a test file.\nIt has multiple lines.\nEach line has its own content.\nStreaming makes file reading efficient.');

// Create a ReadStream to read from the file
const readStream = fs.createReadStream(sampleFile, {
  // Options
  encoding: 'utf8',  // Set the encoding (utf8, ascii, binary, etc.)
  highWaterMark: 64, // Buffer size in bytes
  autoClose: true    // Automatically close the file descriptor when the stream ends
});

console.log('File ReadStream properties:');
console.log(`- Path: ${readStream.path}`);
console.log(`- Pending: ${readStream.pending}`);
console.log(`- High Water Mark: ${readStream.readableHighWaterMark} bytes`);

// Handle stream events
readStream.on('open', (fd) => {
  console.log(`File opened with descriptor: ${fd}`);
});

readStream.on('ready', () => {
  console.log('ReadStream is ready');
});

// Handle data events
readStream.on('data', (chunk) => {
  console.log('\nReceived chunk:');
  console.log('-'.repeat(20));
  console.log(chunk);
  console.log('-'.repeat(20));
  console.log(`Bytes read so far: ${readStream.bytesRead}`);
});

readStream.on('end', () => {
  console.log('\nReached end of file');
  console.log(`Total bytes read: ${readStream.bytesRead}`);
  
  // Clean up the sample file
  fs.unlinkSync(sampleFile);
  console.log('Sample file removed');
});

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

readStream.on('error', (err) => {
  console.error(`Error: ${err.message}`);
});

const fs = require('fs');
const path = require('path');

// Create a large sample file for the flow control example
const flowFile = path.join(__dirname, 'flow-control-example.txt');
let sampleContent = '';
for (let i = 1; i <= 1000; i++) {
  sampleContent += `This is line ${i} of the test file.\n`;
}
fs.writeFileSync(flowFile, sampleContent);

// Create a ReadStream
const readStream = fs.createReadStream(flowFile, {
  encoding: 'utf8',
  highWaterMark: 1024 // 1KB buffer
});

let chunkCount = 0;
let totalBytes = 0;
let isPaused = false;

// Handle data chunks with flow control
readStream.on('data', (chunk) => {
  chunkCount++;
  totalBytes += chunk.length;
  
  console.log(`Chunk #${chunkCount} received, size: ${chunk.length} bytes`);
  
  // Pause the stream every 5 chunks to demonstrate flow control
  if (chunkCount % 5 === 0 && !isPaused) {
    console.log('\nPausing the stream for 1 second...');
    isPaused = true;
    
    // Pause the stream
    readStream.pause();
    
    // Resume after 1 second
    setTimeout(() => {
      console.log('Resuming the stream...\n');
      isPaused = false;
      readStream.resume();
    }, 1000);
  }
});

readStream.on('end', () => {
  console.log(`\nFinished reading file. Received ${chunkCount} chunks, ${totalBytes} bytes total.`);
  
  // Clean up the sample file
  fs.unlinkSync(flowFile);
  console.log('Sample file removed');
});

readStream.on('error', (err) => {
  console.error(`Error: ${err.message}`);
});

const fs = require('fs');
const path = require('path');

// Create a sample file
const readableFile = path.join(__dirname, 'readable-example.txt');
fs.writeFileSync(readableFile, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'.repeat(100));

// Create a ReadStream without auto-flowing
const readStream = fs.createReadStream(readableFile, {
  highWaterMark: 32 // Small buffer to demonstrate multiple reads
});

console.log('Using the readable event and read() method:');

// Using the 'readable' event for manual reading
readStream.on('readable', () => {
  let chunk;
  // read() returns null when there is no more data to read
  while (null !== (chunk = readStream.read(16))) {
    console.log(`Read ${chunk.length} bytes: ${chunk.toString('utf8').substring(0, 10)}...`);
  }
});

readStream.on('end', () => {
  console.log('End of stream reached');
  
  // Clean up the sample file
  fs.unlinkSync(readableFile);
  console.log('Sample file removed');
});

readStream.on('error', (err) => {
  console.error(`Error: ${err.message}`);
});

const fs = require('fs');
const path = require('path');
const zlib = require('zlib');

// Source and destination file paths
const sourceFile = path.join(__dirname, 'pipe-source-example.txt');
const destFile = path.join(__dirname, 'pipe-destination.txt');
const compressedFile = path.join(__dirname, 'pipe-compressed.gz');

// Create sample content
fs.writeFileSync(sourceFile, 'This is the source content for the pipe example.\n'.repeat(100));

// Create ReadStream and various WriteStreams
const readStream = fs.createReadStream(sourceFile);
const writeStream = fs.createWriteStream(destFile);
const compressStream = zlib.createGzip(); // Compression transform stream
const compressedWriteStream = fs.createWriteStream(compressedFile);

// Pipe the ReadStream directly to the WriteStream
readStream.pipe(writeStream);

// Listen for completion events
writeStream.on('finish', () => {
  console.log(`File copied from ${sourceFile} to ${destFile}`);
  
  // Create a new ReadStream to demonstrate chained pipes
  const readStream2 = fs.createReadStream(sourceFile);
  
  // Chain multiple pipes: read -> compress -> write
  readStream2.pipe(compressStream).pipe(compressedWriteStream);
  
  compressedWriteStream.on('finish', () => {
    console.log(`File compressed from ${sourceFile} to ${compressedFile}`);
    
    // Compare file sizes
    const originalSize = fs.statSync(sourceFile).size;
    const compressedSize = fs.statSync(compressedFile).size;
    console.log(`Original size: ${originalSize} bytes`);
    console.log(`Compressed size: ${compressedSize} bytes`);
    console.log(`Compression ratio: ${(compressedSize / originalSize * 100).toFixed(2)}%`);
    
    // Clean up files after demonstration
    [sourceFile, destFile, compressedFile].forEach(file => {
      fs.unlinkSync(file);
    });
    console.log('All sample files removed');
  });
});

// Handle errors
readStream.on('error', (err) => console.error(`Read error: ${err.message}`));
writeStream.on('error', (err) => console.error(`Write error: ${err.message}`));
compressStream.on('error', (err) => console.error(`Compression error: ${err.message}`));
compressedWriteStream.on('error', (err) => console.error(`Compressed write error: ${err.message}`));

// process.stdin is a ReadStream

console.log('Enter some text (press Ctrl+D or Ctrl+C to end input):');

// Set the encoding to utf8 to get strings instead of Buffer objects
process.stdin.setEncoding('utf8');

let inputData = '';

// Handle data from stdin
process.stdin.on('data', (chunk) => {
  console.log(`Received chunk: "${chunk.trim()}"`);
  inputData += chunk;
});

// Handle the end of input
process.stdin.on('end', () => {
  console.log('\nEnd of input.');
  console.log(`Total input received: ${inputData.length} characters`);
  console.log('You entered:');
  console.log('-'.repeat(20));
  console.log(inputData);
  console.log('-'.repeat(20));
});

// Handle Ctrl+C (SIGINT)
process.on('SIGINT', () => {
  console.log('\nInput interrupted with Ctrl+C');
  process.exit();
});

// Note: This example requires user input in a terminal
// Can't be effectively demonstrated in the W3Schools TryIt editor

const http = require('http');

// Create an HTTP server
const server = http.createServer((req, res) => {
  // req is an http.IncomingMessage, which is a ReadStream
  console.log(`Received ${req.method} request to ${req.url}`);
  
  // Set response headers
  res.setHeader('Content-Type', 'text/plain');
  
  // Handle different types of requests
  if (req.method === 'GET') {
    res.end('Send a POST request with a body to see the ReadStream in action');
  }
  else if (req.method === 'POST') {
    // Set encoding for the request stream
    req.setEncoding('utf8');
    
    let body = '';
    
    // Handle data events from the request stream
    req.on('data', (chunk) => {
      console.log(`Received chunk of ${chunk.length} bytes`);
      body += chunk;
      
      // Implement a simple flood protection
      if (body.length > 1e6) {
        // If body is too large, destroy the stream
        body = '';
        res.writeHead(413, {'Content-Type': 'text/plain'});
        res.end('Request entity too large');
        req.destroy();
      }
    });
    
    // Handle the end of the request stream
    req.on('end', () => {
      console.log('End of request data');
      
      try {
        // Try to parse as JSON
        const data = JSON.parse(body);
        console.log('Parsed JSON data:', data);
        
        // Send a response
        res.writeHead(200, {'Content-Type': 'application/json'});
        res.end(JSON.stringify({
          message: 'Data received',
          size: body.length,
          data: data
        }));
      } catch (e) {
        // If not valid JSON, just echo back the raw data
        console.log('Could not parse as JSON, treating as plain text');
        
        res.writeHead(200, {'Content-Type': 'text/plain'});
        res.end(`Received ${body.length} bytes of data:\n${body}`);
      }
    });
  }
  else {
    // For other HTTP methods
    res.writeHead(405, {'Content-Type': 'text/plain'});
    res.end('Method not allowed');
  }
});

// Start the server
const PORT = 8080;
server.listen(PORT, () => {
  console.log(`HTTP ReadStream example server running at http://localhost:${PORT}`);
  console.log('To test:');
  console.log(`1. Open http://localhost:${PORT} in a browser for GET request`);
  console.log(`2. Use curl or Postman to send POST requests with a body to http://localhost:${PORT}`);
});

// Note: To test with curl:
// curl -X POST -H "Content-Type: application/json" -d '{"name":"John","age":30}' http://localhost:8080

const fs = require('fs');
const path = require('path');

// Function to create and handle a ReadStream with proper error handling
function readWithErrorHandling(filePath) {
  console.log(`Attempting to read: ${filePath}`);
  
  // Create the ReadStream
  const readStream = fs.createReadStream(filePath);
  
  // Set up promise to capture result or error
  return new Promise((resolve, reject) => {
    let data = '';
    
    // Handle data events
    readStream.on('data', (chunk) => {
      data += chunk;
    });
    
    // Handle successful completion
    readStream.on('end', () => {
      console.log(`Successfully read ${readStream.bytesRead} bytes from ${filePath}`);
      resolve(data);
    });
    
    // Handle errors
    readStream.on('error', (err) => {
      console.error(`Error reading ${filePath}: ${err.message}`);
      reject(err);
    });
    
    // Handle stream closure (always happens, even if there's an error)
    readStream.on('close', () => {
      console.log(`Stream for ${filePath} closed`);
    });
  });
}

// Test with both existing and non-existing files
const existingFile = path.join(__dirname, 'test-existing.txt');
const nonExistingFile = path.join(__dirname, 'non-existing-file.txt');

// Create the test file
fs.writeFileSync(existingFile, 'This is test content for error handling example');

// Example 1: Reading an existing file
console.log('Example 1: Reading an existing file');
readWithErrorHandling(existingFile)
  .then(data => {
    console.log('File content:', data);
    
    // Example 2: Reading a non-existing file
    console.log('\nExample 2: Reading a non-existing file');
    return readWithErrorHandling(nonExistingFile);
  })
  .catch(err => {
    console.log('Error caught in Promise catch:', err.message);
  })
  .finally(() => {
    // Clean up the test file
    if (fs.existsSync(existingFile)) {
      fs.unlinkSync(existingFile);
      console.log('Test file removed');
    }
  });

// Example 3: Demonstrating destroyed streams
console.log('\nExample 3: Demonstrating destroyed streams');
const destroyTestFile = path.join(__dirname, 'destroy-test.txt');
fs.writeFileSync(destroyTestFile, 'A'.repeat(10000));

const destroyStream = fs.createReadStream(destroyTestFile);

destroyStream.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes before destroying the stream`);
  
  // Destroy the stream after receiving the first chunk
  console.log('Deliberately destroying the stream');
  destroyStream.destroy(new Error('Stream manually destroyed'));
});

destroyStream.on('error', (err) => {
  console.error(`Destruction error: ${err.message}`);
});

destroyStream.on('close', () => {
  console.log('Destroyed stream closed');
  
  // Clean up
  fs.unlinkSync(destroyTestFile);
  console.log('Destroy test file removed');
});