DEV Community: Manoj Kumar Patra

Creational Design Patterns

Manoj Kumar Patra — Wed, 02 Oct 2024 08:59:34 +0000

# Factory

Ability to decouple the creation of an object from one particular implementation.

Inside the factory, we can choose to create a new instance of a class using the new operator, or leverage closures to dynamically build a stateful object literal, or even return a different object type based on a particular condition.

A factory can also be used as an encapsulation mechanism, thanks to closures.

⭐ Encapsulation refers to controlling the access to some internal details of a component by preventing external code from manipulating them directly.

EXAMPLE 1


function createPerson (name) {
  const privateProperties = {};
  const person = {
    setName (name) {
      if (!name) {
        throw new Error('A person must have a name');
      }
      privateProperties.name = name;
    }
    getName () {
      return privateProperties.name;
    }
  };
  person.setName(name);
  return person;
}

Different ways to enforce encapsulation:

Using closures
🔗 Using private class fields (the hashbang # prefix syntax)
🔗 Using WeakMaps
🔗 Using symbols
🔗 Defining private variables in a constructor

EXAMPLE 2


class Profiler {
  constructor(label) {
    this.label = label;
    this.lastTime = null;
  }

  start() {
    /**
     * process.hrtime() is used to measure performance 
     * in a more precise manner than Date.now(), 
     * as it provides timestamps in nanoseconds.
     * 
     * The returned value is an array with two elements:
     * The first element is the number of seconds.
     * The second element is the number of nanoseconds.
     */
    this.lastTime = process.hrtime();
  }

  end() {
    const diff = process.hrtime(this.lastTime);
    console.log(`Timer "${this.label}" took ${diff[0]} seconds ` +
      `and ${diff[1]} nanoseconds`);
  }
}

const noopProfiler = {
  start() {},
  end() {},
};

export function createProfiler(label) {
  if (process.env.NODE_ENV === 'production') {
    return noopProfiler;
  }

  return new Profiler(label);
}

Examples of npm libraries using factory pattern - knex

# Builder

Builder is a creational design pattern that simplifies the creation of complex objects by providing a fluent interface, which allows us to build the object step by step.

EXAMPLE 1


class Boat {
  constructor(hasMotor, motorCount, motorBrand,
              motorModel, hasSails, sailsCount,
              sailsMaterial, sailsColor, hullColor,
              hasCabin) {}

  // ...
}

class BoatBuilder {
  withMotors(count, brand, model) {
    this.hasMotor = true;
    this.motorCount = count;
    this.motorBrand = brand;
    this.motorModel = model;
    return this;
  }

  withSails(count, material, color) {
    this.hasSails = true;
    this.sailsCount = count;
    this.sailsMaterial = material;
    this.sailsColor = color;
    return this;
  }

  hullColor(color) {
    this.hullColor = color;
    return this;
  }

  withCabin() {
    this.hasCabin = true;
    return this;
  }

  build() {
    return new Boat({
      hasMotor: true,
      motorCount: 2,
      motorBrand: 'Best Motor Co.',
      motorModel: 'OM123',
      hasSails: true,
      sailsCount: 1,
      sailsMaterial: 'fabric',
      sailsColor: 'white',
      hullColor: 'blue',
      hasCabin: false
    });
  }
}

const myBoat = new BoatBuilder()
  .withMotors(2, 'Best Motor Co.', 'OM123')
  .withSails(1, 'fabric', 'white')
  .withCabin()
  .hullColor('blue')
  .build();

⭐ Using a builder that is separate from the target class has the advantage of always producing instances that are guaranteed to be in a consistent state.

EXAMPLE 2


export class Url {
  constructor (protocol, username, password, hostname,
    port, pathname, search, hash) {
    this.protocol = protocol;
    this.username = username;
    this.password = password;
    this.hostname = hostname;
    this.port = port;
    this.pathname = pathname;
    this.search = search;
    this.hash = hash;

    this.validate();
  }

  validate() {
    if (!this.protocol || !this.hostname) {
      throw new Error(`Must specify at least a protocol and a hostname`);
    }
  }

  toString() {
    let url = '';
    url += `${this.protocol}://`;
    if (this.username && this.password) {
      url += `${this.username}:${this.password}`;
    }
    url += this.hostname;
    if (this.port) {
      url += this.port;
    }
    if (this.pathname) {
      url += this.pathname;
    }
    if (this.search) {
      url += `?${this.search}`;
    }
    if (this.hash) {
      url += `#${this.hash}`;
    }
    return url;
  }
}

export class UrlBuilder {
  setProtocol(protocol) {
    this.protocol = protocol;
    return this;
  }

  setAuthentication(username, password) {
    this.username = username;
    this.password = password;
    return this;
  }

  setHostname(hostname) {
    this.hostname = hostname;
    return this;
  }

  setPort(port) {
    this.port = port;
    return this;
  }

  setPathname(pathname) {
    this.pathname = pathname;
    return this;
  }

  setSearch(search) {
    this.search = search;
    return this;
  }

  setHash(hash) {
    this.hash = hash;
    return this;
  }

  build() {
    return new Url(this.protocol, this.username, this.password,
      this.hostname, this.port, this.pathname, this.search,
      this.hash);
  }
}

const url = new UrlBuilder()
  .setProtocol('https')
  .setAuthentication('user', 'pass')
  .setHostname('example.com')
  .build();

# Revealing constructor

Useful when we want to allow an object's internals to be manipulated only during its creation phase.

Use cases

Creating objects that can be modified only at creation time
Creating objects whose custom behavior can be defined only at creation time
Creating objects that can be initialized only once at creation time

Syntax


//                  (1)               (2)       (3)
const object = new SomeClass(function executor(revealedMembers) { // manipulation code ...
})

Immutable objects

Immutable refers to the property of an object by which its data or state becomes unmodifiable once it's been created.

Modifying an immutable object can only be done by creating a new copy and can make the code more maintainable and easier to reason about.
Efficient change detection with strict equality operator

Immutable version of the Node.js Buffer component


const MODIFIER_NAMES = ["swap", "write", "fill"];

export class ImmutableBuffer {
  constructor(size, executor) {
    const buffer = Buffer.alloc(size);
    const modifiers = {};
    for (const prop in buffer) {
      if (typeof buffer[prop] !== "function") {
        continue;
      }

      if (MODIFIER_NAMES.some(modifier => prop.startsWith(modifier))) {
        modifiers[prop] = buffer[prop].bind(buffer);
      } else {
        this[prop] = buffer[prop].bind(buffer);
      }
    }

    executor(modifiers);
  }
}

// USAGE

const hello = "Hello!";
const immutable = new ImmutableBuffer(hello.length, ({ write }) => {
  write(hello);
});

console.log(String.fromCharCode(immutable.readInt8(0)));

A popular application of the Revealing Constructor pattern is in the JavaScript Promise class.

# Singleton

Reasons to use a single instance across all the components of an application:

For sharing stateful information
For optimizing resource usage
To synchronize access to a resource

The module is cached using its full path as the lookup key, so it is only guaranteed to be a singleton within the current package.

❓ What happens when package-a and package-b have a dependency on the mydb package, but package-a depends on version 1.0.0 of the mydb package, while package-b depends on version 2.0.0 of the same package?


app/
`-- node_modules
    |-- package-a
    |  `-- node_modules
    |      `-- mydb
    `-- package-b
        `-- node_modules
            `-- mydb

In such a case, a typical package manager such as npm or yarn would not "hoist" the dependency to the top node_modules directory, but it will instead install a private copy of each package in an attempt to fix the version incompatibility.

With the directory structure we just saw, both package-a and package-b have a dependency on the mydb package; in turn, the app package, which is our root package, depends on both package-a and package-b.


import { getDbInstance as getDbFromA } from 'package-a';
import { getDbInstance as getDbFromB } from 'package-b';
const isSame = getDbFromA() === getDbFromB();
console.log('Is the db instance in package-a the same ' +
  `as package-b? ${isSame ? 'YES' : 'NO'}`); // NO

If instead, both package-a and package-b required two versions of the mydb package compatible with each other, for example, ^2.0.1 and ^2.0.7, then the package manager would install the mydb package into the top-level node_modules directory (a practice known as dependency hoisting), effectively sharing the same instance with package-a, package-b, and the root package.

# Wiring Modules using dependency injection

The Node.js module system and the Singleton pattern can serve as great tools for organizing and wiring together the components of an application. However, they might introduce a tighter coupling between components.

📖 Dependency Injection (DI) is a very simple pattern in which the dependencies of a component are provided as input by an external entity, often referred to as the injector.

The injector has the goal of providing an instance that fulfills the dependency for the service.

Disadvantage of DI

Dependency Injection forces us to build the dependency graph of the entire application by hand, making sure that we do it in the right order. This can become unmanageable when the number of modules to wire becomes too high.

Alternative - Inversion of Control

📖 Inversion of Control, allows us to shift the responsibility of wiring the modules of an application to a third-party entity. This entity can be a service locator (a simple component used to retrieve a dependency, for example, serviceLocator.get('db')) or a dependency injection container (a system that injects the dependencies into a component based on some metadata specified in the code itself or in a configuration file).

Coding with Streams

Manoj Kumar Patra — Wed, 02 Oct 2024 08:58:52 +0000

# Buffer vs stream

With buffer mode, for an input operation, all the data coming from a resource spread over time will be collected into a buffer until the operation is completed. It is then passed back to the caller as one single blob of data.

With streams, the data is processed as soon as it arrives from the resource.

Streams are more efficient both in terms of space (memory usage) and time (computation clock time).

Spatial efficiency

Buffers in V8 have limited size. To find out the maximum size of a buffer:


import buffer from "buffer";
console.log(buffer.constants.MAX_LENGTH);

Trying to read a file that is bigger than the maximum allowed buffer size will result in the following error:


RangeError [ERR_FS_FILE_TOO_LARGE]: File size (8130792448) is greater
than possible Buffer: 2147483647 bytes

Example:

Gzip using buffers


import { promises as fs } from 'fs';
import { gzip } from 'zlib';
import { promisify } from 'util';

const gzipPromise = promisify(gzip);

const filename = process.argv[2];

async function main () {
  const data = await fs.readFile(filename);
  const gzippedData = await gzipPromise(data);
  await fs.writeFile(`${filename}.gz`, gzippedData);
  console.log('File successfully compressed');
}

main();

Gzip using streams - This code runs smoothly against files of any size and with constant memory utilization.


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

const filename = process.argv[2];

createReadStream(filename)
  .pipe(createGzip())
  .pipe(createWriteStream(`${filename}.gz`))
  .on('finish', () => console.log('File successfully compressed'));

Time efficiency

Consider an application where the client compresses a file and uploads it to a remote HTTP server, which, in turn, decompresses it and saves it on the filesystem.

Implementing this using streams:

// gzip-receive.js

import { createServer } from 'http';
import { createWriteStream } from 'fs';
import { createGunzip } from 'zlib';
import { basename, join } from 'path';
import { createDecipheriv, randomBytes } from 'crypto';

const server = createServer((req, res) => {
  // basename is used to remove any possible path from the name of the received file
  const filename = basename(req.headers['X-Filename']);
  const iv = Buffer.from(
    req.headers['X-Initialization-Vector'], 'hex'
  );
  const destFilename = join('received_files', filename);
  console.log(`File request received: ${filename}`);

  // req is a stream object
  req.pipe(createDecipheriv('aes192', secret, iv))
    .pipe(createGunzip())
    .pipe(createWriteStream(destFilename))
    .on('finish', () => {
      res.writeHead(201, { 'Content-Type': 'text/plain' });
      res.end('OK\n');
      console.log(`File saved: ${destFilename}`);
    });
});

server.listen(3000, () => console.log('Listening on http://localhost:3000'));

// gzip-send.js

import { request } from 'http';
import { createGzip } from 'zlib';
import { createReadStream } from 'fs';
import { basename } from 'path';
import { createCipheriv, randomBytes } from  'crypto';

const filename = process.argv[2];
const serverHost = process.argv[3];
const secret = Buffer.from(process.argv[4], 'hex');
const iv = randomBytes(16);

const httpRequestOptions = {
  hostname: serverHost,
  port: 3000,
  path: '/',
  method: 'PUT',
  headers: {
    'Content-Type': 'application/octet-stream',
    'Content-Encoding': 'gzip',
    'X-Filename': basename(filename),
    'X-Initialization-Vector': iv.toString('hex')
  },
};

const req = request(httpRequestOptions, (res) => {
  console.log(`Server response: ${res.statusCode}`);
});

// using streams to read the data from the file, 
// and then compressing and sending each chunk as soon as it is read from the filesystem.
createReadStream(filename)
  .pipe(createGzip())
  .pipe(createCipheriv('aes192', secret, iv))
  .pipe(req)
  .on('finish', () => {
    console.log('File successfully sent');
  });

Using streams, when first chunk is received, it continues through the write process. When the next chunk is received, it continues through the write process as well in parallel as these tasks are asynchronous. Node.js streams take care of the order in which chunks are received. Thus, the entire process takes less time to complete.

# Stream basics

Every stream in Node.js is an implementation of one of the four base abstract classes available in the stream core module:

Readable
Writable
Duplex
Transform

Each stream class is also an instance of EventEmitter.

Streams support two operating modes:

Binary mode: To stream data in the form of chunks, such as buffers or strings
Object mode: To stream data as a sequence of discrete objects

# Readable streams

Reading from a stream

Non-flowing mode

This is the default operating mode for streams.

readable.read([size])


process.stdin
  .on("readable", () => {
    let chunk;
    console.log("New data available");
    // read the data continuously until the internal buffer is emptied
    // synchronous operation
    // chunk is a Buffer object if the stream is working in binary mode
    while((chunk = process.stdin.read()) !== null) {
      console.log(`Chunk read (${chunk.length} bytes): "${chunk.toString()}"`)
    }
  })
  .on("end", () => {
    console.log("End of stream");
  })

To read strings from the stream, call chunk.setEncoding(encoding)

Flowing mode


process.stdin
  .on("data", (chunk) => {
    console.log("New data available");
    console.log(
      `Chunk read (${chunk.length} bytes): "${chunk.toString()}"`
    )
  })
  // stop the stream from emitting data events
  // any incoming data to be cached is sent in the internal buffer until paused
  // switches to non-flowing mode
  .pause()
  // switches to flowing mode
  .resume()
  .on("end", () => {
    console.log("End of stream");
  });

Async iterators


async function main () {
  for await (const chunk of process.stdin) {
    console.log('New data available');
    console.log(
      `Chunk read (${chunk.length} bytes): "${chunk.toString()}"`
    );
  }
  console.log('End of stream');
}
main();

Readable stream implementation


import { Readable } from "stream";
import Chance from "chance";

const chance = new Chance();

export class RandomStream extends Readable {
  constructor(options) {
    super(options);
    this.emittedBytes = 0;
  }

  _read (size) {
    const chunk = chance.string({ length: size });
    this.push(chunk, "utf-8");
    this.emittedBytes += chunk.length;
    if (chance.bool({ likelihood: 5 })) {
      // End the stream
      this.push(null);
    }
  }
}

const randomStream = new RandomStream();
randomStream
  .on("data", (chunk) => {
    console.log(`Chunk received (${chunk.length} bytes): ${chunk.toString()}`);
  })
  .on("end", () => {
    console.log(`Produced ${randomStream.emittedBytes} bytes of random data`);
  });


interface StreamOptions<T extends Stream> extends Abortable {
  emitClose?: boolean | undefined;
  // The upper limit of the data stored in the internal buffer, after which no more reading from the source should be done
  // Defaults to 16KB
  highWaterMark?: number | undefined;
  // A flag to enable object mode -> defaults to false
  objectMode?: boolean | undefined;
  construct?(this: T, callback: (error?: Error | null) => void): void;
  destroy?(this: T, error: Error | null, callback: (error: Error | null) => void): void;
  autoDestroy?: boolean | undefined;
}

interface ReadableOptions extends StreamOptions<Readable> {
  // Used to convert buffers into strings -> defaults to null
  encoding?: BufferEncoding | undefined;
  read?(this: Readable, size: number): void;
}

⭐ Readable.from() -> to create Readable streams from arrays or other iterable objects (generators, iterators and async iterators).

Simplified readable


import { Readable } from 'stream';
import Chance from 'chance';
const chance = new Chance();

let emittedBytes = 0;
const randomStream = new Readable({
  read (size) {
    const chunk = chance.string({ length: size })
    this.push(chunk, 'utf8')
    emittedBytes += chunk.length
    if (chance.bool({ likelihood: 5 })) {
      this.push(null)
    }
  }
});

# Writable streams

writable.write(chunk, [encoding], [callback])

writable.end([chunk], [encoding], [callback]) -> here, the callback function is equivalent to registering a listener to the finish event, which is fired when all the data written in the stream has been flushed into the underlying resource.


import { createServer } from "http";
import Chance from "chance";

const chance = new Chance();
const server = createServer((req, res) => {
  res.writeHead(200, { 'Content-Type': 'text/plain' });
  while (chance.bool({ likelihood: 95 })) {
    res.write(`${chance.string()}\n`);
  }
  res.end("\n\n");
  res.on("finish", () => {
    console.log("All data sent");
  });
});

server.listen(8080, () => {
  console.log("listening on http://localhost:8080");
});

Backpressure

When the size of the internal buffer exceeds highWatermark, writable.write() returns false. Once the buffer is empty, the drain event is emitted, communicating that it's safe to start writing again.


const { createServer } = require("http");
const Chance = require("chance");

const chance = new Chance();

const server = createServer((req, res) => {
  res.writeHead(200, { 'Content-Type': 'text/plain' });
  function generateMore () {
    while (chance.bool({ likelihood: 95 })) {
      const randomChunk = chance.string({
        length: (16 * 1024) - 1
      });
      const shouldContinue = res.write(`${randomChunk}`);
      if (!shouldContinue) {
        console.log("back-pressure");
        return res.once("drain", generateMore);
      }
    }
    res.end("\n\n");
  }
  generateMore();
  res.on("finish", () => console.log('All data sent'));
});

Writable stream implementation


import { Writable } from "stream";
import { promises as fs } from "fs";
import { dirname, join } from "path";
import mkdirp from "mkdirp-promise";

class ToFileStream extends Writable {
  constructor(options) {
    super({ ...options, objectMode: true });
  }

  _write(chunk, encoding, cb) {
    mkdirp(dirname(chunk.path))
      .then(() => fs.writeFile(chunk.path, chunk.content))
      .then(() => cb())
      .catch(cb);
  }
}

const tfs = new ToFileStream();
tfs.write({
  path: join('files', 'file1.txt'), content: 'Hello' })
tfs.write({
  path: join('files', 'file2.txt'), content: 'Node.js' })
tfs.write({
  path: join('files', 'file3.txt'), content: 'streams' })
tfs.end(() => console.log('All files created'))

`ts

interface WritableOptions extends StreamOptions {
decodeStrings?: boolean | undefined;
defaultEncoding?: BufferEncoding | undefined;
write?(this: Writable, chunk: any, encoding: BufferEncoding, callback: (error?: Error | null) => void): void;
writev?(
this: Writable,
chunks: Array<{
chunk: any;
encoding: BufferEncoding;
}>,
callback: (error?: Error | null) => void
): void;
final?(this: Writable, callback: (error?: Error | null) => void): void;
}

Simplified writable

`js

const tfs = new Writable({
objectMode: true,
write (chunk, encoding, cb) {
mkdirp(dirname(chunk.path))
.then(() => fs.writeFile(chunk.path, chunk.content))
.then(() => cb())
.catch(cb)
};
});

# Duplex streams

A Duplex stream is a stream that is both Readable and Writable.

Used for an entity that is both a data source and a data destination
Inherit both from stream.Readable and stream.Writable
Both read() and write() methods are available
Can listen on both readable and drain events
To build a Duplex stream, we need to provide both _read() and _write() methods

Options of importance:

allowHalfOpen -> This option defaults to true, if set to false will cause both parts of the stream to end if only one of them does.
readableObjectMode
writableObjectMode

# Transform streams

Transform streams are a special kind of Duplex stream that are specifically designed to handle data transformations.

Examples: zlib.createGzip(), crypto.createCipheriv()

To build a Transform stream, we need to provide the following methods:

_read()
_write()
_transform()
_flush()

Transform stream implementation

Implement a Transform stream that replaces all the occurrences of a given string:

`js

import { Transform } from "stream";

export class ReplaceStream extends Transform {
constructor(searchStr, replaceStr, options) {
super({ ...options });
this.searchStr = searchStr;
this.replaceStr = replaceStr;
this.tail = "";
}

_transform(chunk, encoding, callback) {
const pieces = (this.tail + chunk).split(this.searchStr);
const lastPiece = pieces[pieces.length - 1];
const tailLen = this.searchStr.length - 1;
this.tail = lastPiece.slice(-tailLen);
pieces[pieces.length - 1] = lastPiece.slice(0, -tailLen);
this.push(pieces.join(this.replaceStr));
callback();
}

_flush(callback) {
this.push(this.tail);
callback();
}
}

const replaceStream = new ReplaceStream("
", ":)");
replaceStream.on("data", (chunk) => {
console.log(chunk.toString());
});
replaceStream.write("Hello
Wo");
replaceStream.write("rld
");
replaceStream.write("This is a test
");
replaceStream.end();

Filtering and aggregating data with Transform streams:

`csv

type,country,profit
Household,Namibia,597290.92
Baby Food,Iceland,808579.10
Meat,Russia,277305.60
Meat,Italy,413270.00
Cereal,Malta,174965.25
Meat,Indonesia,145402.40
Household,Italy,728880.54

`js

import { createReadStream } from "fs";
import { parse } from "csv-parse";
import { Transform } from "stream";

class FilterByCountry extends Transform {
constructor(country, options = {}) {
options.objectMode = true;
super(options);
this.country = country;
}

_transform(record, encoding, callback) {
if (record.country === this.country) {
this.push(record);
}
callback();
}
}

class SumProfit extends Transform {
constructor(options = {}) {
options.objectMode = true;
super(options);
this.total = 0;
}

_transform(record, encoding, callback) {
this.total += Number.parseFloat(record.profit);
callback();
}

_flush(callback) {
this.push(this.total.toString());
}
}

const csvParse = parse({ columns: true });

createReadStream("./data.csv")
.pipe(csvParse)
.pipe(new FilterByCountry("Italy"))
.pipe(new SumProfit())
.pipe(process.stdout);

# Passthrough Streams

Passthrough stream is a special type of Transform that outputs every data chunk without applying any transformation.

Observability

`js

import { PassThrough } from 'stream';

let bytesWritten = 0;

const monitor = new PassThrough();

monitor.on('data', (chunk) => {
bytesWritten += chunk.length;
});

monitor.on('finish', () => {
console.log(${bytesWritten} bytes written);
});

monitor.write('Hello!');
monitor.end();

createReadStream(filename)
.pipe(createGzip())
.pipe(monitor)
.pipe(createWriteStream(${filename}.gz));

Late piping

Use a PassThrough stream when you need to provide a placeholder for data that will be read or written in the future.

`js

import { createReadStream } from 'fs';
import { createBrotliCompress } from 'zlib';
import { PassThrough } from 'stream';
import { basename } from 'path';
import { upload } from './upload.js';

const filepath = process.argv[2];
const filename = basename(filepath);

const contentStream = new PassThrough();

upload(${filename}.br, contentStream)
.then((response) => {
console.log(Server response: ${response.data});
})
.catch((err) => {
console.error(err);
process.exit(1);
});

createReadStream(filepath)
.pipe(createBrotliCompress())
.pipe(contentStream);

Alternate implementation:

`js

function createUploadStream (filename) {
const connector = new PassThrough();
upload(filename, connector);
return connector;
}

const upload = createUploadStream('a-file.txt');
upload.write('Hello World');
upload.end();

# Lazy streams

❓ Why are lazy streams required?

Creating a stream instance might initialize expensive operations straight away (for example, open a file or a socket, initialize a connection to a database, and so on), even before we actually start to use such a stream. This might not be desirable if we are creating a large number of stream instances for later consumption. For this reason, we should consider lazy implementation of streams.

lazystream

This library allows us to effectively create proxies for actual stream instances, where the proxied instance is not created until some piece of code is actually starting to consume data from the proxy.

Custom implementation

`js

import { Readable, Writable } from 'stream';

class LazyStream {
constructor(streamFactory) {
this.streamFactory = streamFactory;
this.stream = null;
}

_initializeStream() {
if (!this.stream) {
this.stream = this.streamFactory();
}
}

createReadableStream() {
return new Readable({
read: (size) => {
this._initializeStream();
this.stream.on('data', (chunk) => {
this.push(chunk);
});
this.stream.on('end', () => {
this.push(null);
});
},
});
}

createWritableStream() {
return new Writable({
write: (chunk, encoding, callback) => {
this._initializeStream();
this.stream.write(chunk, encoding, callback);
},
});
}
}

module.exports = LazyStream;

// ------------------------------

const fs = require('fs');
const LazyStream = require('./LazyStream');

const createReadStream = () => {
return fs.createReadStream('path/to/file.txt');
};

const lazyReadStream = new LazyStream(createReadStream);

const readable = lazyReadStream.createReadableStream();
readable.pipe(process.stdout);

const createWriteStream = () => {
return fs.createWriteStream('path/to/output.txt');
};

const lazyWriteStream = new LazyStream(createWriteStream);

const writable = lazyWriteStream.createWritableStream();
process.stdin.pipe(writable);

# Pipes and error handling

The following implementation allows us to ensure that all the allocated resources are properly released, and the error is handled gracefully:

`js

function handleError (err) {
console.error(err);
stream1.destroy();
stream2.destroy();
}

stream1
.on('error', handleError)
.pipe(stream2)
.on('error', handleError);

Error handling with `pipeline`

pipeline(stream1, stream2, stream3, ... , cb)

All the streams are properly destroyed when the pipeline completes successfully or when it's interrupted by an error.

Example:

`js

import { createGzip, createGunzip } from "zlib";
import { Transform, pipeline } from "stream";

const uppercasify = new Transform({
transform (chunk, enc, cb) {
this.push(chunk.toString().toUpperCase());
cb();
}
});

pipeline(
process.stdin,
createGunzip(),
uppercasify,
createGzip(),
process.stdout,
(err) => {
if (err) {
console.error(err);
process.exit(1);
}
},
);

# Asynchronous control flow patterns with streams

Sequential execution

`js

import { createWriteStream, createReadStream } from "fs";
import { Readable, Transform } from "stream";

Unordered parallel execution

`js

import { Transform } from "stream";

class ParallelStream extends Transform {
constructor (userTransform, opts) {
super({ objectMode: true, ...opts });
this.userTransform = userTransform;
this.running = 0;
this.terminateCb = null;
}

_transform(chunk, enc, done) {
this.running++;
this.userTransform(
chunk,
enc,
this.push.bind(this),
this._onComplete.bind(this),
);
done();
}

_flush(done) {
if (this.running > 0) {
this.terminateCb = done;
} else {
done();
}
}

_onComplete (err) {
this.running--;
if (err) {
return this.on("error", err);
}
if (this.running === 0) {
this.terminateCb && this.terminateCb();
}
}
}

// -----------------------------------------

import { pipeline } from "stream";
import { createReadStream, createWriteStream } from "fs";
import split from "split";
import superagent from "superagent";

pipeline(
createReadStream(process.argv[2]),
split(),
new ParallelStream(
async (url, enc, push, done) => {
if (!url) {
return done();
}
try {
await superagent.head(url, { timeout: 5 * 1000 })
push(${url} is up\n);
} catch (err) {
push(${url} is down\n);
}
done();
},
),
createWriteStream('results.txt'),
(err) => {
if (err) {
console.error(err);
process.exit(1);
}
console.log('All urls have been checked');
}
)

Unordered limited parallel execution

`js

class LimitedParallelStream extends Transform {
constructor (concurrency, userTransform, opts) {
super({ objectMode: true, ...opts });
this.userTransform = userTransform;
this.concurrency = concurrency;
this.running = 0;
this.continueCb = null;
this.terminateCb = null;
}

_transform(chunk, enc, done) {
this.running++;
this.userTransform(
chunk,
enc,
this.push.bind(this),
this._onComplete.bind(this),
);
if (this.running < this.concurrency) {
done();
} else {
this.continueCb = done;
}
}

_flush(done) {
if (this.running > 0) {
this.terminateCb = done;
} else {
done();
}
}

_onComplete (err) {
this.running--;
if (err) {
return this.on("error", err);
}
const tmpCb = this.continueCb;
this.continueCb = null;
tmpCb && tmpCb();
if (this.running === 0) {
this.terminateCb && this.terminateCb();
}
}
}

Ordered parallel execution

parallel-transform - allows us to run our transforms in parallel without changing the order of the output.

parallel-transform implementation opts for predictable memory utilization and maintains an internal buffer that will not grow more than the specified maximum concurrency.

`js

import { pipeline } from "stream";
import { createReadStream, createWriteStream } from "fs";
import split from "split";
import parallelTransform from 'parallel-transform';

pipeline(
createReadStream(process.argv[2]),
split(),
parallelTransform(4, async function (url, done) {
if (!url) {
return done();
}
console.log(url);
try {
await request.head(url, { timeout: 5 * 1000 });
this.push(${url} is up\n);
} catch (err) {
this.push(${url} is down\n);
}
done();
}),
createWriteStream('results.txt'),
(err) => {
if (err) {
console.error(err);
process.exit(1);
}
console.log('All urls have been checked');
}
);

# Piping patterns

Combining streams

Advantages of a combined stream:

We can redistribute it as a black box by hiding its internal pipeline.
We have simplified error management, as we don't have to attach an error listener to each stream in the pipeline, but just to the combined stream itself.

const combinedStream = pumpify(streamA, streamB, streamC) - pumpify

`js

import { createGzip, createGunzip } from "zlib";
import {
createCipheriv,
createDecipheriv,
scryptSync,
randomBytes,
} from "crypto";
import { Transform } from "stream";
import pumpify from "pumpify";

function createPassword(password) {
return scryptSync(password, "salt", 24);
}

function generateIV() {
return randomBytes(16);
}

class PrependIV extends Transform {
constructor(iv, options) {
super(options);
this.iv = iv;
this.ivPrepended = false;
}

_transform(chunk, _, done) {
if (!this.ivPrepended) {
this.push(iv);
this.ivPrepended = true;
}
this.push(chunk);
done();
}
}

function createCompressAndEncrypt(password) {
const key = createKey(password);
const iv = generateIV();
const prependedStream = new PrependIV(iv);
const combinedStream = pumpify(
prependedStream,
createGzip(),
createCipheriv("aes192", key, iv)
);
return combinedStream;
}

`js

// COMPRESS

const [,, password, source] = process.argv;
const destination = ${source}.gz.enc;

pipeline(
createReadStream(source),
createCompressAndEncrypt(password),
createWriteStream(destination),
(err) => {
if (err) {
console.error(err)
process.exit(1)
}
console.log(${destination} created)
}
);

`js

// DECOMPRESS

const [,, password, source] = process.argv;
const destination = source.replace('.gz.enc', '.out');

pipeline(
createReadStream(source),
createDecryptAndDecompress(password),
createWriteStream(destination),
(err) => {
if (err) {
console.error(err)
process.exit(1)
}
console.log(${destination} created)
}
);

Forking streams

This is useful:

when we want to send the same data to different destinations
when we want to perform different transformations on the same data
when we want to split the data based on some criteria

`js

import { createReadStream, createWriteStream } from 'fs';
import { createHash } from 'crypto';

const filename = process.argv[2];
const sha1Stream = createHash('sha1').setEncoding('hex');
const md5Stream = createHash('md5').setEncoding('hex');
const inputStream = createReadStream(filename);

inputStream
.pipe(sha1Stream)
.pipe(createWriteStream(${filename}.sha1));

inputStream
.pipe(md5Stream)
.pipe(createWriteStream(${filename}.md5));

Both streams will end automatically unless { end: false } is specified.
The two forks of the stream will receive the same data chunks.
If one destination pauses the source stream to handle backpressure for a long time, all the other destinations will be waiting as well.
If we pipe to an additional stream after we've started consuming the data at source (async piping), the new stream will only receive new chunks of data.

Merging streams

Use the option { end: false } when piping multiple sources to a single destination and then invoke end() on the destination only when all the sources have completed reading.

`js

import { createReadStream, createWriteStream } from 'fs';
import split from 'split';

const dest = process.argv[2];
const sources = process.argv.slice(3);

const destStream = createWriteStream(dest);

To merge streams in order, we can modify the code as follows:

`js

import { createReadStream, createWriteStream } from 'fs';
import split from 'split';

const dest = process.argv[2];
const sources = process.argv.slice(3);

const destStream = createWriteStream(dest);

function processStream(index) {
if (index >= sources.length) {
destStream.end();
console.log(${dest} created);
return;
}

const sourceStream = createReadStream(sources[index], { highWaterMark: 16 });
sourceStream.on("end", () => processStream(index + 1));
sourceStream
.pipe(split(line => line + "\n"))
.pipe(destStream, { end: false })
}

processStream(0);

multistream - package that deals with merging streams in order.

Multiplexing and demultiplexing

The operation of combining multiple streams (in this case, also known as channels) to allow transmission over a single stream is called multiplexing. The process of reconstructing the original streams from the data received from a shared stream, is called demultiplexing. The devices that perform these operations are called multiplexer (or mux) and demultiplexer (or demux), respectively.

Multiplexing

`js

import { fork } from "child_process";
import { connect } from "net";

// Create a TCP client connection to localhost:3000
const socket = connect(3000, () => {
const child = fork(
process.argv[2],
process.argv.slice(3),
// ensures child process does not inherit stdout and stderr of the parent
{ silent: true }
);
multiplexChannels([child.stdout, child.stderr], socket);
})

Demultiplexing

`js

import { createWriteStream } from "fs";
import { createServer } from "net";

function demultiplexChannel(source, destinations) {
let currentChannel = null;
let currentLength = null;

source
.on("readable", () => {
let chunk;
if (currentChannel === null) {
chunk = source.read(1);
currentChannel = chunk && chunk.readUInt8(0);
}
if (currentLength === null) {
chunk = source.read(4);
currentLength = chunk && chunk.readUInt32BE(0);
if (currentLength === null) {
return null;
}
chunk = source.read(currentLength);
if (chunk === null) {
return null;
}

    console.log(`Received packet from: ${currentChannel}`);
    destinations[currentChannel].write(chunk);
    currentChannel = null;
    currentLength = null;
  }
})
.on("end", () => {
  destinations.forEach(destination => {
    destination.end();
  });
  console.log("Source channel closed");
});

}

const server = createServer(socket => {
const stdoutStream = createWriteStream('stdout.log');
const stderrStream = createWriteStream('stderr.log');
demultiplexChannel(socket, [stdoutStream, stderrStream]);
});

server.listen(3000, () => console.log("Server started"));

ternary-stream - implement an if...else statement for streams.

Asynchronous Control Flow Patterns

Manoj Kumar Patra — Wed, 02 Oct 2024 08:57:58 +0000

Basic principles that can help us keep the nesting level low and improve the organization of our code in general in callbacks:

Exit as soon as possible.
Create named functions for callbacks, keeping them out of closures and passing intermediate results as arguments.
Modularize the code.

# Sequential execution

Executing a set of tasks in sequence means running them one at a time, one after the other.

Different variations of this flow:

Executing a set of known tasks in sequence, without propagating data across them.
Using the output of a task as the input for the next (also known as chain, pipeline, or waterfall).
Iterating over a collection while running an asynchronous task on each element, one after the other.

Executing a known set of tasks in sequence


function task1 (cb) {
  asyncOperation(() => {
    task2(cb);
  });
}

function task2 (cb) {
  asyncOperation(() => {
    task3(cb);
  });
}

function task3 (cb) {
  asyncOperation(() => {
    cb();
  });
}

task1(() => {
  // executed when task1, task2 and task3 are completed
  console.log('tasks 1, 2 and 3 executed')
});

Sequential iteration

This pattern is helpful when we want to execute an asynchronous operation for each item in a collection.

Pattern with callback


function iterate (index) {
  if (index === tasks.length) {
    return cb();
  }

  const task = tasks[index];
  task(() => {
    iterate(index + 1);
  });
}

function cb () {
  // iteration completed
}

iterate(0);

Pattern with Promise

Promisify a callback based API


function promisify (cbBasdApi) {
  return function promisified (...args) {
    return new Promise((resolve, reject) => {
      const newArgs = [
        ...args,
        function (err, result) {
          if (err) {
            return reject(err);
          }

          resolve(result);
        }
      ];

      cbBasedApi(...newArgs);
    });
  };
}


function iterate(index) {
  let promise = Promise.resolve();

  if (index === tasks.length) {
    return promise.then(finish);
  }

  for (const task of tasks) {
    promise = promise.then(() => iterate(index + 1));
  }

  return promise;
}

function finish() {
  // iteration completed
  console.log('All tasks completed');
}


const promise = tasks.reduce((prev, task) => {
  return prev.then(() => {
    return task();
  });
}, Promise.resolve());

❗ This algorithm becomes recursive if task() is a synchronous operation. There might be a risk of hitting the maximum call stack size limit.

⭐ This pattern can be used to create asynchronous versions of Array methods.


function iterateSeries(collection, iteratorCallback, finalCallback) {
  if (collection.length === 0) {
    finalCallback();
    return;
  }

  const currentItem = collection.shift();

  iteratorCallback(currentItem, () => {
    iterateSeries(collection, iteratorCallback, finalCallback);
  });
}

// Example
const collection = [1, 2, 3, 4, 5];

function iteratorCallback(item, callback) {
  // Simulate asynchronous operation (e.g., API call, file I/O) with setTimeout
  setTimeout(() => {
      console.log(item);
      callback(); // Call the callback to proceed to the next iteration
  }, 1000); // Delay of 1 second for demonstration
}

function finalCallback() {
  console.log("All iterations are complete");
}

iterateSeries(collection, iteratorCallback, finalCallback);

# Parallel execution

In parallel execution, we are not concerned about the order of execution of a set of asynchronous tasks. Instead, all we need is to get notified when all those running tasks are completed.

Pattern with callback


const tasks = [ /* ... */ ];

let completed = 0;
tasks.forEach(task => {
  task(() => {
    if (++completed === tasks.length) {
      finish(); 
    }
  }); 
});

function finish () {
  // all the tasks completed
}

Pattern with Promise


const tasks = [/* ... */];

// Map each task to a Promise that resolves when the task is completed
const promises = tasks.map(task => new Promise(resolve => task(resolve)));

Promise.all(promises)
  .then(() => {
    cb();
  })
  .catch(error => {
    cb(error);
  });

function cb () {
  // all the tasks completed
}

The Unlimited Parallel Execution pattern

The Unlimited Parallel Execution pattern, also known as Parallel Limitless, allows you to execute multiple asynchronous tasks concurrently without any predefined limit on the number of tasks running simultaneously. This pattern is useful when you have a large number of asynchronous operations to perform, such as making multiple API requests, and you want to maximize efficiency by executing as many tasks in parallel as possible without overwhelming the system.


async function unlimitedParallelExecution(tasks, taskExecutor) {
  const promises = [];

  for (const task of tasks) {
    promises.push(taskExecutor(task));
  }

  await Promise.all(promises);
}

// Example:

async function taskExecutor(task) {
  // Simulate asynchronous operation (e.g., API call, file I/O) with setTimeout
  return new Promise((resolve) => {
    setTimeout(() => {
      console.log('Task completed:', task);
      resolve();
    }, Math.random() * 2000); // Random delay up to 2 seconds for demonstration
  });
}

const tasks = ['Task 1', 'Task 2', 'Task 3', 'Task 4', 'Task 5'];

unlimitedParallelExecution(tasks, taskExecutor)
  .then(() => {
    console.log('All tasks completed');
  })
  .catch((error) => {
    console.error('Error:', error);
  });

Fixing race conditions

To avoid race conditions, all we need to do is to introduce a variable of some type to mutually exclude multiple similar tasks.

# Limited parallel execution

A server that spawns unbounded parallel tasks to handle a user request could be exploited with a denial-of-service (DoS) attack.

Limiting the number of parallel tasks is, in general, a good practice that helps with building resilient applications.

Limiting concurrency


const tasks = [/* ... */];

const concurrencyLimit = 3;

async function executeTasksWithLimit(tasks, concurrencyLimit) {
  const runningTasks = [];
  const results = [];

  const runTask = task => {
    task((err, res) => {
      if (err) {
        results.push({ error });
        return;
      }
      results.push(res);
    });

    // Remove the task from the list of running tasks
    runningTasks.splice(runningTasks.indexOf(task), 1);
    // If there are more tasks to run, start the next one
    if (tasks.length > 0) {
      startNextTask();
    }
  };

  const startNextTask = () => {
    const task = tasks.shift(); // Get the next task
    if (task) {
      runningTasks.push(task);
      runTask(task);
    }
  };

  for (let i = 0; i < concurrencyLimit && i < tasks.length; i++) {
    startNextTask();
  }
}

Limited concurrency pattern works very well when we have a predetermined set of tasks to execute, or when the set of tasks grows linearly over time.

Limited concurrency with queues


import { EventEmitter } from 'events';

export class TaskQueue {
  constructor (concurrency) {
    super();
    this.concurrency = concurrency;
    this.running = 0;
    this.queue = [];
  }

  runTask (task) {
    return new Promise((resolve, reject) => {
      this.queue.push(() => {
        return task().then(resolve, reject);
      });
      process.nextTick(this.next.bind(this));
    });
  }

  next () {
    if (this.running === 0 && this.queue.length === 0) {
      return this.emit("empty");
    }

    while (this.running < this.concurrency && this.queue.length) {
      const task = this.queue.shift();
      task()
        .catch(err => {
          this.emit("error", err);
        })
        .finally(() => {
          this.running--;
          process.nextTick(this.next.bind(this));
        });
      this.running++;
    }
  }
}

Example of spider web app

Sequential execution and iteration


function spiderLinks (currentUrl, content, nesting, queue) {
  let promise = Promise.resolve();
  if (nesting === 0) {
    return promise;
  }

  const links = getPageLinks(currentUrl, content);
  for (const link of links) {
    promise = promise.then(() =>
      spiderTask(link, nesting - 1, queue));
  }

  return promise;
}

Parallel execution


function spiderLinks (currentUrl, content, nesting, queue) {
  if (nesting === 0) {
    return Promise.resolve();
  }

  const links = getPageLinks(currentUrl, content);
  const promises = links
    .map(link => spiderTask(link, nesting - 1, queue));
  return Promise.all(promises);
}


const spidering = new Set();

function spiderTask (url, nesting, queue) {
  if (spidering.has(url)) {
    return Promise.resolve();
  }

  spidering.add(url);

  const fileName = urlToFilename(url);

  return queue
    .runTask(() => {
      return fsPromises.readFile(filename, 'utf8')
        .catch(err => {
          if (err.code !== 'ENOENT') {
            throw err;
          }

          return download(url, fileName);
        });
    })
    .then(content => spiderLinks(url, content, nesting, queue));
}

export function spider (url, nesting, concurrency) {
  const queue = new TaskQueue(concurrency);
  return spiderTask(url, nesting, queue);
}

Refresher on Promises

new Promise((resolve, reject) => {}) creates a new Promise instance that fulfills or rejects based on the behavior of the function provided as an argument.

Useful Promise methods:

Promise.resolve(obj)
Promise.reject(err)
Promise.all(iterable)

Promise.allSettled(iterable) => returns an object with structure


{
  status: 'fulfilled' | 'rejected';
  reason?: any; // present in case of rejection
  value?: any; // present in case of fulfillment
}

Promise.race(iterable)
promise.then(onFulfilled, onRejected)
promise.catch(onRejected)
promise.finally(onFinally)

Refresher on async/await

In an async function, at each await expression, the execution of the function is put on hold, its state saved, and the control returned to the event loop. Once the Promise that has been awaited resolves, the control is given back to the async function, returning the fulfilment value of the Promise.

❗❗ BEWARE


links.forEach(async function iteration(link) {
  await spider(link, nesting - 1);
});

This piece of code will start the tasks in parallel instead of sequential iteration and the execution continues immediately after invoking forEach(), without waiting for all the tasks to complete.

Limited parallel execution with async/await


export class TaskQueuePC {
  constructor (concurrency) {
    this.taskQueue = [];
    this.consumerQueue = [];

    // Spawn as many consumers as the concurrency we want to attain
    for (let i = 0; i < concurrency; i++) {
      this.consumer();
    }
  }

  async consumer () {
    while (true) {
      try {
        const task = await this.getNextTask();
        await task();
      } catch (err) {
        console.error(err);
      }
    }
  }

  async getNextTask () {
    return new Promise((resolve) => {
      if (this.taskQueue.length !== 0) {
        return resolve(this.taskQueue.shift());
      }

      // If queue is empty, then we postpone the resolution of the Promise by queuing the resolve callback into the consumerQueue.
      this.consumerQueue.push(resolve);
    });
  }

  runTask (task) {
    return new Promise((resolve, reject) => {
      const taskWrapper = () => {
        const taskPromise = task();
        taskPromise.then(resolve, reject);
        return taskPromise;
      };

      // If the consumerQueue is not empty, then there is at least one consumer that is asleep, waiting for a new task to run.
      if (this.consumerQueue.length !== 0) {
        const consumer = this.consumerQueue.shift();
        consumer(taskWrapper);
      // If the consumerQueue is empty, means all the consumers are busy
      } else {
        this.taskQueue.push(taskWrappeer);
      }
    });
  }
}

# Tackling a problem of memory leak associated with Promises

Consider the following code snippets:


function leakingLoop () {
  return delay(1)
    .then(() => {
      console.log(`Tick ${Date.now()}`);
      return leakingLoop();
    });
}

async function leakingLoopAsync () {
  await delay(1);
  console.log(`Tick ${Date.now()}`);
  return leakingLoopAsync();
}

This situation creates a chain of promises that never settle, and it will cause a memory leak in Promise implementations.

The solution is to break the chain of Promise resolution while considering error handling from anywehere deep in the recursion as follows:


function nonLeakingLoop () {
  return new Promise((resolve, reject) => {
    (function internalLoop () {
      delay(1)
        .then(() => {
          console.log(`Tick ${Date.now()}`);
          internalLoop();
        })
        .catch(err => {
          reject(err);
        });
    })();
  });
}

Callbacks and Events

Manoj Kumar Patra — Wed, 02 Oct 2024 08:56:00 +0000

# The callback pattern

Continuous-passing style

In JavaScript, a callback is a function that is passed as an argument to another function and is invoked with the result when the operation completes. In functional programming, this way of propagating the result is called continuation-passing style (CPS).

An unpredictable function:


import { readFile } from "fs";
const cache = new Map();
function inconsistentRead(filename, cb) {
  if (cache.has(filename)) {
    // invoked synchronously
    cb(cache.get(filename));
  } else {
    // asynchronous function
    readFile(filename, "utf8", (err, data) => {
      cache.set(filename, data);
      cb(data);
    });
  }
}

Using synchronous APIs

Always choose a direct style for purely synchronous functions.

❗ Use blocking APIs sparingly and only when they don't affect the ability of the application to handle concurrent asynchronous operations.


import { readFileSync } from "fs";
const cache = new Map();
function consistentReadSync(filename) {
  if (cache.has(filename)) {
    return cache.get(filename);
  } else {
    const data = readFileSync(filename, "utf8");
    cache.set(filename, data);
    return data;
  }
}

Using synchronous I/O in Node.js is strongly discouraged in many circumstances, but in some situations, this might be the easiest and most efficient solution. Example: It makes perfect sense to use a synchronous blocking API to load a configuration file while bootstrapping an application.

Asynchronous behaviour with deferred execution


import { readFile } from "fs";
const cache = new Map();
function consistentReadAsync(filename, callback) {
  if (cache.has(filename)) {
    // deferred callback invocation
    process.nextTick(() => callback(cache.get(filename)));
  } else {
    // asynchronous function
    readFile(filename, "utf8", (err, data) => {
      cache.set(filename, data);
      callback(data);
    });
  }
}

A callback is invoked asynchronously by deferring its execution using process.nextTick().

`process.nextTick() vs setImmediate()`

Callbacks deferred with process.nextTick() are called microtasks and they are executed just after the current operation completes, even before any other I/O event is fired.

Callbacks with setImmediate() are queued in an event loop phase that comes after all I/O events have been processed.

Since, process.nextTick() runs before any already scheduled I/O, it will be executed faster, but under certain circumstances, it might also delay the running of any I/O callback indefinitely (also known as I/O starvation), such as in the presence of a recursive invocation. This can never happen with setImmediate().

setImmediate() are executed faster than those scheduled with setTimeout(callback, 0).

NodeJS callback convention

readFile(filename, [options], callback)

callback is of the format (err, data) => { ... }. Here, err must always be of type Error.

Fail-fast approach


process.on("uncaughtException", (err) => {
  console.error(`This will catch at last the JSON parsing exception: ${err.message}`);
  // Terminates the application with 1 (error) as exit code.
  // Without the following line, the application would continue
  process.exit(1);
});

An uncaught exception leaves the application in a state that is not guaranteed to be consistent, which can lead to unforeseeable problems. That's why it is always advised, especially in production, to never leave the application running after an uncaught exception is received. Instead, the process should exit immediately, optionally after having run some necessary cleanup tasks, and ideally, a supervising process should restart the application. This is known as the fail-fast approach.

# The observer pattern

The Observer pattern defines an object (called subject) that can notify a set of observers (or listeners) when a change in its state occurs.

📖 The main difference from the Callback pattern is that the subject can actually notify multiple observers, while a traditional CPS callback will usually propagate its result to only one listener, the callback.


import { EventEmitter } from 'events'
const emitter = new EventEmitter()

on(event, listener): EventEmitter

once(event, listener): EventEmitter

emit(event, [arg1], [...]): EventEmitter

removeListener(event, listener): EventEmitter

Propagating errors

The convention is to emit a special event, called error, and pass an Error object as an argument.

Extending `EventEmitter` class

To make any object observable, have the class extend the EventEmitter class.

Examples: Node.js streams, HTTP Module

When the count of listeners registered to an event exceeds a specific amount (by default, 10), the EventEmitter will produce a warning. Sometimes, registering more than 10 listeners is completely fine, so we can adjust this limit by using the setMaxListeners() method of the EventEmitter.

Example:


class FindRegex extends EventEmitter {
   constructor(regex) {
      super();
      this.regex = regex;
      this.files = [];
   }
   addFile(file) {
      this.files.push(file);
      return this;
   }
   find() {
      for (const file of this.files) {
        readFile(file, 'utf8', (err, content) => {
          if (err) {
              return this.emit('error', err);
          }
          this.emit('fileread', file);
          const match = content.match(this.regex);
          if (match) {
              match.forEach(elem => this.emit('found', file, elem));
          }
        });
      }
   }
   return this;
}

Using the above code:


const findRegexInstance = new FindRegex(/hello \w+/);
findRegexInstance
  .addFile('fileA.txt')
  .addFile('fileB.json')
  .find()
  .on('found', (file, match) => console.log(`Matched "${match}" in file ${file}`))
  .on('error', err => console.error(`Error emitted ${err.message}`));

Dealing with event emitters and memory leaks

When subscribing to observables with a long life span, it is extremely important that we unsubscribe our listeners once they are no longer needed to avoid memory leaks.

We can use the convenience method once(event, listener) in place of on(event, listener) to automatically unregister a listener after the event is received for the first time. However, be advised that if the event we specify is never emitted, then the listener is never released, causing a memory leak.

When events are emitted asynchronously, we can register new listeners, even after the task that produces the events is triggered, up until the current stack yields to the event loop. This is because the events are guaranteed not to be fired until the next cycle of the event loop, so we can be sure that we won't miss any events.

# Difference between callbacks and event emitters

The general differentiating rule is semantic: callbacks should be used when a result must be returned in an asynchronous way, while events should be used when there is a need to communicate that something has happened.

# References

The module system

Manoj Kumar Patra — Wed, 02 Oct 2024 08:55:11 +0000

# Why are modules required?

Having a way to split the codebase into multiple files.
Allowing code reuse across different projects.
Encapsulation (or information hiding).
Managing dependencies.

Module vs module system

💡 Module is the actual unit of software, while a module system is the syntax and the tooling that allows us to define modules and to use them within our projects.

ECMAScript 6

ECMAScript 6 defined only the formal specification for ESM in terms of syntax and semantics, but it didn't provide any implementation details.

Browser companies and the Node.js community were responsible for implementing the ESM.

Node.js ships with stable support for ESM starting from version 13.2.

# The revealing module pattern


const myModule = (() => {
  const privateFoo = () => {};
  const privateBar = [];
  const exported = {
    publicFoo: () => {},
    publicBar: () => {},
  };
  return exported;
})();

Here, we use an Immediately Invoked Function Expression (IIFE) to create a private scope, exporting only the parts that are meant to be public.

# CommonJS modules

require is a function that allows you to import a module from the local filesystem
exports and module.exports are special variables that can be used to export public functionality from the current module

Module loader implementation


function loadModule(filename, module, require) {
  const wrappedSrc = `(function (module, exports, require) {
      ${fs.readFileSync(filename, "utf8")}
    })(${module}, ${module.exports}, ${require})`;
  eval(wrappedSrc);
}

Here, we are using readFileSync to read the module's content. While it is generally not recommended to use the synchronous version of the filesystem APIs, here it makes sense to do so. The reason for that is that loading modules in CommonJS are deliberately synchronous operations. This approach makes sure that, if we are importing multiple modules, they (and their dependencies) are loaded in the right order.

`require` implementation


function require(moduleName) {
  console.log(`Require invoked for module: ${moduleName}`);
  const id = require.resolve(moduleName);
  if (require.cache[id]) {
    return require.cache[id].exports;
  }
  // module metadata
  const module = {
    exports: {},
    id,
  };
  // Update the cache
  require.cache[id] = module; // load the module
  loadModule(id, module, require); // return exported variables
  return module.exports;
}

require.cache = {};
require.resolve = (moduleName) => {
  /* resolve a full module id from the moduleName */
};

require() function is synchronous.

Everything inside a module is private unless it's assigned to the module.exports variable. The content of this variable is then cached and returned when the module is loaded using require().

`module.exports` vs `exports`

The exports variable is just a reference to the initial value of module.exports.
We can only attach new properties to the object referenced by the exports variable.
Reassigning the exports variable doesn't have any effect, because it doesn't change the content of module.exports. It will only reassign the variable itself.
If we want to export something other than an object literal, such as a function, an instance, or even a string, we have to reassign module.exports.

# The resolving algorithm

File modules:
- moduleName starts with / => absolute path to the module and returned as it is.
- moduleName starts with ./ => a relative path, which is calculated starting from the directory of the requiring module.
Core modules: moduleName not prefixed with / or ./ => the algorithm will first try to search within the core Node.js modules.
Package modules: If no core module is found matching moduleName, then the search continues by looking for a matching module in the first node_modules directory that is found navigating up in the directory structure starting from the requiring module. The algorithm continues to search for a match by looking into the next node_modules directory up in the directory tree, until it reaches the root of the filesystem.
For file and package modules, both files and directories can match moduleName as follows:
- <moduleName>.js
- <moduleName>/index.js
- The directory/file specified in the main property of <moduleName>/package.json

The module cache

Each module is only loaded and evaluated the first time it is required, since any subsequent call of require() will simply return the cached version.

The module cache is exposed via the require.cache variable.

Circular dependencies


// Module a.js
exports.loaded = false;
const b = require("./b");
module.exports = {
  b,
  loaded: true,
};

// Module b.js
exports.loaded = false;
const a = require("./a");
module.exports = {
  a,
  loaded: true,
};

// main.js
const a = require("./a");
const b = require("./b");
console.log("a ->", JSON.stringify(a, null, 2));
console.log("b ->", JSON.stringify(b, null, 2));

# Module definition patterns

Named exports


// module.js
exports.info = (message) => {
  console.log(`info: ${message}`);
};

// app.js
const logger = require("./logger");
logger.info("This is an informational message");

Exporting a function


// module.js
module.exports = (message) => {
  console.log(`info: ${message}`);
};

module.exports.verbose = (message) => {
  console.log(`verbose: ${message}`);
};

// app.js
const logger = require("./logger");
logger("This is an informational message");
logger.verbose("This is a verbose message");

Exporting a class


class Logger {
  constructor(name) {
    this.name = name;
  }
  log(message) {
    console.log(`[${this.name}] ${message}`);
  }
  info(message) {
    this.log(`info: ${message}`);
  }
  verbose(message) {
    this.log(`verbose: ${message}`);
  }
}
module.exports = Logger;

Exporting an instance


class Logger {
  constructor (name) {
    ...
  }
  log (message) {
    ...
  }
}
module.exports = new Logger('DEFAULT');

NOTE: Multiple instances can still be created using the constructor property as follows:


const customLogger = new logger.constructor('CUSTOM');
customLogger.log('This is an informational message');

Monkey patching (Modifying other modules on the global scope)


// patcher.js
require('./logger').customMessage = function () {
  console.log('This is a new functionality');
};

// app.js
require('./patcher');
const logger = require('./logger');
logger.customMessage();

We should avoid using this technique as much as possible.

A real-life use case for this technique would be to mock the http module when writing test cases so that it provides mocked responses instead of a real HTTP request. Example: nock module.

# ESM: EcmaScript Modules

Difference between ESM and CommonJS modules

ESMs are static.

import statements are described at the top level of every module and outside any control flow statement.
The name of the imported ESMs cannot be dynamically generated at runtime using expressions, only constant strings are allowed.

Benefits of static imports

static analysis of the dependency tree
dead code elimination (tree shaking)

Using ESM in Node.js

.mjs file extension
In package.json, add field type with value module

Named exports and imports

In an ESM, everything is private by default and only entities exported with the export keyword are publicly accessible from other modules.


export function log (message) {
  console.log(message);
}

Namespace import


import * as loggerModule from './logger.js';

When we use this type of import statement, the entities are imported into the current scope, so there is a risk of a name clash.


import { log, Logger } from './logger.js';

To resolve this clash,


import { log as log2 } from './logger.js';

With ESM, it is important to specify the file extension for the imported module.

Default exports and imports

To export a single unnamed entity like in CommonJS with module.exports, we can do the following:


// The entity exported is registered under the name `default`
export default class Logger {
  ...
}

// Import
import MyLogger from './logger.js';

❓ What happens if we do import { default } from './logger.js'; ??

The execution will fail with a SyntaxError: Unexpected reserved word error. This happens because the default keyword cannot be used as a variable name directly in the scope. However, we can do import * as loggerModule from './logger.js';.

Named exports are better than default export

Named exports are explicit and thus, allows IDEs to support the developer with automatic imports, autocomplete, and refactoring tools.
Default exports might make it harder to apply dead code elimination (tree shaking).

Module identifiers

Relative specifiers like ./logger.js or ../logger.js
Absolute specifiers like file:///opt/nodejs/config.js
Bare specifiers - modules in the node_modules folder
Deep import specifiers - refer to a path within a package in node_modules

Async imports (Dynamic imports)

Async imports can be performed at runtime using the special import() operator.


import(dynamicModule)
  .then((res) => ...)

ESM loading process

Create a dependency graph - the dependency graph is needed by the interpreter to figure out how modules depend on each other and in what order the code needs to be executed. The starting point for the dependency resolution is called the entry point.
- Phase 1 - Construction (or parsing)
- Find all the imports and recursively load the content of every module from the respective file. This is done in a depth-first manner. Modules already visited in this process are ignored on revisits.
- Phase 2 - Instantiation
- The interpreter walks the tree view obtained from the previous phase from the bottom to the top. For every exported entity in each module, it keeps a named reference in memory, but doesn't assign any value just yet. After this, the interpreter will do another pass to link the exported names to the modules importing them.
- Phase 3 - Evaluation
- Node.js finally executes the code so that all the previously instantiated entities can get an actual value. The execution order is bottom-up respecting the post-order depth-first visit of our original dependency graph.

Loading process difference between CommonJS and ESM

CommonJS will execute all the files while the dependency graph is explored whereas in ESM, no code can be executed until the dependency graph has been fully built, and therefore module imports and exports have to be static.

Read-only live bindings

In ESM, when an entity is imported in the scope, the binding to its original value cannot be changed (read-only binding) unless the bound value changes within the scope of the original module itself (live binding), which is outside the direct control of the consumer code.

How is this different from CommonJS?

In CommonJS, the entire exports object is copied (shallow copy) when required from a module. This means that, if the value of primitive variables like numbers or string is changed at a later time, the requiring module won't be able to see those changes.

In ESM, imported modules are tracked as references, so, we can be sure every module has an up-to-date picture of the other modules, even in the presence of circular dependencies.

Monkey patching with ESM


import fs from 'fs';

export function mockEnable (respondWith) {
  mockedResponse = respondWith;
  fs.readFile = () => "Hello, world!";
}

The following statements won't work as they would give read-only live bindings:


import * as fs from 'fs';
import { readFile } from 'fs';

Another approach would be to use syncBuiltinESMExports:


import fs, { readFileSync } from 'fs';
import { syncBuiltinESMExports } from 'module';

fs.readFileSync = () => Buffer.from('Hello, ESM');
syncBuiltinESMExports();

console.log(fs.readFileSync === readFileSync); // true

syncBuiltinESMExports works only for built-in Node.js modules.

# More differences between ESM and CommonJS modules

ESM runs in strict mode, unlike CommonJS, where we need to add "use strict" statements at the beginning of the file.
require, exports, module.exports, __filename, and __dirname are not defined in ESM.

To get the values for the same, we can do the following:


// Create __filename and __dirname
import { fileURLToPath } from 'url';
import { dirname } from 'path';

// import.meta.url is a reference to the current module
// format: file:///path/to/current_module.js
const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);

// Create require
import { createRequire } from 'module';
const require = createRequire(import.meta.url);

this is undefined in the global scope of ESM, whereas in CommonJS this refers to exports.
ESM cannot import JSON files directly as modules. To do this, use the create require approach as mentioned above.

The module system

Manoj Kumar Patra — Wed, 02 Oct 2024 08:55:11 +0000

# Why are modules required?

Having a way to split the codebase into multiple files.
Allowing code reuse across different projects.
Encapsulation (or information hiding).
Managing dependencies.

Module vs module system

💡 Module is the actual unit of software, while a module system is the syntax and the tooling that allows us to define modules and to use them within our projects.

ECMAScript 6

ECMAScript 6 defined only the formal specification for ESM in terms of syntax and semantics, but it didn't provide any implementation details.

Browser companies and the Node.js community were responsible for implementing the ESM.

Node.js ships with stable support for ESM starting from version 13.2.

# The revealing module pattern


const myModule = (() => {
  const privateFoo = () => {};
  const privateBar = [];
  const exported = {
    publicFoo: () => {},
    publicBar: () => {},
  };
  return exported;
})();

Here, we use an Immediately Invoked Function Expression (IIFE) to create a private scope, exporting only the parts that are meant to be public.

# CommonJS modules

require is a function that allows you to import a module from the local filesystem
exports and module.exports are special variables that can be used to export public functionality from the current module

Module loader implementation


function loadModule(filename, module, require) {
  const wrappedSrc = `(function (module, exports, require) {
      ${fs.readFileSync(filename, "utf8")}
    })(${module}, ${module.exports}, ${require})`;
  eval(wrappedSrc);
}

`require` implementation


function require(moduleName) {
  console.log(`Require invoked for module: ${moduleName}`);
  const id = require.resolve(moduleName);
  if (require.cache[id]) {
    return require.cache[id].exports;
  }
  // module metadata
  const module = {
    exports: {},
    id,
  };
  // Update the cache
  require.cache[id] = module; // load the module
  loadModule(id, module, require); // return exported variables
  return module.exports;
}

require.cache = {};
require.resolve = (moduleName) => {
  /* resolve a full module id from the moduleName */
};

require() function is synchronous.

Everything inside a module is private unless it's assigned to the module.exports variable. The content of this variable is then cached and returned when the module is loaded using require().

`module.exports` vs `exports`

The exports variable is just a reference to the initial value of module.exports.
We can only attach new properties to the object referenced by the exports variable.
Reassigning the exports variable doesn't have any effect, because it doesn't change the content of module.exports. It will only reassign the variable itself.
If we want to export something other than an object literal, such as a function, an instance, or even a string, we have to reassign module.exports.

# The resolving algorithm

File modules:
- moduleName starts with / => absolute path to the module and returned as it is.
- moduleName starts with ./ => a relative path, which is calculated starting from the directory of the requiring module.
Core modules: moduleName not prefixed with / or ./ => the algorithm will first try to search within the core Node.js modules.
Package modules: If no core module is found matching moduleName, then the search continues by looking for a matching module in the first node_modules directory that is found navigating up in the directory structure starting from the requiring module. The algorithm continues to search for a match by looking into the next node_modules directory up in the directory tree, until it reaches the root of the filesystem.
For file and package modules, both files and directories can match moduleName as follows:
- <moduleName>.js
- <moduleName>/index.js
- The directory/file specified in the main property of <moduleName>/package.json

The module cache

Each module is only loaded and evaluated the first time it is required, since any subsequent call of require() will simply return the cached version.

The module cache is exposed via the require.cache variable.

Circular dependencies


// Module a.js
exports.loaded = false;
const b = require("./b");
module.exports = {
  b,
  loaded: true,
};

// Module b.js
exports.loaded = false;
const a = require("./a");
module.exports = {
  a,
  loaded: true,
};

// main.js
const a = require("./a");
const b = require("./b");
console.log("a ->", JSON.stringify(a, null, 2));
console.log("b ->", JSON.stringify(b, null, 2));

# Module definition patterns

Named exports


// module.js
exports.info = (message) => {
  console.log(`info: ${message}`);
};

// app.js
const logger = require("./logger");
logger.info("This is an informational message");

Exporting a function


// module.js
module.exports = (message) => {
  console.log(`info: ${message}`);
};

module.exports.verbose = (message) => {
  console.log(`verbose: ${message}`);
};

// app.js
const logger = require("./logger");
logger("This is an informational message");
logger.verbose("This is a verbose message");

Exporting a class


class Logger {
  constructor(name) {
    this.name = name;
  }
  log(message) {
    console.log(`[${this.name}] ${message}`);
  }
  info(message) {
    this.log(`info: ${message}`);
  }
  verbose(message) {
    this.log(`verbose: ${message}`);
  }
}
module.exports = Logger;

Exporting an instance


class Logger {
  constructor (name) {
    ...
  }
  log (message) {
    ...
  }
}
module.exports = new Logger('DEFAULT');

NOTE: Multiple instances can still be created using the constructor property as follows:


const customLogger = new logger.constructor('CUSTOM');
customLogger.log('This is an informational message');

Monkey patching (Modifying other modules on the global scope)


// patcher.js
require('./logger').customMessage = function () {
  console.log('This is a new functionality');
};

// app.js
require('./patcher');
const logger = require('./logger');
logger.customMessage();

We should avoid using this technique as much as possible.

A real-life use case for this technique would be to mock the http module when writing test cases so that it provides mocked responses instead of a real HTTP request. Example: nock module.

# ESM: EcmaScript Modules

Difference between ESM and CommonJS modules

ESMs are static.

import statements are described at the top level of every module and outside any control flow statement.
The name of the imported ESMs cannot be dynamically generated at runtime using expressions, only constant strings are allowed.

Benefits of static imports

static analysis of the dependency tree
dead code elimination (tree shaking)

Using ESM in Node.js

.mjs file extension
In package.json, add field type with value module

Named exports and imports

In an ESM, everything is private by default and only entities exported with the export keyword are publicly accessible from other modules.


export function log (message) {
  console.log(message);
}

Namespace import


import * as loggerModule from './logger.js';

When we use this type of import statement, the entities are imported into the current scope, so there is a risk of a name clash.


import { log, Logger } from './logger.js';

To resolve this clash,


import { log as log2 } from './logger.js';

With ESM, it is important to specify the file extension for the imported module.

Default exports and imports

To export a single unnamed entity like in CommonJS with module.exports, we can do the following:


// The entity exported is registered under the name `default`
export default class Logger {
  ...
}

// Import
import MyLogger from './logger.js';

❓ What happens if we do import { default } from './logger.js'; ??

Named exports are better than default export

Named exports are explicit and thus, allows IDEs to support the developer with automatic imports, autocomplete, and refactoring tools.
Default exports might make it harder to apply dead code elimination (tree shaking).

Module identifiers

Relative specifiers like ./logger.js or ../logger.js
Absolute specifiers like file:///opt/nodejs/config.js
Bare specifiers - modules in the node_modules folder
Deep import specifiers - refer to a path within a package in node_modules

Async imports (Dynamic imports)

Async imports can be performed at runtime using the special import() operator.


import(dynamicModule)
  .then((res) => ...)

ESM loading process

Create a dependency graph - the dependency graph is needed by the interpreter to figure out how modules depend on each other and in what order the code needs to be executed. The starting point for the dependency resolution is called the entry point.
- Phase 1 - Construction (or parsing)
- Find all the imports and recursively load the content of every module from the respective file. This is done in a depth-first manner. Modules already visited in this process are ignored on revisits.
- Phase 2 - Instantiation
- The interpreter walks the tree view obtained from the previous phase from the bottom to the top. For every exported entity in each module, it keeps a named reference in memory, but doesn't assign any value just yet. After this, the interpreter will do another pass to link the exported names to the modules importing them.
- Phase 3 - Evaluation
- Node.js finally executes the code so that all the previously instantiated entities can get an actual value. The execution order is bottom-up respecting the post-order depth-first visit of our original dependency graph.

Loading process difference between CommonJS and ESM

Read-only live bindings

How is this different from CommonJS?

In ESM, imported modules are tracked as references, so, we can be sure every module has an up-to-date picture of the other modules, even in the presence of circular dependencies.

Monkey patching with ESM


import fs from 'fs';

export function mockEnable (respondWith) {
  mockedResponse = respondWith;
  fs.readFile = () => "Hello, world!";
}

The following statements won't work as they would give read-only live bindings:


import * as fs from 'fs';
import { readFile } from 'fs';

Another approach would be to use syncBuiltinESMExports:


import fs, { readFileSync } from 'fs';
import { syncBuiltinESMExports } from 'module';

fs.readFileSync = () => Buffer.from('Hello, ESM');
syncBuiltinESMExports();

console.log(fs.readFileSync === readFileSync); // true

syncBuiltinESMExports works only for built-in Node.js modules.

# More differences between ESM and CommonJS modules

ESM runs in strict mode, unlike CommonJS, where we need to add "use strict" statements at the beginning of the file.
require, exports, module.exports, __filename, and __dirname are not defined in ESM.

To get the values for the same, we can do the following:


// Create __filename and __dirname
import { fileURLToPath } from 'url';
import { dirname } from 'path';

// import.meta.url is a reference to the current module
// format: file:///path/to/current_module.js
const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);

// Create require
import { createRequire } from 'module';
const require = createRequire(import.meta.url);

this is undefined in the global scope of ESM, whereas in CommonJS this refers to exports.
ESM cannot import JSON files directly as modules. To do this, use the create require approach as mentioned above.

The Node.js Platform

Manoj Kumar Patra — Wed, 02 Oct 2024 08:53:28 +0000

Node.js uses the concept of a module as the fundamental means for structuring the code of a program.

Node.js modules are created to be used, rather than extended. This has the advantage of reducing use cases, simplifying implementation, facilitating maintenance, and increasing usability.

# How Node.js works?

I/O (short for input/output) is definitely the slowest among the fundamental operations of a computer.

Node.js uses non-blocking I/O with the concept of event demultiplexing.

Multiplexing refers to the method by which multiple signals are combined into one so that they can be easily transmitted over a medium with limited capacity. Demultiplexing refers to the opposite operation, whereby the signal is split again into its original components.

Example code for event demultiplexing:


watchedList.add(socketA, FOR_READ)
watchedList.add(fileB, FOR_READ)
while (events = demultiplexer.watch(watchedList)) {
  // event loop
  for (event of events) {
    // This read will never block and will always return data data = event.resource.read()
    if (data === RESOURCE_CLOSED) {
      // the resource was closed, remove it from the watched list
      demultiplexer.unwatch(event.resource)
    } else {
      // some actual data was received, process it
      consumeData(data)
    }
  } 
}

With this approach, the tasks are spread over time, instead of being spread across multiple threads.

Why not multiple threads??

Having a long-running thread for each connection and not using it for most of the time means wasting precious memory and CPU cycles.

Why not polling??

Polling algorithms usually result in a huge amount of wasted CPU time.

# The reactor pattern

Handles I/O by blocking until new events are available from a set of observed resources, and then reacts by dispatching each event to an associated handler.

The asynchronous behavior - The application expresses interest in accessing a resource at one point in time (without blocking) and provides a handler, which will then be invoked at another point in time when the operation completes.

# Libuv

Libuv is the I/O engine of Node.js.

Libuv makes Node.js compatible with all the major operating systems and normalizes the non-blocking behavior of the different types of resource.

Libuv also implements the reactor pattern, thus providing an API for creating event loops, managing the event queue, running asynchronous I/O operations, and queuing other types of task.

The full Node.js platform consists of the following:

A set of bindings responsible for wrapping and exposing libuv and other low-level functionalities to JavaScript.
V8, the JavaScript engine originally developed by Google for the Chrome browser which makes Node.js is so fast and efficient.
A core JavaScript library that implements the high-level Node.js API.

# JavaScript in Node.js vs the browser

Node.js doesn't have a DOM and nor a window or a document.
Node.js has access to a set of services offered by the underlying OS that are not available in the browser.
While the browser provides a higher-level abstraction over the operating system resources, which makes it easier to control and contain the code that runs in it, which will also inevitably limit its capabilities, with Node.js, on the other hand we can virtually have access to all the services exposed by the operating system.
Node.js officially provides support for implementing native modules via the N-API interface.

When developing a library meant to be used by third parties, we have to take into account that our code may run on different versions of Node.js. The general pattern in this case is to target the oldest active long-term support (LTS) release and specify the engines section in our package.json, so that the package manager will warn the user if they are trying to install a package that is not compatible with their version of Node.js.

Using React.memo

Manoj Kumar Patra — Sat, 10 Sep 2022 04:51:12 +0000

# Definition

`React.memo`

A higher order component that can be used to get some performance boost if a component renders the same results given the same props
Checks for prop changes, and skips rendering the component, and reuse the last rendered result in case previous and current props are the same


function PizzaComponent({ name, price }) {
  /* render using props */
  return (
    <div>Pizza: {name}</div>
    <div>Total: ${price}</div>
  )
}

export default React.memo(PizzaComponent);

This component is a candidate for memoization as given a 🍕 image, it always renders the same 🍕 details.

By default, it does a shallow comparison of props
For custom control on props comparison, a custom comparison function can be provided as a second argument.

Let's extend our PizzaComponent to also show different toppings available for a 🍕. Toppings will be an array of strings. As mentioned above, React.memo does a shallow comparison of props, so, in this case, even though the toppings array remains same, it will still re-render.

So, how do we fix this?

We pass a custom comparison function as follows:


function PizzaComponent({ name, price, toppings }) {
  function areEqual(prevProps, nextProps) {
    /*
    return true if passing nextProps to render would return
    the same result as passing prevProps to render,
    otherwise return false
    */
    return (
      prevProps.name === nextProps.name,
      prevProps.price === nextProps.price,
      prevProps.toppings.every(topping => nextProps.toppings.includes(topping))
    )
  }

  /* render using props */
  return (
    <div>Pizza: {name}</div>
    <div>Total: ${price}</div>
    {toppings.map((topping, index) => (
      <div key={`${name}_${topping}`}>{topping}</div>
    ))}
  )
}

export default React.memo(PizzaComponent, areEqual);

# Using React memo

Use React.memo to get a performance boost when:

Component re-renders often
Component is usually provided with the same props during re-rendering
Component contains a relevant amount of elements to reason props equality check

If a component with hooks is wrapped with React.memo, then it will still re-render when state or context changes.

# React memo with function as props

Let's add the functionality to order a 🍕 in our PizzaComponent called onOrder as follows:


function PizzaComponent({ name, price, toppings, onOrder }) {
  function areEqual(prevProps, nextProps) {
    /*
    return true if passing nextProps to render would return
    the same result as passing prevProps to render,
    otherwise return false
    */
    return (
      prevProps.name === nextProps.name,
      prevProps.price === nextProps.price,
      prevProps.toppings.every(topping => nextProps.toppings.includes(topping))
    )
  }

  /* render using props */
  return (
    <div>Pizza: {name}</div>
    <div>Total: ${price}</div>
    {toppings.map((topping, index) => (
      <div key={`${name}_${topping}`}>{topping}</div>
    ))}
    <button onClick={onOrder}>Order</button>
  )
}

export default React.memo(PizzaComponent, areEqual);

and let's use it as follows:


function App ({ store }) {
  const { pizza } = store;
  return (
    <PizzaComponent
      name={pizza.name}
      price={pizza.price}
      toppings={pizza.toppings}
      onOrder={() => placeOrder(pizza.price, pizza.coupon)}
    />
  )
}

Since, React.memo does a shallow comparison, this will do a re-render everytime as it will see the callback function onOrder as a new prop everytime.

To fix this, we can wrap the function passed to onOrder with useCallback as follows:


function App ({ store }) {
  const { pizza } = store;
  /**
   * This will always return the same function instance as long as pizza is the same.
   */
  const onOrder = useCallback(
    () => placeOrder(pizza.price, pizza.coupon),
    [pizza],
  );

  return (
    <PizzaComponent
      name={pizza.name}
      price={pizza.price}
      toppings={pizza.toppings}
      onOrder={onOrder}
    />
  )
}

Intersection Observer in React

Manoj Kumar Patra — Sat, 10 Sep 2022 04:49:43 +0000

# The `useIntersectionObserver` hook

This React Hook can be used to detect visibility of a component on the viewport using the 🔗 IntersectionObserver API natively present in the browser.

Some of the use cases include:

Lazy-loading of images
Infinite scrolling
Start animations

Our hook will take two arguments of the following type:


{
  options?: Omit<IntersectionObserverInit, 'root'>;
  observerCallback: IntersectionObserverCallback;
}

We will use observerCallback to define behaviour on intersection with viewport. It has a type of IntersectionObserverCallback which is defined as an interface as follows:


interface IntersectionObserverCallback {
    (entries: IntersectionObserverEntry[], observer: IntersectionObserver): void;
}

Finally, our hook returns an object with two properties:


{
  observable: (node: Element) => void;
  root: (node: Element) => void;
}

We define three reference variables:

nodeRefs - To hold references for observable nodes as an array of reference objects
observerRef - To hold reference for intersection observer
rootRef - To hold reference for root node with which observables intersect


export function useIntersectionObserver({
  options = { rootMargin: '0px', threshold: [0] },
  observerCallback,
}: {
  options?: Omit<IntersectionObserverInit, 'root'>;
  observerCallback: IntersectionObserverCallback;
}) {
  const nodeRefs = useRef<Element[]>([]);
  const rootRef = useRef<Element>(null);
  const observerRef = useRef<IntersectionObserver | null>(null);
  ...
};

Then, we define rootRefCallback and observableRefCallback mapped to root and observable respectively as follows:

‼️ These are used to set the root and observable nodes for intersection observer.


const observableRefCallback = useCallback<(node: Element) => void>(
  (node) => {
    nodeRefs.current.push(node);
    initializeObserver();
  },
  [initializeObserver]
);

const rootRefCallback = useCallback<(node: Element) => void>(
  (rootNode) => {
    rootRef.current = rootNode;
    initializeObserver();
  },
  [initializeObserver]
);

In both cases, once the root and observable nodes are set, we initialize the intersection observer with function initializeObserver. This function basically resets the intersection observer as follows:


const unobserve = useCallback(() => {
  // stops watching all of its target elements for visibility changes.
  observerRef.current?.disconnect();
  // sets to null
  observerRef.current = null;
}, []);

const observe = useCallback(() => {
  // If observable nodes exist
  const nodes = nodeRefs.current;
  if (nodes.length > 0) {
    const root = rootRef.current;
    const { rootMargin, threshold } = options;
    // Create a new IntersectionObserver instance
    const observer = new IntersectionObserver(observerCallback, {
      root,
      rootMargin,
      threshold,
    });
    // Observe all observable nodes
    nodes.forEach((node) => observer.observe(node));
    // Set observer reference
    observerRef.current = observer;
  }
}, [options.rootMargin, options.threshold]);


const initializeObserver = useCallback(() => {
  unobserve();
  observe();
}, [observe, unobserve]);

# Using the intersection observer hook

Define observer callback


const observerCallback = (entries: IntersectionObserverEntry[], observer: IntersectionObserver) => {
  entries.forEach((entry) => {
    ...
  });
};

Call `useIntersectionObserver` to get observable and root callbacks


const { observable, root } = useIntersectionObserver({
  observerCallback,
});

Attach observable to observable nodes


React.useEffect(() => {
  document.querySelectorAll('.anim').forEach(observable);
}, []);

Attach root to root node


<div ref={root}>
  <div className="anim">...</div>
  <div className="anim">...</div>
  <div className="anim">...</div>
</div>

# References

🔗 Online Playground

🤩 Scroll down on the browser in the playground to see the animation.

Design Patterns for Resilient Serving - Continued Model Evaluation

Manoj Kumar Patra — Sun, 13 Feb 2022 09:41:18 +0000

This design pattern helps detect and take action when a deployed model is no longer fit-for-purpose.

Reasons for model degradation

Concept drift
Data drift

Concept drift	Data drift
Relationship between the model inputs and target have changed	Any change that has occurred to the data being fed to the model for predictions as compared to the data that was used for training.

Identifying model deterioration

Continuous monitoring of the model's predictive performance over time
Assess this performance with the same evaluation metrics used during development

Continuous model evaluation provides a framework to evaluate a deployed mode's performance exclusively on new data. This means we can detect a model's staleness as early as possible. This information helps to do either of the following:

Retrain the model
Replace the existing model with a new version entirely.

This is done by capturing the following:

ground truth
prediction inputs and outputs for comparing with ground truth values
model versions
timestamp of prediction requests

Triggers for retraining

Whether to retrain or not based on an evaluation report depends on what amount of deterioration of performance is acceptable in relation to the cost of retraining.

Setting a higher threshold for model performance ensures a higher-quality model in production but will require more frequent retraining jobs which is costly.

Having a lower threshold on the other hand, would mean more cost effective but there is a chance of having a stale model in production.

Scheduled retraining

Continuous evaluation	Scheduled retraining
May happen everyday	May occur only every week or every month

Design Patterns for Resilient Serving - Batch Serving

Manoj Kumar Patra — Sun, 06 Feb 2022 05:03:01 +0000

Batch serving is useful when we need to carry out predictions asynchronously unlike stateless serving function where we can process one instance or at most a few 1000 instances embedded in a single request.

Examples include:

Determine whether to reorder a stock-keeping unit - needs to be carried out on an hourly basis
Creating personalized songs playlist
Recommendation engines with periodic refresh rates - say, the periodic refresh rate is per hour, then, we carry out inferences for only those users who visited the website in last on hour

To achieve asynchronous predictions, batch serving makes use of distributed data processing infrastructures such as BigQuery, Apache Beam, etc.

Consider this example below, where we run inference on approx. 1.5 million rows of data using BigQuery:

WITH all_complaints AS (
SELECT * FROM ML.PREDICT(MODEL external_model,
  (SELECT consumer_complaint_narrative AS reviews
   FROM `bigquery-public-data`.cfpb_complaints.complaint_database
   WHERE consumer_complaint_narrative IS NOT NULL
  )
))
SELECT * FROM all_complaints
ORDER BY positive_review_probability DESC LIMIT 5

Here, the following operations take place in order:

Read consumer_complaint_narrative column from dataset where consumer_complaint_narrative is not NULL. Let's assume this is a total of X values. These are then distributed across N shards.
N workers process each of N shards to read the data and do the inference using the model files.
Each of the N workers find the 5 most positive complaints from the shard they processed.
Take the (5 * N) complaints, sort them and then select 5 from the actual result.

Design Patterns for Resilient Serving - Stateless Serving Function

Manoj Kumar Patra — Sat, 15 Jan 2022 09:42:55 +0000

Using this design pattern, we can have a production ML system synchronously handle millions of prediction requests per second.

Stateless components	Stateful components
Output is determined purely by the inputs	Output depends both on inputs and internal state
No state => can be shared by multiple clients	Need to store each client's conversational state
Highly scalable => initialized on first request and destroyed when client terminates or timed out	Expensive and difficult to manage

Exporting a model as a stateless function means the stateful variables such as epoch number, learning rate, etc. need to be tracked separately and not to be included in the exported file.

Demerits of carrying out inferences on an in-memory object

Loading the entire model (which can be large in size) into memory
Limits on latency on predictions
Programming language dependency
The model input and output may not be user-friendly

Achieving statelessness

Export the model into a format that is programming language independent
Restore the model as a stateless function in production
Make the stateless function available via REST endpoint

To save a model in Keras: model.save(export_path)

This will export the model as a <model>.pb file - protocol buffer and extracts out other stateful variables into separate files.

model.save also takes an optional argument signatures. This can be used to define a dictionary stating different serving functions. If not specified, the model's forward pass is exported.

To determine the signature of the stateless function that we will use for serving:

!saved_model_cli show --dir {export_path} --tag_set serve --signature_def serving_default

Finally, we can use this as follows:

restored = tf.keras.models.load_model(export_path)
infer = restored.signatures['serving_default']
outputs = infer(input)

DEV Community: Manoj Kumar Patra

Creational Design Patterns

# Factory

EXAMPLE 1

EXAMPLE 2

# Builder

EXAMPLE 1

EXAMPLE 2

# Revealing constructor

Use cases

Syntax

Immutable objects

Immutable version of the Node.js Buffer component

# Singleton

# Wiring Modules using dependency injection

Disadvantage of DI

Alternative - Inversion of Control

Coding with Streams

# Buffer vs stream

Spatial efficiency

Time efficiency

# Stream basics

# Readable streams

Reading from a stream

Non-flowing mode

Flowing mode

Async iterators

Readable stream implementation

Simplified readable

# Writable streams

Backpressure

Writable stream implementation

Simplified writable

# Duplex streams

# Transform streams

Transform stream implementation

# Passthrough Streams

Observability

Late piping

# Lazy streams

Custom implementation

# Pipes and error handling

Error handling with pipeline

# Asynchronous control flow patterns with streams

Sequential execution

Unordered parallel execution

Unordered limited parallel execution

Ordered parallel execution

# Piping patterns

Combining streams

Forking streams

Merging streams

Multiplexing and demultiplexing

Multiplexing

Demultiplexing

Asynchronous Control Flow Patterns

# Sequential execution

Executing a known set of tasks in sequence

Sequential iteration

Pattern with callback

Pattern with Promise

Promisify a callback based API

# Parallel execution

Pattern with callback

Pattern with Promise

The Unlimited Parallel Execution pattern

Fixing race conditions

# Limited parallel execution

Limiting concurrency

Limited concurrency with queues

Example of spider web app

Sequential execution and iteration

Parallel execution

Refresher on Promises

Refresher on async/await

Limited parallel execution with async/await

# Tackling a problem of memory leak associated with Promises

Callbacks and Events

# The callback pattern

Continuous-passing style

Error handling with `pipeline`

`process.nextTick() vs setImmediate()`

Extending `EventEmitter` class

`require` implementation

`module.exports` vs `exports`

`require` implementation

`module.exports` vs `exports`