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.

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)
  );

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;
    })
  );