Explicitly defining side-effects outside of .subscribe

[Sawtaytoes]

I wrote this article about "explicit dispatch" in Redux-Observable a few years ago:
https://itnext.io/the-best-practice-anti-pattern-5e8bd873aadf

I want to discuss this approach, and why it does or doesn't make sense in vanilla RxJS.

RxJS's Fall-Through Pattern

RxJS design patterns require fall-through; by that I mean, all values should fall through to the next stage until they ultimately reach a .subscribe function where the subscriber performs some unit of work on the value such as console logging, calling .next on a subject, or doing nothing with the provided value.

Side-Effects

With Redux-Observable, you can't do side-effects in .subscribe because you don't control subscriptions and only have access to .pipe. In those cases, it makes sense to do side-effects, like pushing analytics, bug-tracking, chat clients scripts, etc in a tap or catchError.

This tells me, side-effects go in tap or catchError. I would assume this is the same for vanilla RxJS, but unit testing is near-impossible without wrapper functions passing a dependencies. In my opinion, that puts RxJS into a similar camp as Redux-Observable.

Unit Testing RxJS

When I write vanilla RxJS, I follow a same pattern to Redux-Observable epics by creating a wrapper function with dependencies and an input$ observable. This allows for easy testing and decouples these pipelines from the .subscribe function.

It looks similar to this example when called:

processData({
  fromAxios,
  input$: jsonData$,
})

In the past, I'd assemble observables that flow into each other using a pipeline of mergeMap functions. I tried something different recently with an explicit outputValue callback to resemble Redux-Observable epics:

processData({
  fromAxios,
  input$: jsonData$,
  outputValue: () => {
    writeFile$
    .next(value)
  },
})

I'm actually unsure what's good or bad about this approach as I've only used it recently.

Implicit Dispatch

In the case of Redux-Observable, the dispatch function is called whenever a value emits from the pipeline. Consumers have to aware when they're emitting values and ensure they're valid Redux actions.

If your consumers aren't going to deal with a .subscribe function, it makes more sense to give them access to dispatch directly using dependency-injection. In many cases, you explicitly want to call dispatch at places in your pipeline to perform a side-effect the same as you could any other 3rd party service.

Use Case

A good example is error handling. Any time an observable errors at a particular catchError, I might want to tell my error reporting platform. I could either pass the library's client directly or provide a reportError output function. Either way, I'm still peforming a side-effect in catchError. Why have one side-effect handled with a callback and another handled implicitly by .subscribe?

Flawed?

My argument is flawed in some ways because fromEvent doesn't take a callback nor do other observables. The entire reason for RxJS is that observables emit values. But that only matters when one observable is supposed to flow into another. In my opinion, that doesn't make sense when you want to perform side-effects like dispatching to Redux.

Your Thoughts

I don't claim to know everything about RxJS or how to unit test it, and the majority of my RxJS experience over the last 5 years is with Redux-Observable. In that particular environment, "splitting the pipeline" is by-far one of the most-problematic things for people understand because it's so unintuitive. I'm curious how that same notion translates in the vanilla RxJS world.

[voliva]

I think the general rule of thumb on where to put side effects in "vanilla" RxJS is subscribe, while keeping the chain as side-effect free as posible. However, there are cases where tap makes more sense. The way I see it, you might find one more suitable than the other depending on the use case.

RxJS design patterns require fall-through; by that I mean, all values should fall through to the next stage until they ultimately reach a .subscribe function where the subscriber performs some unit of work on the value such as console logging, calling .next on a subject, or doing nothing with the provided value.

That's not always the case: It's not unusual to have some observables nesting, e.g. when using concatMap or switchMap. I feel this vision of fall through is too simplified, because streams are lazy, composable and can be multicast. Take this example:

const timeSinceStart$ = interval(1000).pipe(
  map(t => `seconds ellapsed: ${t}`),
  share()
)
timeSinceStart$.subscribe(console.log);

const lapButtonPress$ = fromEvent(lapButton, 'click');
const laps$ = lapButtonPress$.pipe(
  withLatestFrom(timeSinceStart$),
  map(([_, time]) => time),
  scan((laps, newLap) => [ ...laps, newLap ], [])
)

laps$.subscribe(console.log);

At what point would you say the values from timeSinceStart$ reach the .subscribe? In a way, this stream is getting forked into two, one where it reaches subscribe immediately, and another which gets consumed by laps$ instead.

Also note that pipe is not magic: every operator is creating a new observable which internally calls .subscribe to the previous observable.

And lastly we have operators that completely break the "fall-through" concept: things like retry, repeat, expand, or ignoreElements to some extend.

This tells me, side-effects go in tap or catchError. I would assume this is the same for vanilla RxJS, but unit testing is near-impossible without wrapper functions passing a dependencies. In my opinion, that puts RxJS into a similar camp as Redux-Observable.

It's way easier to test observables that don't have any side effect in their chain you need to assert or mock. So I think the general principle is try to minimise them.

But if you do, you can always use jest mocks to mock specific imports of the file you're testing, or using any other strategy. I think the one you use is also valid (the one inspired by redux-observable)

However, following the example you provided:

processData({
  fromAxios,
  input$: jsonData$,
  outputValue: value => {
    writeFile$
    .next(value)
  },
})

In RxJS the way I see more idiomatic is the other way around. Instead of making processData() to invoke a side effect outputValue, make processData return a stream, that you can later compose to make the side effect you want. Something like:

const dataProcessed$ = processData({
  fromAxios,
  input$: jsonData$
});

// when needed
dataProcessed$.subscribe(value => writeFile$.next(value));

This way processData doesn't do any side effect at all. It would even posible to make writeFile$ a stream composed from others (one of them, dataProcessed$) and completely remove any side effect on this bit of code (and move the side effect that reads from writeFile$ to write a file where it's needed).

[Sawtaytoes]

Thank you so much for responding to my post! I feel bad leaving you hangin' for so long; although, I didn't notice this message until recently.

Side-effects in subscribe aren't ideal

You are correct in your assessment.

It is far better to do the side-effect in subscribe than somewhere else; although, it was too difficult for other devs to follow the code; including me.

This has always been an issue I've run into when using RxJS without Redux-Observable which is why I almost always use Redux-Observable (even without Redux).

.subscribe isn't a good wiring tool

Instead of assembling a bunch of .subscribe calls, which is what I ended up with after allowing fall-through, I came up with another idea.

I only came up with this idea after reading all of RxJS's documentation from start to end including every observable and operator I didn't already know.

Separate states are important

The reason I made these separate Redux-Observable-like functions is because:

• It's easier to unit test a segment.
• It's easier to document and have other developers understand what I'm doing.
• Each function contains state (from scan).

Wrapping them in pipe would be easier to read, but doesn't allow me to separately subscribe to each one.

Multicasting with publish

Now that I understand (or did last month) what multicasting is, I came up with a good solution:

somethingWithData$
.pipe(
  publish((
    input$,
  ) => (
    processData({
      fromAxios,
      input$,
    })
  )),
  publish((
    input$,
  ) => (
    doMoreDataProcessing({
      fsWriteFile,
      input$,
    })
  )),
  // ... etc. I have 4 of these pipeline chains now.
)

What do you think of this solution?

• State is maintained in each function's scan operator.
• I can still pass input$ and other mocked observable creators for unit testing.
• The output still comes in .subscribe which honestly reads nicer.