Node.js ReadStream Reference

ReadStream Object

(when reading) - For reading from network connections http.IncomingMessage

ReadStream Properties

Property

ReadStream Methods

Method

Wraps an old - style readable stream as a new - style readable stream.

ReadStream Events

Event

Description

Reading from a File

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.)

});
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}`);
});

Controlling Stream Flow

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}`);
});

Using the read() Method

This example demonstrates using the read()

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, {

});
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}`);
});

Piping between Streams

This example demonstrates how to use pipe()

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)}%`);

[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}`));

Reading from Standard Input

ReadStream:

// 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();
});

HTTP ReadStream Example

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 {

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}`);
});

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

Error Handling with ReadStreams

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');
});

Best Practices for ReadStreams

Always handle errors

Clean up resources

Use pipe() for most cases

Set appropriate buffer sizes

Choose the right mode