Declarative, Reactive, Data and Action Streams in Angular

First things first

Let's go ahead and get introductions out of the way. I'm Chandler, a full-stack dev from Alabama, US focusing on Angular and web technologies. Currently, I'm on the Engineering Team at nSide, The School Safety Platform, where I'm currently building IOT dashboards for various devices.

TLDR;

We can use action and data streams declaratively to react to user actions. Leveraging the power of RxJs operators we can transform our data and allow our template to subscribe to the Observable using the Async pipe. Here's a StackBlitz...

That title has a lot of big scary words

I get it if you're new to Angular that sentence could be overwhelming, What does reactive mean? What is declarative? Ok, data fetching I get and I know about the HTTP Client but what are you talking about? Well first let's nail down some key terms, some basic fundamentals, and then let's iterate from there. In this article, I hope to take you to the beginning of the journey I'm still going through. Let's get started!

HTTP Client

There are some basic assumptions that I'm making here. First, I assume you know Angular up to this point in the docs https://angular.io/tutorial/toh-pt6. Second, this isn't going to be an in-depth post on RxJs that would require a series of posts that I'm definitely not qualified TO teach, but I do assume some basic knowledge of things like Observables, Subjects, and subscribing.

So typically in Angular applications, you'll need to get some data from a server and display it for your users to read and interact with. Most of the time we'll use methods like ...

getUsers() {
    return this.http.get<Users[]>(`${this._rootUrl}/users`)
        .pipe(catchError(err=> (
            this.handleError(err)
    ))
}

Following good practices, this method lives in our service and then is called in our component when the component is initialized.

ngOnInit(): void {
//dont forget to unsubscribe! 
    this.subcription = this.userService.getUsers()
        .subscribe(res => this.users = res)
}

Then with the users property in our class, we can use structural directives in our template and display our UI.

<div *ngFor="let user of users">{{user.name}}</div>

This is a completely valid way, but is there a better way?

This is the question I started asking myself. There are a few improvements that could be made. I can think of one specifically by using the Async Pipe to subscribe to the observable so I don't have to manage my own subscriptions. That in of itself is a huge win and makes this a technique that a lot of people use. But I wasn't satisfied and I ran into this problem at work that involved taking data from two different APIs. Using a procedure like pattern didn't feel like the solution to me. Then I saw a talk from Deborah Kurata and was hungry to learn more. Enter in the Declarative and Reactive approach.

Key Terms

First, let's talk about what reactive and declarative mean. For starters, we may recognize the imperative or procedure like way of coding. This is where you describe each and every step of what you want to accomplish. For instance say you want to navigate to a particular file in your filesystem that's nestled deep in a folder structure. You don't remember what's all in there and you can only use the terminal. You're going to be doing a lot of ls and cd until you get where you need to go. Those are procedures. You're describing exactly each step until you get to where you wanna go. Now, what would a declarative approach look like? Simply saying whatever-the-file-is and then the machine figures out the rest by itself. This is nice if the machine knows how to do it but most of the time it does not and we need to describe what we want to happen and all the steps to achieve that. Now reactive programming is a bit harder to explain and I think I'll punt to this article https://gist.github.com/staltz/868e7e9bc2a7b8c1f754
But the short of it is...

//C REACTS to changes in a or b
let a = 1;
let b = 2;
let c = a + b; //3 at this point in time
b = 3;
// C will now be 4. It recomputed its value based on changes to the things that make up its value.

Declarative Data Streams

Now that we know where we have been let's talk about where we are going. Let's declare a data stream.

allUsers$ = this.http.get<User[]>(`${this._baseUrl}/users`).pipe(
    map(users =>
      users.map(
        user =>
          ({
            ...user,
            username: `${user.username}@${user.website}`
          } as User)
      )
    ),
    catchError(err => {
      console.error(err);
      return throwError(err);
    }),
    shareReplay(1)
  );

So let's do a breakdown of this code. In our service, we are declaring a property in our class called allUsers$. Now the $ at the end is a community convention to let you know that this is an Observable stream. Leveraging RxJs operators we could do any kind of data transformation that we want but in this case, I'm using the map operator to receive that User[] value, Array.map() over every User object and then return a new object with a new property called username. Then we cast that return value back as User for good measure in case our typings got messed up and TS couldn't infer the type. Next, we do some error handling with catchError. Now typically you'd log it using a Logger service and keep the logs on a server somewhere but for now, we will just error in the console. I do this here so if there was a problem we could see in the service where it was and then we return throwError so that it propagates up to the object that subscribes to it (our component) and they can handle it there. Lastly, we shareReplay(1). This allows us to cache and reuse the data we've already got if someone subscribes to the Observable later on. For instance, say we have two components that subscribe to the Observable at different times. The first subscriber will kick off the HTTP request. Since we are sharing and replaying the result, the next subscriber gets the value the first one did and another HTTP call does not have to be made.

How to use Data streams

Using a data stream is incredibly simple! Here is a recipe for using the stream,

1. Inject the service into the target component using private userService: UsersService
2. Capture a reference of the data stream from your service. For example.

this.users$ = this.userService.allUsers$.pipe(
//Our error thrown from the service bubbles to the component where we handle
//it. I'm just simply setting a property to true
//You have to return an Observable so I just return a empty observable that completes
      catchError(err => {
        this.error = true;
        return EMPTY;
      })
    );

1. Now we subscribe from our template with the Async pipe!

<ng-container *ngIf="users$ | async as users">

Reactive Data Streams with Action Streams

Some of the times our data in our applications is read-only. This makes things easy for us because we just subscribe to the Observable and display the data in the template and call it a day. Other times we want our users to have the ability to modify the data or take actions on that data. We can call these user actions Action Streams. We can create action streams using RxJS Subjects and throwing values into those streams. Taking an action stream we can have our data stream react to those actions and leverage RxJs operators to modify our data. Here's an example of declaring an action stream that emits a selectedUser.

private selectedUserSubject = new BehaviorSubject<number>(null);
  selectedUserAction$ = this.selectedUserSubject.asObservable();
  onSelectedUser(id) {
    this.selectedUserSubject.next(id);
  }

To break this down we have our BehaviorSubject that emits numbers and its Observable counterpart. We then have a helper method that when called from the component emits the ID of the selected user into that action stream. Using this information combined with the allUser$data stream we can create a stream that emits the selected user and reacts to the user's action.

selectedUserData$: Observable<User> = combineLatest([
    this.allUser$,
    this.selectedUserAction$
  ]).pipe(
    map(([allUsers, selectedUser]) => allUsers.find(u => u.id === selectedUser))
  );

We set the selectedUserData$ property to be the result of the combineLatest operator. What this does is takes the last value emitted from both streams and return those values as an array. Using array destructuring in the map operator we return the result of the allUsers array find function. So now every time we emit a new ID into the action stream this pipe runs returning us a new user.

Thanks for sticking around! I'm trying to get into writing more so if you have any tips or things I can improve on please let me know!