Angular Reactive Development

Introduction

Background

RxJs originally was developed by Microsoft in Rx.NET. It works on Java and other platforms. The definition is "RxJs is a library for composing asynchronous and event-based programs using observable sequences"

Other approaches

• Callbacks - difficult to use with nested async operations
• Promises - can only handle a single emission and is not cancellable
• Asyc/await - can only handle a single emission and is not cancellable

Why Rxjs

• One - technical for all data e.g. mouse clicks, api calls
• Compositional - with the operators easy to transform and combine
• Watchful - With its push model to notify subscribers
• Laxy - Only executes when subscribed to
• Handles Error - Built in
• Cancellable

Angular already use RxJs in

• Routing
• Reactive Forms
• HttpClient

Reactive Development

This is characterized by

• Quick to react to user - drrrrrr
• Resilient to failure -error checking all the time
• Reactive to state changes

Resources

The demos were at https://github.com/DeborahK/Angular-RxJS

Terms and Syntax

Observer

Observes the stream and responds to its notification

• Next item
• Error occurred
• Complete

const observer = {
  next: apple => console.log(`Apple emitted ${apple}`),
  error: err => console.log(`Error occurred ${err}`),
  complete: ()=> console.log(`No more Apples`)
}

Obeservable Stream

Also call an Observable sequence, a stream and can have a stream of any type of data such as number, string, events, objects or other streams and of course http requests

const appleStream = new Observerable(appleObserver => {
   appleObserver.next(`Applie 1`)
   appleObserver.next(`Applie 2`)
   appleObserver.complete()
})

Subscription

In order to start the stream we need to subscribe. We do this by passing an observer to the stream.

const sub = applestream.subscribe(observer)

It is unusual to see this in real code. In reality must developers would just pass it in

const sub = appleStream.subscribe(
  apple => console.log(`Apple emitted ${apple}`),
  err => console.log(`Error occurred ${err}`),
  ()=> console.log(`No more Apples`)
)

There are several ways to unsubscribe or stop the stream.

• Call Complete
• Use completing operators like take(1)
• Throw an error
• Call unsubscribe

Without stopping the stream memory leaks can occur

Creating an Observable

of

Creates an observable from a set of defined values. If emits each item and then calls complete.

const appleStream = of(`Apple 1`, `Apple 2`)

from

Same as of but create an item from each value in the array.

const appleStream = from([`Apple 1`, `Apple 2`])

Differences

The difference is more apparent between of and from when you consider passing [`Apple 1`, `Apple 2`] to the of function. This will only produce one item. To get the same result you would need to use the spread operator of(...apples) to get two results.

Operators

Introduction

There are lots of these. With pipe the result of the operator is passed to the next operator.

of(2,4,6)
 .pipe(
    map(item => item * 2),
    tap(item => console.log(item)),
    take(2)
 ).subscribe(console.log)

Marble Diagrams

These are presented back to front in my view. As the first item is shown on the left and the last item on the right. Having said that, here is an example.

Declarative vs Imperative

Bit of a sideline but this kept coming up in the course and needed to understand what they meant. So looking at https://ui.dev/imperative-vs-declarative-programming I got the following
I’m going to ask you a question. I want you to think of both an imperative response and a declarative response.
“I’m right next to Wal-Mart. How do I get to your house from here?”
Imperative response Go out of the north exit of the parking lot and take a left. Get on I-15 North until you get to the 12th street exit. Take a right off the exit like you’re going to Ikea. Go straight and take a right at the first light. Continue through the next light then take your next left. My house is #298.
A declarative response My address is 298 West Immutable Alley, Eden, Utah 84310

Right now this is sounding more like requirements, declarative, vs imperative, design. Still confused. But I think declarative means that we right functional code wrapping around the implementation. Just like the old days.

Going Reactive

Async Pipes

Good for reactive UI

• subscribes to observable
• returns each emitted value
• when a new item is emitted, component is marked to be checked for changes
• Unsubscribes when component is destroyed

Error Handling

Catch and Replace

This where you catch the error and replace the error with something else. In this case it is very stupid.

return this.http.get<Product[]>(this.productUrl)
 .pipe(
    catchError(err => {
       console.error(err)
       return of([
           {id:1, productName: 'cart'}, 
           {id:2, productName: 'hammer'}
       ])
  })

We need to remember catchError is just an operator.

Catch and Rethrow

From the course this resulted propagating the error further us the stream.

return this.http.get<Product[]>(this.productUrl)
 .pipe(
    catchError(err => {
       console.error(err)
       return throwError(err)
  })

So instead of throwError a function is used to do this.

private handleError(err: any): Observable<never> {
    // in a real world app, we may send the server to some remote logging infrastructure
    // instead of just logging it to the console
    let errorMessage: string;
    if (err.error instanceof ErrorEvent) {
      // A client-side or network error occurred. Handle it accordingly.
      errorMessage = `An error occurred: ${err.error.message}`;
    } else {
      // The backend returned an unsuccessful response code.
      // The response body may contain clues as to what went wrong,
      errorMessage = `Backend returned code ${err.status}: ${err.body.error}`;
    }
    console.error(err);
    return throwError(errorMessage);
  }

This is eventually passed on to the component.

    this.products$ = this.productService.getProducts().pipe(
      catchError(err => {
        this.errorMessage = err;
        return EMPTY;
      })
    );

Change Detection

Using Async Pipes means, no need to subscribe or unsubscribe and improves change detection. There are two types of change detection. Default (check always)

• every component is check when ANY change is detected

OnPush (mimimizes change detection) when

• @Input properties change
• Event emits
• A bound Observable emits (using the async pipe)

So we change change this in our app

@Component({
  templateUrl: './product-list.component.html',
  styleUrls: ['./product-list.component.css'],
  changeDetection: ChangeDetectionStrategy.OnPush
})

However because we use the OnPush the component, although we change the errormessage in the catchError the UI is not updated.

Declarative Pattern

Previously the pattern was

// Service
  getProducts(): Observable<Product[]> {
    return this.http.get<Product[]>(this.productsUrl)
      .pipe(
        tap(data => console.log('Products: ', JSON.stringify(data))),
        catchError(this.handleError)
      );
  }
// Component
    this.products$ = this.productService.getProducts().pipe(
      catchError(err => {
        this.errorMessage = err;
        return EMPTY;
      })
    );

We can change this to be more declarative

// Service
 products$: Observable<Product[]> = this.http.get<Product[]>(this.productsUrl)
  .pipe(
    catchError(this.handleError)
  );

// Components
  products$ = this.productService.products$.pipe(
    catchError(err => {
      this.errorMessage = err;
      return EMPTY;
    })
  );

Mapping Returned Data

Mapping a Return Array

As mentioned previous apple[] is not the same as apple. To make this work they used the map function on the array which I am unsure if this is the right approach but it works.

  products$: Observable<Product[]> = this.http.get<Product[]>(this.productsUrl)
  .pipe(
    map(products =>
      products.map(product => ({
        ...product,
        price: product.price * 1.5,
        searchKey: [product.productName]
      }) as Product)
    ),
    catchError(this.handleError)
  );

So their recommendation is to leave the array as is and do

• Map emmitted array
• Map the array elements
• Transform each array element

Combining Streams

Introduction

We my want to do this to

• Map id to string
• Combine with other sources e.g. share price, reviews
• React to actions e.g. re-filter
• Simplify for template code

CombineLatest

This is a function

• Takes in set of stream
• Creates and output stream

At least one value must have been emitted from each stream and it completes when all streams are complete.

ForkJoin

This is a function Creates an output stream using the lastest data from the input streams

WithLatest

This is a function

• Takes in set of stream
• Creates and output stream

At least one value must have been emitted from each stream, it emits a value each time the source stream gets a value. It completes when all streams are complete.

An Example

For the lookup table here is the example. Combine the two API calls to provide the additional data. The key thing is to understand the timings of when the data is calculated. For this case we have the latest data.

  productsWithCategory = combineLatest([
    this.products$,
    this.productCategoryService.productsCategories$,
  ]).pipe(
    map(([products, categories]) =>
      products.map(
        (product) =>
          ({
            ...product,
            price: product.price * 1.5,
            category: categories.find((c) => product.categoryId === c.id).name,
            searchKey: [product.productName],
          } as Product)
      )
    ),
    catchError(this.handleError)
  );

Reacting to Actions

Filtering

No surprises, we map the data and add an array.filter. Change the UI to use this new Observable.

  productsSimpleFilter$ = this.productService.productsWithCategory.pipe(
    map(products =>
      products.filter(product =>
        this.selectedCategoryId ?
          product.categoryId === this.selectedCategoryId : true))
    );

An Aside

In the course they created a combobox which held a list of categories which could be used for filtering. To give an example so I can copy here is the code.
For the observable

  categories$ = this.productCategoryService.productsCategories$.pipe(
    catchError(err => {
      this.errorMessage = err;
      return EMPTY;
    })
  );

For the template

          <select class="form-control"
                  (change)="onSelected($event.target.value)">
            <option value="0">- Display All -</option>
            <option *ngFor="let category of categories$ | async"
                    [value]="category.id">{{ category.name }}</option>
          </select>

And for the select

  onSelected(value): void {
    this.selectedCategoryId = +value;
  }

Back to the rest

Creating Action Stream

Create an action stream

private categorySelectedSubject = new Subject<number>()

Expose the subjects Observable

categorySelectedAction$ = this.categorySelectedSubject.asObservable()

Combine with the stream

  products$ = combineLatest([
    this.productService.productsWithCategory,
    this.categorySelectedAction$
  ]).pipe(
    map(([products, selectedCategoryId]) =>
      products.filter(product =>
        selectedCategoryId
          ? product.categoryId === selectedCategoryId
          : true
      )
    ),
    catchError((err) => {
      this.errorMessage = err;
      return EMPTY;
    })
  );

Emit value on change

Finally emit a value to cause the combineLatest to be re-evaluated

  onSelected(value): void {
    this.categorySelectedSubject.next(+value);
  }

Initial Value

Subject does not set an initial value. To resolve this there are two options add a startWith operator to the subject.

  products$ = combineLatest([
    this.productService.productsWithCategory,
    this.categorySelectedAction$.pipe(
      startWith(0),
    ),
  ]).pipe(
...
  );

Or use a behaviourSubject

private categorySelectedSubject = new BehaviorSubject<number>(0);