Edit: The REPL for this article is up at repl.it, and this commit contains more specifics.

Simplifying chains of async callbacks with Rx-Helper

Enough Callbacks Already!

A colleague of mine, who was getting started with some really cool JavaScript stuff in an AWS Lambda function found how gross chains of callbacks can become. But, not to be stopped, he continued putting stuff together, using the async npm library, custom synchronous functions that he was forced to do in callback style to satisfy async, and by the time I suggested Promises and utils/promisify there were 3 types of async code in one short lambda function!

Then console.log debugging came into play, and the typical smattering of console.log throughout occurred (you do remember to remove those, or have a linting rule reminding you to do so, right?).

The task

At a high-level, and good enough for us to understand, the basic essence of the issue was to read lines from a log file, look for interesting ones, and log those interesting ones to a new file.

We want to do this in a way that allows for easy expression of sync or async parts of the chain, while using the same async-handling style throughout, rather than part callbacks, part Promises, part await, part generators, etc.. We'll use the Rx-Helper to create a clean model of our app's events and consequences.

Rx Helper and a Model of Your App

Rx Helper guides you toward modeling your application into named events, connected by a field called type, triggered upon matches of that field, and returning RxJS Observables. Our Event Model could be:

• load - we must load a file of some name, to split it into lines
• logLines - we will be provided our lines in an event of type logLines
• goodLines - the interesting lines will come in an event of type goodLines

Also, we want our app to do stuff, so in addition to our Event Model, we have a Consequence Model, explaining what "side-effects" are triggered upon our event.

• load - When we see load, we should open the filename in our payload, and read it through, to create a logLines event.
• logLines - No side-effects. A sync handler will return an event of type goodLines containing the "interesting" lines in its payload.
• goodLines - When we see goodLines, we should write a file (name ending in .out)

Rx-Helper and CrossCutting Concerns

Remember we often need to add console.log statements throughout the code?

Once we've broken our app down into events, each event processed through the agent, we can add logging for every event in the single line:

agent.spy(({ event }) => console.log(JSON.stringify(event)))

And conveniently, we can turn all the logging off in one place too, or make it dependent on an environment variable in a single place too. Cool!

Rx-Helper Fixes Callback Hell

The code shows two ways to eliminate callbacks, both of which come from the RxJS Observable library itself. The first I'll explain, the handler for goodLines, explicitly wraps an async callback in an Observable, and is comparable to returning a Promise. The second is to use the bindNodeCallback function from RxJS, which is comparable to using promisify to automatically wrap a function like fs.writeFile that follows Node Callback conventions.

The long way:

  agent.on("goodLines", ({ event }) => {
    const lines = event.payload

    // This is effectively what bindNodeCallback did for us for the case of fs.readFile
    return new Observable(notify => {
      fs.writeFile(FILE_PATH + ".out", lines.join("\n").toUpperCase(), "UTF8", err => {
        // tell the agent our outcome
        err ? notify.error(err) : notify.complete()
      })
    })
  })
}

The Arguments for agent.on are:

• A condition for when we run (string, regex, or boolean-returning function)
• A handler function returning an Observable-wrapped consequence
• A config argument indicating what event type our Observable's output gets wrapped in

For the first argument we simply trigger on an event of type goodLines. In our case, we're not emitting anything after writing the goodLines, so we omit the final argument.

In the middle argument, the handler, we see a call to fs.writeFile, in whose callback we call either the notify.error or notify.complete method. This takes place in the call to new Observable, which wraps our behavior in the type Rx-Helper likes to receive- that of an Observable. We could return a Promise, in this simple example, and it would work the same, but it's a better practice to simply use Observables which notify of progress, success, and error conditions in a standard way, and are cancelable. Observables compose better than Promises.

However, we don't need to write that bridge to Observables from callbacks ourselves. The code grows even cleaner if we use bindNodeCallback - a higher-order function that takes creates an Observable-returning function from a callback-accepting function. For the handler of type load:

const fs = require("fs")
const notifyOfFileContents = bindNodeCallback(fs.readFile)

agent.on(
  "load",
  ({ event: { payload } }) => notifyOfFileContents(payload, "UTF8"),
  { type: "logLines" })

We run upon an event of type load, and we return the Observable returned by notifyOfFileContents in an event of type logLines. That is so cool. No Promises, no async/await, just a way showing how to produce the event we want triggered after our event of type load.

Finally

What Rx Helper helps you do is use two great concepts: events, and Observables, to orchestrate chains of functionality, linked by event names of your choosing. It allows you to easily add in cross-cutting concerns like persistence, authentication, validation, and as we saw, logging through the use of spy/filter. And it can keep your code from proliferating different ways to handle async. This can keep code reviews shorter, and help with the overall intelligibility of your codebase. Code is maintained far more than it is written, so a unified way of handling async, rather than a combination of all possible techniques will be in the interest of any team wanting to move quickly and accurately.