Node.js Streams with TypeScript — SitePoint

Node.js is famend for its capacity to deal with I/O operations effectively, and on the coronary heart of this functionality lies the idea of streams. Streams permit you to course of information piece by piece, slightly than loading the whole lot into reminiscence directly—good for dealing with massive recordsdata, community requests, or real-time information. If you pair streams with TypeScript’s sturdy typing, you get a robust combo: efficiency meets security.

On this information, we’ll dive deep into Node.js streams, discover their sorts, and stroll by means of sensible examples utilizing TypeScript. Whether or not you’re a Node.js beginner or a TypeScript fanatic trying to degree up, this put up has you coated.

Why Streams Matter?

Image this: you’re tasked with processing a 50GB log file. Loading it totally into reminiscence would exhaust your server’s assets, resulting in crashes or sluggish efficiency. Streams resolve this by letting you deal with information because it flows, like sipping from a straw as a substitute of chugging a gallon jug.

This effectivity is why streams are a cornerstone of Node.js, powering the whole lot from file operations to HTTP servers. TypeScript enhances this by including sort definitions, catching errors at compile time, and bettering code readability. Let’s dive into the basics and see how this synergy works in apply.

The 4 Sorts of Streams

Node.js provides 4 principal stream sorts, every with a selected function:

1. Readable Streams: Knowledge sources you possibly can learn from (e.g., recordsdata, HTTP responses).
2. Writable Streams: Locations you possibly can write to (e.g., recordsdata, HTTP requests).
3. Duplex Streams: Each readable and writable (e.g., TCP sockets).
4. Rework Streams: A particular duplex stream that modifies information because it passes by means of (e.g., compression).

TypeScript enhances this by permitting us to outline interfaces for the information flowing by means of them. Let’s break them down with examples.

Setting Up Your TypeScript Setting

Earlier than we dive into code, guarantee you will have Node.js and TypeScript put in.

Create a brand new challenge:

mkdir node-streams-typescript
cd node-streams-typescript
npm init -y
npm set up typescript @sorts/node --save-dev
npx tsc --init

Replace your tsconfig.json to incorporate:

{
  "compilerOptions": {
    "goal": "ES2020",
    "module": "commonjs",
    "strict": true,
    "outDir": "./dist"
  },
  "embrace": ["src/**/*"]
}

Create a src folder and let’s begin coding!

Instance 1: Studying a File with a Readable Stream

Let’s learn a textual content file chunk by chunk. First, create a file named information.txt within the root listing of your challenge with some pattern textual content (e.g., “Whats up, streams!”).

Now, in src/readStream.ts:

import { createReadStream } from 'fs';
import { Readable } from 'stream';

const readStream: Readable = createReadStream('information.txt', { encoding: 'utf8' });

readStream
  .on('information', (chunk: string) => {
    console.log('Chunk obtained:', chunk);
  })
  .on('finish', () => {
    console.log('Completed studying the file.');
  })
  .on('error', (err: Error) => {
    console.error('Error:', err.message);
  });

Run it with:

npx tsc && node dist/readStream.js

Right here, TypeScript ensures the chunk adheres to our Chunk interface, and the error occasion handler expects an Error sort. This stream reads information.txt in chunks (default 64KB for recordsdata) and logs them.

Instance 2: Writing Knowledge with a Writable Stream

Now, let’s write information to a brand new file. In src/writeStream.ts:

import { createWriteStream } from 'fs';
import { Writable } from 'stream';

const writeStream: Writable = createWriteStream('output.txt', { encoding: 'utf8' });

const information: string[] = ['Line 1n', 'Line 2n', 'Line 3n'];

information.forEach((line: string) => {
  writeStream.write(line);
});

writeStream.finish(() => {
  console.log('Completed writing to output.txt');
});

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

Compile and run:

npx tsc && node dist/writeStream.js

This creates output.txt with three traces. TypeScript ensures the road is a string and gives autocompletion for stream strategies.

Instance 3: Piping with a Rework Stream

Piping is the place streams shine, connecting a readable stream to a writable stream. Let’s add a twist with a Rework stream to uppercase our textual content.

In src/transformStream.ts:

import { createReadStream, createWriteStream } from 'fs';
import { Rework, TransformCallback } from 'stream';


class UppercaseTransform extends Rework {
  _transform(chunk: Buffer, encoding: string, callback: TransformCallback): void {
    const upperChunk = chunk.toString().toUpperCase();
    this.push(upperChunk);
    callback();
  }
}

const readStream = createReadStream('information.txt', { encoding: 'utf8' });
const writeStream = createWriteStream('output_upper.txt');
const transformStream = new UppercaseTransform();

readStream
  .pipe(transformStream)
  .pipe(writeStream)
  .on('end', () => {
    console.log('Rework full! Test output_upper.txt');
  })
  .on('error', (err: Error) => {
    console.error('Error:', err.message);
  });

Run it:

npx tsc && node dist/transformStream.js

This reads information.txt, transforms the textual content to uppercase, and writes it to output_upper.txt.

TypeScript’s TransformCallback sort ensures our _transform technique is accurately applied.

Instance 4: Compressing Recordsdata with a Duplex Stream

Let’s deal with a extra superior state of affairs: compressing a file utilizing the zlib module, which gives a duplex stream. It comes with the ‘@sorts/node’ bundle, which we put in earlier. 

In src/compressStream.ts:

import { createReadStream, createWriteStream } from 'fs';
import { createGzip } from 'zlib';
import { pipeline } from 'stream';

const supply = createReadStream('information.txt');
const vacation spot = createWriteStream('information.txt.gz');
const gzip = createGzip();

pipeline(supply, gzip, vacation spot, (err: Error | null) => {
  if (err) {
    console.error('Compression failed:', err.message);
    return;
  }
  console.log('File compressed efficiently! Test information.txt.gz');
});

Run it:

npx tsc && node dist/compressStream.js

Right here, the pipeline ensures correct error dealing with and cleanup. The gzip stream compresses information.txt into information.txt.gz. TypeScript’s sort inference retains our code clear and secure.

Instance 5: Streaming HTTP Responses

Streams shine in community operations. Let’s simulate streaming information from an HTTP server utilizing axios. Set up it:

npm set up axios @sorts/axios

In src/httpStream.ts:

import axios from 'axios';
import { createWriteStream } from 'fs';
import { Writable } from 'stream';

async operate streamHttpResponse(url: string, outputFile: string): Promise<void> {
  const response = await axios({
    technique: 'get',
    url,
    responseType: 'stream',
  });

  const writeStream: Writable = createWriteStream(outputFile);
  response.information.pipe(writeStream);

  return new Promise((resolve, reject) => {
    writeStream.on('end', () => {
      console.log(`Downloaded to ${outputFile}`);
      resolve();
    });
    writeStream.on('error', (err: Error) => {
      console.error('Obtain failed:', err.message);
      reject(err);
    });
  });
}

streamHttpResponse('https://instance.com', 'instance.html').catch(console.error);

Run it:

npx tsc && node dist/httpStream.js

This streams an HTTP response (e.g., an online web page) to instance.html. TypeScript ensures the url and outputFile parameters are strings, and the Promise typing provides readability.

​​We are able to additionally use Node.js’s built-in Fetch API (obtainable since Node v18) or libraries like node-fetch, which additionally help streaming responses, though the stream sorts could differ (Net Streams vs. Node.js Streams).

Instance:

const response = await fetch('https://instance.com');
const writeStream = createWriteStream(outputFile);
response.physique.pipe(writeStream);

Instance 6: Actual-Time Knowledge Processing with a Customized Readable Stream

Let’s create a customized, readable stream to simulate real-time information, resembling sensor readings. In src/customReadable.ts:

import { Readable } from 'stream';

class SensorStream extends Readable {
  non-public depend: quantity = 0;
  non-public max: quantity = 10;

  constructor(choices?: any) {
    tremendous(choices);
  }

  _read(): void {
    if (this.depend < this.max) {
      const information = `Sensor studying ${this.depend}: ${Math.random() * 100}n`;
      this.push(information);
      this.depend++;
    } else {
      this.push(null); 
    }
  }
}

const sensor = new SensorStream({ encoding: 'utf8' });

sensor
  .on('information', (chunk: string) => {
    console.log('Obtained:', chunk.trim());
  })
  .on('finish', () => {
    console.log('Sensor stream full.');
  })
  .on('error', (err: Error) => {
    console.error('Error:', err.message);
  });

Run it:

npx tsc && node dist/customReadable.js

This generates 10 random “sensor readings” and streams them. TypeScript’s class typing ensures our implementation aligns with the Readable interface.

Instance 7: Chaining A number of Rework Streams

Let’s chain transforms to course of textual content in phases: uppercase it, then prepend a timestamp. In src/chainTransform.ts:

import { createReadStream, createWriteStream } from 'fs';
import { Rework, TransformCallback } from 'stream';

class UppercaseTransform extends Rework {
  _transform(chunk: Buffer, encoding: string, callback: TransformCallback): void {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
}

class TimestampTransform extends Rework {
  _transform(chunk: Buffer, encoding: string, callback: TransformCallback): void {
    const timestamp = new Date().toISOString();
    this.push(`[${timestamp}] ${chunk.toString()}`);
    callback();
  }
}

const readStream = createReadStream('information.txt', { encoding: 'utf8' });
const writeStream = createWriteStream('output_chain.txt');
const higher = new UppercaseTransform();
const timestamp = new TimestampTransform();

readStream
  .pipe(higher)
  .pipe(timestamp)
  .pipe(writeStream)
  .on('end', () => {
    console.log('Chained remodel full! Test output_chain.txt');
  })
  .on('error', (err: Error) => {
    console.error('Error:', err.message);
  });

Run it:

npx tsc && node dist/chainTransform.js

This reads information.txt, uppercases the information, provides a timestamp, and writes the end result to output_chain.txt. Chaining transforms showcases streams’ modularity.

Finest Practices for Streams in TypeScript

1. Sort Your Knowledge: Outline interfaces for chunks to catch sort errors early.
2. Deal with Errors: At all times connect error occasion listeners to keep away from unhandled exceptions.
3. Use Pipes Properly: Piping reduces handbook occasion dealing with and improves readability.
4. Backpressure: For big information, monitor writeStream.writableHighWaterMark to keep away from overwhelming the vacation spot.

Actual-World Use Case: Streaming API Responses

Think about you’re constructing an API that streams a big dataset. Utilizing specific and streams:

import specific from 'specific';
import { Readable } from 'stream';

const app = specific();

app.get('/stream-data', (req, res) => {
  const information = ['Item 1n', 'Item 2n', 'Item 3n'];
  const stream = Readable.from(information);

  res.setHeader('Content material-Sort', 'textual content/plain');
  stream.pipe(res);
});

app.hear(3000, () => {
  console.log('Server operating on port 3000');
});

Set up dependencies (npm set up specific @sorts/specific), then run it. Go to http://localhost:3000/stream-data to see the information stream in your browser!

Superior Suggestions: Dealing with Backpressure

When a writable stream can’t sustain with a readable stream, backpressure happens. Node.js handles this robotically with pipes, however you possibly can monitor it manually:

const writeStream = createWriteStream('large_output.txt');

if (!writeStream.write('information')) {
  console.log('Backpressure detected! Pausing...');
  writeStream.as soon as('drain', () => {
    console.log('Resuming...');
  });
}

This ensures your app stays responsive underneath heavy hundreds.

Precautions for utilizing Backpressure: When writing massive quantities of knowledge, the readable stream could produce information quicker than the writable stream can eat it. Whereas pipe and pipeline deal with this robotically, if writing manually, verify if write() returns false and await the ‘drain’ occasion earlier than writing extra.

Moreover, async iterators (for await…of) are trendy alternate options for consuming readable streams, which may typically simplify the code in comparison with utilizing .on(‘information’) and .on(‘finish’).

Instance:

async operate processStream(readable: Readable) {
  for await (const chunk of readable) {
    console.log('Chunk:', chunk);
  }
  console.log('Completed studying.');
}

Further factors:

Guarantee Useful resource Cleanup: That is particularly essential in customized stream implementations or when utilizing stream.pipeline. Explicitly name stream.destroy() in error eventualities or when the stream is now not wanted to launch underlying assets and stop leaks. stream.pipeline handles this robotically for piped streams.

Use Readable.from() for Comfort: When you could create a stream from an current iterable (resembling an array) or an async iterable, Readable.from() is commonly the best and most trendy strategy, requiring much less boilerplate code than making a customized Readable class.

Conclusion

Streams are a game-changer in Node.js, and TypeScript enhances them additional by introducing sort security and readability. From studying recordsdata to reworking information in real-time, mastering streams opens up a world of environment friendly I/O prospects. The examples right here—studying, writing, altering, compressing, and streaming over HTTP—scratch the floor of what’s potential.

Experiment with your individual pipelines: strive streaming logs, processing CSV recordsdata, or constructing a dwell chat system. The extra you discover, the extra you’ll admire the flexibility of streams.