How to design declarative streams with RxJS?

[jlemieux]

My Problem

I want to learn more about writing declarative/functional code with RxJS. I'm trying to use splitting, merging, and other forms of stream coordination to encapsulate the logic, rather than imperatively handling things. I have an example scenario that hopefully you can help me with.

Context

When a user clicks, I want to log statistics about their click & hover behavior. I also want to show a notification if the statistics are considered 'primary'.

const click$ = fromEvent(document, 'click');

// emits very frequently, as the user moves the mouse around the screen
const hover$ = new Subject<CustomHoverEvent>();

// The user can hover their mouse around the application, and when
// they click, emit the latest hover event. If there was a click without
// any hovers, dont emit anything.
const lastHoverEvent$ = hover$.pipe(
	buffer(click$),
	map(events => events[events.length - 1]),
	filter(event => !!event)
);

const [primaryHover$, secondaryHover$] = partition(
	lastHoverEvent$,
	(event) => event.isPrimaryHover()
);

const primaryHoverStats$ = primaryHover$.pipe(
	map(event => calculatePrimaryHoverStats(event)
);

const primaryHoverStatsNotification$ = primaryHoverStats$.pipe(
	map(stats => createNotification(stats)
);

const secondaryHoverStats$ = secondaryHover$.pipe(
	map(event => calculateSecondaryHoverStats(event)
);

const hoverStats$ = merge(primaryHoverStats$, secondaryHoverStats$);


hoverStats$.subscribe(stats => logStats(stats));

primaryHoverStatsNotification$.subscribe(notification => notify(notification));

Here is how the above streams currently work. Each color represents the code that is run due to each subscription. All of the upstream work is repeated for each subscriber (buffering the events, and calculating the stats for primary).

But here is how I actually want it to work. Where the work that is done along the pipeline is not repeated, and each subscriber only handles its part.

To fix this, I added few share() operators, in order to multicast the work that has already been done.

const lastHoverEvent$ = hover$.pipe(
	buffer(click$),
	map(events => events[events.length - 1]),
	filter(event => !!event),
	share()
);

...

const primaryHoverStats$ = primaryHover$.pipe(
	map(event => calculatePrimaryHoverStats(event),
	share()
);

The type of flow I am trying to replicate looks like this with plain functions.

const lastHoverEvent = getLastHoverEvent();
let hoverStats;
if (lastHoverEvent.isPrimaryHover()) {
	hoverStats = calculatePrimaryHoverStats(lastHoverEvent);
	const notification = createNotification(hoverStats);
	notify(notification);
} else {
	hoverStats = calculateSecondaryHoverStats(lastHoverEvent);
}
logStats(hoverStats);

or like this with a map in the original code

hover$.pipe(
	buffer(click$),
	map(events => events[events.length - 1]),
	filter(event => !!event),
	map(event => {
		let hoverStats;
		if (event.isPrimaryHover()) {
			hoverStats = calculatePrimaryHoverStats(event);
			const notification = createNotification(hoverStats);
			notify(notification);
		} else {
			hoverStats = calculateSecondaryHoverStats(event);
		}
		return hoverStats;
	})
).subscribe(stats => logStats(stats));

Questions

1. Is adding share to your streams really necessary to allow RxJS to work this way? In the 'plain functions' example above, it seems very simple and nothing out of the ordinary, but when you try to do it by splitting up the streams into separate events, duplicate upstream work is done which seems a bit unintuitive to me.

   I feel like I'm missing something. Is this not a good way to go about writing RxJS? I feels so right, when thinking about the logic in terms of the diagrams above, like interconnected streams... but the overhead of share seems weird. Should I just stick to something like the map solution in the last code example?

2. (unrelated question) I used buffer to get the latest hover event on click. I was thinking about using ReplaySubject and withLatestFrom(), to reduce the in-memory storage of hover events (because I only need the latest one).

   const click$ = fromEvent(document, 'click');
   const hover$ = new ReplaySubject<CustomHoverEvent>(1);
   const lastHoverEvent$ = click$.pipe(
   	withLatestFrom(hover$),
   	map(([click, hover]) => hover),
   	...
   );

   but this doesn't work because I need the click$ event to clear whatever hover events have already passed. For example, if a user clicks 2 times in a row without hovering on anything, I don't want a stale/old hover event to be retrieved.

   Is there a better way to do this without buffer? It seems wasteful to store all the events in memory just to get the last one. I was hoping to find a size configuration on buffer, and set it to 1, but I don't see any.

[voliva]

The convention in RxJS is that observables are "unicast" per subscriber by default. This means that every new subscriber will have some sort of internal state for that observable. For example:

const observable$ = interval(1000);
observable$.subscribe(v => console.log(v)); // Will log 0... 1... 2... 3...
setTimeout(() => {
   observable$.subscribe(v => console.log(v)); // Will still log 0... 1... 2... 3... independently
}, 5000);

In some cases you actually need this behaviour, and for the others, you can create a multicast observable by using one of the share operators.

What I usually do is only share those observables that can be referenced from more than another observable. If it's a small observable that I'm declaring in a file without exporting it, where it's just to break down a bigger problem into small chunks, I usually don't do any share on that. But as soon as I have one where I see that I'll be using it in different places on the same file, or as soon as I export it, then I make it multicast.

In your case it looks like the only observable that needs to be multicast is primaryHoverStats$. This is being used by both primaryHoverStatsNotification$ and hoverStats$. So on that place I would add a share() (or similar)

click$ and hover$ are also being used on different places, but I wouldn't bother with them as they are just event emitters.

As for [2] I would say don't be afraid of creating your own operators. I tried thinking about this specific problem for a little time and I couldn't think of any good default operator that could do this, so I would either use buffer as you do here, or I would maybe create a withLatestOnceFrom. An operator just returns a function that takes in an observable and spits out another observable.

[jlemieux]

thank you :) i really appreciate to hear how you do it. I haven't been able to find many articles around this kind of topic, or had the opportunity yet to work with any heavy rxjs users.

In your case it looks like the only observable that needs to be multicast is primaryHoverStats$

I learned that my merge is making 2 subscriptions via hoverStats$, which was duplicating the buffer. I tried to incorporate that into the diagrams with colors. Would this be something you consider negligible, and not worry about duplicating the work for those kinds of things?

If it's a small observable that I'm declaring in a file without exporting it, where it's just to break down a bigger problem into small chunks, I usually don't do any share on that.

Yes! Breaking down bigger problems into chunks... within same file/function/local scope. I'd love to hear more of your thoughts on that, or any examples/articles you have. Taking my example, instead of using if statements & all the logic within a map or something to separate logic (like my last code snippet), I am trying to split and merge the streams to have the code more closely resemble the flow diagrams. But then you start to need some share unless you want duplicate work. You mentioned you tend to not use share in these cases. How do you deal with these kinds of situations?