The Observer pattern

Another important and fundamental pattern used in Node.js is the Observer pattern. Together with the Reactor pattern and callbacks, the Observer pattern is an absolute requirement for mastering the asynchronous world of Node.js.

The Observer pattern is the ideal solution for modeling the reactive nature of Node.js and a perfect complement for callbacks. Let's give a formal definition, as follows:

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.

The EventEmitter

In traditional object-oriented programming, the Observer pattern requires interfaces, concrete classes, and a hierarchy. In Node.js, all this becomes much simpler. The Observer pattern is already built into the core and is available through the EventEmitter class. The EventEmitter class allows us to register one or more functions as listeners, which will be invoked when a particular event type is fired. Figure 3.2 visually explains this concept:

Figure 3.2: Listeners receiving events from an EventEmitter

The EventEmitter is exported from the events core module. The following code shows how we can obtain a reference to it:

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

The essential methods of the EventEmitter are as follows:

• on(event, listener): This method allows us to register a new listener (a function) for the given event type (a string).
• once(event, listener): This method registers a new listener, which is then removed after the event is emitted for the first time.
• emit(event, [arg1], [...]): This method produces a new event and provides additional arguments to be passed to the listeners.
• removeListener(event, listener): This method removes a listener for the specified event type.

All the preceding methods will return the EventEmitter instance to allow chaining. The listener function has the signature function([arg1], [...]), so it simply accepts the arguments provided at the moment the event is emitted.

You can already see that there is a big difference between a listener and a traditional Node.js callback. In fact, the first argument is not an error, but it can be any data passed to emit() at the moment of its invocation.

Creating and using the EventEmitter

Let's now see how we can use an EventEmitter in practice. The simplest way is to create a new instance and use it immediately. The following code shows us a function that uses an EventEmitter to notify its subscribers in real time when a particular regular expression is matched in a list of files:

import { EventEmitter } from 'events'
import { readFile } from 'fs'
function findRegex (files, regex) {
  const emitter = new EventEmitter()
  for (const file of files) {
    readFile(file, 'utf8', (err, content) => {
      if (err) {
        return emitter.emit('error', err)
      }
      emitter.emit('fileread', file)
      const match = content.match(regex)
      if (match) {
        match.forEach(elem => emitter.emit('found', file, elem))
      }
    })
  }
  return emitter
}

The function we just defined returns an EventEmitter instance that will produce three events:

• fileread, when a file is being read
• found, when a match has been found
• error, when an error occurs during reading the file

Let's now see how our findRegex() function can be used:

findRegex(
  ['fileA.txt', 'fileB.json'],
  /hello \w+/g
)
  .on('fileread', file => console.log(`${file} was read`))
  .on('found', (file, match) => console.log(`Matched "${match}" in ${file}`))
  .on('error', err => console.error(`Error emitted ${err.message}`))

In the code we just defined, we register a listener for each of the three event types produced by the EventEmitter that was created by our findRegex() function.

Propagating errors

As with callbacks, the EventEmitter can't just throw an exception when an error condition occurs. Instead, the convention is to emit a special event, called error, and pass an Error object as an argument. That's exactly what we were doing in the findRegex() function that we defined earlier.

The EventEmitter treats the error event in a special way. It will automatically throw an exception and exit from the application if such an event is emitted and no associated listener is found. For this reason, it is recommended to always register a listener for the error event.

Making any object observable

In the Node.js world, the EventEmitter is rarely used on its own, as you saw in the previous example. Instead, it is more common to see it extended by other classes. In practice, this enables any class to inherit the capabilities of the EventEmitter, hence becoming an observable object.

To demonstrate this pattern, let's try to implement the functionality of the findRegex() function in a class, as follows:

import { EventEmitter } from 'events'
import { readFile } from 'fs'
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
  }
}

The FindRegex class that we just defined extends EventEmitter to become a fully fledged observable class. Always remember to use super() in the constructor to initialize the EventEmitter internals.

The following is an example of how to use the FindRegex class we just defined:

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

You will now notice how the FindRegex object also provides the on() method, which is inherited from the EventEmitter. This is a pretty common pattern in the Node.js ecosystem. For example, the Server object of the core HTTP module inherits from the EventEmitter function, thus allowing it to produce events such as request (when a new request is received), connection (when a new connection is established), or closed (when the server socket is closed).

Other notable examples of objects extending the EventEmitter are Node.js streams. We will analyze streams in more detail in Chapter 6, Coding with Streams.

EventEmitter 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. This allows us to release the memory used by the objects in a listener's scope and prevent memory leaks. Unreleased EventEmitter listeners are the main source of memory leaks in Node.js (and JavaScript in general).

A memory leak is a software defect whereby memory that is no longer needed is not released, causing the memory usage of an application to grow indefinitely. For example, consider the following code:

const thisTakesMemory = 'A big string....'
const listener = () => {
  console.log(thisTakesMemory)
}
emitter.on('an_event', listener)

The variable thisTakesMemory is referenced in the listener and therefore its memory is retained until the listener is released from emitter, or until the emitter itself is garbage collected, which can only happen when there are no more active references to it, making it unreachable.

You can find a good explanation about garbage collection in JavaScript and the concept of reachability at nodejsdp.link/garbage-collection.

This means that if an EventEmitter remains reachable for the entire duration of the application, all its listeners do too, and with them all the memory they reference. If, for example, we register a listener to a "permanent" EventEmitter at every incoming HTTP request and never release it, then we are causing a memory leak. The memory used by the application will grow indefinitely, sometimes slowly, sometimes faster, but eventually it will crash the application. To prevent such a situation, we can release the listener with the removeListener() method of the EventEmitter:

emitter.removeListener('an_event', listener)

An EventEmitter has a very simple built-in mechanism for warning the developer about possible memory leaks. 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.

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.

Synchronous and asynchronous events

As with callbacks, events can also be emitted synchronously or asynchronously with respect to the moment the tasks that produce them are triggered. It is crucial that we never mix the two approaches in the same EventEmitter, but even more importantly, we should never emit the same event type using a mix of synchronous and asynchronous code, to avoid producing the same problems described in the Unleashing Zalgo section. The main difference between emitting synchronous and asynchronous events lies in the way listeners can be registered.

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.

The FindRegex() class we defined previously emits its events asynchronously after the find() method is invoked. This is why we can register the listeners after the find() method is invoked, without losing any events, as shown in the following code:

findRegexInstance
  .addFile(...)
  .find()
  .on('found', ...)

On the other hand, if we emit our events synchronously after the task is launched, we have to register all the listeners before we launch the task, or we will miss all the events. To see how this works, let's modify the FindRegex class we defined previously and make the find() method synchronous:

find () {
  for (const file of this.files) {
    let content
    try {
      content = readFileSync(file, 'utf8')
    } catch (err) {
      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
}

Now, let's try to register a listener before we launch the find() task, and then a second listener after that to see what happens:

const findRegexSyncInstance = new FindRegexSync(/hello \w+/)
findRegexSyncInstance
  .addFile('fileA.txt')
  .addFile('fileB.json')
  // this listener is invoked
  .on('found', (file, match) => console.log(`[Before] Matched "${match}"`))
  .find()
  // this listener is never invoked
  .on('found', (file, match) => console.log(`[After] Matched "${match}"`))

As expected, the listener that was registered after the invocation of the find() task is never called; in fact, the preceding code will print:

[Before] Matched "hello world"
[Before] Matched "hello NodeJS"

There are some (rare) situations in which emitting an event in a synchronous fashion makes sense, but the very nature of the EventEmitter lies in its ability to deal with asynchronous events. Most of the time, emitting events synchronously is a telltale sign that we either don't need the EventEmitter at all or that, somewhere else, the same observable is emitting another event asynchronously, potentially causing a Zalgo type of situation.

The emission of synchronous events can be deferred with process.nextTick() to guarantee that they are emitted asynchronously.

EventEmitter versus callbacks

A common dilemma when defining an asynchronous API is deciding whether to use an EventEmitter or simply accept a callback. 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.

But besides this simple principle, a lot of confusion is generated from the fact that the two paradigms are, most of the time, equivalent and allow us to achieve the same results. Consider the following code as an example:

import { EventEmitter } from 'events'
function helloEvents () {
  const eventEmitter = new EventEmitter()
  setTimeout(() => eventEmitter.emit('complete', 'hello world'), 100)
  return eventEmitter
}
function helloCallback (cb) {
  setTimeout(() => cb(null, 'hello world'), 100)
}
helloEvents().on('complete', message => console.log(message))
helloCallback((err, message) => console.log(message))

The two functions helloEvents() and helloCallback() can be considered equivalent in terms of functionality. The first communicates the completion of the timeout using an event, while the second uses a callback. But what really differentiates them is the readability, the semantics, and the amount of code that is required for them to be implemented or used.

While a deterministic set of rules for you to choose between one style or the other can't be given, here are some hints to help you make a decision on which method to use:

• Callbacks have some limitations when it comes to supporting different types of events. In fact, we can still differentiate between multiple events by passing the type as an argument of the callback, or by accepting several callbacks, one for each supported event. However, this can't exactly be considered an elegant API. In this situation, the EventEmitter can give a better interface and leaner code.
• The EventEmitter should be used when the same event can occur multiple times, or may not occur at all. A callback, in fact, is expected to be invoked exactly once, whether the operation is successful or not. Having a possibly repeating circumstance should make us think again about the semantic nature of the occurrence, which is more similar to an event that has to be communicated, rather than a result to be returned.
• An API that uses callbacks can notify only one particular callback, while using an EventEmitter allows us to register multiple listeners for the same event.

Combining callbacks and events

There are some particular circumstances where the EventEmitter can be used in conjunction with a callback. This pattern is extremely powerful as it allows us to pass a result asynchronously using a traditional callback, and at the same time return an EventEmitter, which can be used to provide a more detailed account on the status of an asynchronous process.

One example of this pattern is offered by the glob package (nodejsdp.link/npm-glob), a library that performs glob-style file searches. The main entry point of the module is the function it exports, which has the following signature:

const eventEmitter = glob(pattern, [options], callback)

The function takes a pattern as the first argument, a set of options, and a callback that is invoked with the list of all the files matching the provided pattern. At the same time, the function returns an EventEmitter, which provides a more fine-grained report about the state of the search process. For example, it is possible to be notified in real time when a match occurs by listening to the match event, to obtain the list of all the matched files with the end event, or to know whether the process was manually aborted by listening to the abort event. The following code shows what this looks like in practice:

import glob from 'glob'
glob('data/*.txt',
  (err, files) => {
    if (err) {
      return console.error(err)
    }
    console.log(`All files found: ${JSON.stringify(files)}`)
  })
  .on('match', match => console.log(`Match found: ${match}`))

Combining an EventEmitter with traditional callbacks is an elegant way to offer two different approaches to the same API. One approach is usually meant to be simpler and more immediate to use, while the other is targeted at more advanced scenarios.

The EventEmitter can also be combined with other asynchronous mechanisms such as promises (which we will look at in Chapter 5, Asynchronous Control Flow Patterns with Promises and Async/Await). In this case, just return an object (or array) containing both the promise and the EventEmitter. This object can then be destructured by the caller, like this: {promise, events} = foo().