Case Study - Days Without Accident

Consider the following stream of events, which I would guess would be intelligible to the average developer:

  dailyReport: date: 2019-01-01, daysWithoutAccident: 270
  dailyReport: date: 2019-01-02, daysWithoutAccident: 271
  accident: date: 2019-01-03, severity: 7
  dailyReport: date: 2019-01-03, daysWithoutAccident: 0
  dailyReport: date: 2019-01-04, daysWithoutAccident: 1
  ...

You could reasonably infer that it is the combined output of two processes (aka streams):

• A stream of daily Reports, which continue to count days without accidents
• A stream of accidents, which may or may not occur on a given day.

Our ability to infer this is aided by each event being labeled with its type: dailyReport|accident, and then some fields that are relevant to that type. This is basically the Flux Standard Action pattern, which groups fields into a payload:

{ type: 'dailyReport', payload: { date: "2019-01-04", daysWithoutAccident: 1 } }

Now - How would you go about building such an app that creates output like the above? Take some time to sketch out your own solution if you want, before reading about the Rx-Helper way.

The Agent

When you use Rx-Helper, you get a helping agent, It has a concept of an Action Stream - a work queue essentially - just like the V8 JavaScript runtime. You tell it to do work for you in one of a couple of ways. The first, is to tell it to process a Flux Standard Action. Let's process an FSA of type start, which we can use to trigger other behavior.

import { agent } from 'rx-helper'

agent.process({ type: 'start' })

Actions given to process become part of the Action Stream, as well as consequential actions, as we'll see shortly.

Events Over Time

The first stream is that of an accident happening, midday each day, with probability 1 in 100.

We'll use an event-handler-like syntax to deinfe a process that — upon startup — randomly causes an accident event to be added to the Action Stream. We're using the rxjs-compat syntax for clarity, combined with the on method from the rx-helper library, which together produce the cleanest-possible RxJS code.

agent.on('start', () => {
    return interval(1000*60*60*24)           // Every day
       .delay(1000*60*60*12).                // at midday
       .filter(() => Math.random() < 0.01)   // if we're unlucky
       .map(() => ({                         // create the accident info
           date: new Date(),
           severity: Math.random().toString(7)[2]
       }))
}, {type: 'accident'})           // And add it to the Action Stream as type: 'accident'

There's no setInterval, async/await or even local variables to track state - magic! Let's dig in..

The first argument to on is the trigger for when the function will be run - an action of type start. We already added an event to the Action Stream that will ensure this function is run once. Once run, the Observable of accidents-over-time will be running, causing accident FSA's to be processed on random days - We have our Chaos Monkey!

Concurrently...

We also need an Observable of daily reports to be part of the Action Stream. We can use scan, an RxJS reducer-over-time, to produce an object that would be the payload of a dailyReport event.

agent.on('start', () => {
    return interval(1000*60*60*24)  // Every day at the beginning of the day
       .scan(({count}) => ({        // Create a {date, count} Object
           date: new Date(),        //   with today's date
           count: count + 1         //   that increments
       }), {count: 0})              //   starting at zero
}, {
    type: 'dailyReport'     // Add it to the Action Stream as type: 'dailyReport'
}

Beautiful! But any time an accident occurs, we need to stop the counting and start again. In other words we need to cut off the previous count.

Using Rx-Helper, the above needs only two modifications:

• Start counting on either a start or an accident
• Terminate (and restart) counting upon an accident

If you know the switchMap operator in RxJS, that is the behavior of such a counter, but let's see how Rx-helper lets us do it:

agent.on(/start|accident/, () => {        // Upon an accident *or* the start of the app
    return interval(1000*60*60*24)
       .scan(({count}) => ({
           date: new Date(),
           count: count + 1
       }), {count: 0})
}, {
    type: 'dailyReport',
    concurrency: 'cutoff'           // Replace the last counter, starting a new one
}

Now that is some clean code! If we want to see every event in a log, that's what a filter is for:

agent.filter(() => true, ({ action }) => console.log(action))

And now you can plainly see all of our accident-reporting logic, and you're ready to bind it to a view layer.

The Most Frequently Asked Question

Where are all the Moving Parts? The beauty of this declarative, Observable style is that there are no moving parts! 🤯

The business rules you are given translate very cleanly into declarative logic. A rule such as:

On startup, produce an accident every X seconds, with probability Y

A rule such as that is implemented by establishing a mapping from the cause (an action of type 'start'), to the effect: an object that represents "an accident every X seconds, with probability Y". Our 'event handler'—the second argument to on— returns an Observable. And since the agent subscribes (and possibly unsubscribes) from these returned Observables according to the concurrency mode, the work of opening the subscription and managing it is hidden from you. You simply say what should be and the agent will 'make it so'. That's a helpful agent!

Takeaway: Observables are Powerful Encapsulation

The 'aha' insight is that you can even have a variable representing "an accident every X seconds, with probability Y". Before I worked with Observables, such as in my days with Ruby, I knew of no such data type, so it was unthinkable that the return value of a function could be "an accident every X seconds, with probability Y". In the real world, we have no problem conceiving of streams of items over time. For example, if I successfully apply for a job, the return value I expect is a Salary - a paycheck every X days, not a Promise for a single paycheck! It is strange however that one can program so long before encountering a data type like Observable that allows for the first-class representation of values-over-time.

Understanding Observables is the hard part. The Rx-Helper agent just helps you tie them together using event names of your own choosing, which makes for scalable, readable applications. Happy Coding!

Live version

Check out a working version of this on Runkit, and leave your comments, share and lets create a discussion!