Managing the RxJs loose subscriptions

Many of us have struggled a lot through issues resulting from loose subscriptions as its difficult to trace and debug.
And its very common that the developers skip or forget to clean the subscriptions and later on put themselves in trouble of debugging for some weird errors.

What happens when we subscribe to some observable stream ?

When we subscribe to some observable stream it returns you a subscription(a channel of trust between observable and the observer). Now the communication channel is open, So the observable stream becomes active and is free to send data and the observer(the subscriber) will listen to its observable until and unless it unsubscribes from the resulting subscription or the respective observable stream completes.
We should be concerned about the cleanup of the formed subscription. Now the question arises is “Why”.

Why to close any open observable stream?

A loose subscription can be harmful as it can lead to unwanted memory leaks as the observable stream is left open, potentially even after a component has been destroyed.

Solutions,

There are two approaches for cleaning the potential active observable stream.
1. Unsubscribing from the formed subscription (traditional approach).
2. Making the observable stream complete (utilising the power of RxJs).

Unsubscribing from the subscriptions

This is a traditional way to unsubscribe from the subscriptions. This way is suitable when we have to deal with single observable stream in the component class.
But when we possess multiple subscriptions at a place this approach becomes cumbersome.
See in the above example how I’m creating class level variable for each subscriptions and later on in the OnDestroy life-cycle hook of the component we have to do so much(more subscription objects to manage) to unsubscribe from each subscription. More code means less readability and maintainability too.
So, lets get to a better version of it.

Unsubscribing from subscription (Subscription Class)

The above example is better in a way that we don’t have to create separate variable for each subscription. We can just keep our subscriptions as a child of single subscription and finally in the ngOnDestroy callback function unsubscribe from the parent subscription.
But wait don’t you think it stills looks messy, that’s why I don’t like this approach.

Making the observables complete (using takeUntil)

It was getting tedious and we want a cleaner approach of writing code. So now we are going to utilise the power of RxJs and its operators to declaratively manage subscriptions.

The solution is to compose our subscriptions with the takeUntil operator and inside takeUntil we just need an observable to emit which will trigger the completion of the corresponding observable.
Advantages of using this approach are cleaner code, we don’t need to keep references to our subscriptions anymore, easier to locate wired up RxJs operator, more maintainable. Since it completes the observable and hence triggering any completion event on the observable.

Although we are going slightly misleading here in terms of RxJs semantics but shouldn’t be bothered about. The semantics says that completing an observable is a sign that the producer wants to tell the consumers it’s done, whereas unsubscribing is the consumer telling the producer it no longer cares about the data.

Other operators that completes the observable

The choice of RxJs operator we use to finish off the communication depends on the scenario how or when should we kill the observable stream.
Suppose at the time of component destruction we want all the Observable-Observer channel to end, our choice would be “takeUntil”. But there are other operators which kill the observable stream.

• take(n): emits N values before stopping the observable.
• takeWhile(predicate Fn): tests the emitted values against a predicate, if it returns ‘false’ it will complete.
• first(): emits the first value and completes.
• first(predicate Fn): checks each value against a predicate function, if it returns ‘true’, it emits that value and completes.

Making Observables complete(centralised with utility function)

If we have many components, so we have to do this rework of cleanup of subscriptions inside each component, So we are simply repeating the code i.e. violating the DRY(Don’t Repeat Yourself, a fundamental concept of good programming).
As we know that takeUntil needs a notifier(inner observable) to emit so that it can complete the outer observable stream. We are going to pass a function that eventually will be returning an observable but that function would be at a single place (written only once), perhaps in some service.

This above function named “componentDestroyed” is supposed to be passed inside takeUntil like
SomeObservable.pipe(takeUntil(componentDestroyed(this))),subscribe()
If you notice, we are passing a reference to the current component through ‘this’ keyword on which our utility function(componentDestroyed Fn) operates.
But this approach restricts that particular component to implement ngOnDestroy lifeCycle hook. We can further abstract it with the help of an annotation function. So now lets get to it.

Making Observables complete(further abstracted through annotation)

Now, what this annotation function would do is just augmenting ngOnDestroy method and making sure that the component contains this method, so now our utility function doesn’t need to restrict the component to contain such method.

There are many ways further we can play around these loose subscriptions using RxJs operators, Subscription Class, utility functions and annotation functions for their management. Finally as a bonus I will be presenting you with an another technique to autocomplete Observable using takeWhile RxJs operator.

Making Observables complete using takeWhile

We’ll be using this magical operator to kill the observable stream as soon as it finds the returned value from the predicate function as false. Let’s look at the following example.

Custom RxJs operator with all functionality encapsulated

This one is my favourite of all. It doesn’t need further explanation if you are well worse with creating your own custom RxJs operators(prior is covered above).

From here you can take it ahead to explore new possibilities and solutions. I hope you enjoyed this article, if so please make a clap to appreciate.

GEMM is my Code. ~Ankit Kaushik