RxJS Patterns: Creating an Action Stream
Tags: angular, rxjs, rxjspatterns
Published: 2019-10-09 18:15:50

[RxJS] is an extremely powerful observable based library, used at the very core of Angular, that allows us to write our applications in a truly reactive paradigm.

[RxJS]: https://rxjs.dev

[Creation operators] are standalone functions that can create a new observable.

[Creation operators]: https://rxjs.dev/guide/operators#creation-operators

```ts
of(1,2,3)
  .subscribe((val) => {
    console.log('::val', val) // output: 1, 2, 3
  })
```

These type of operators pose a new challenge when you want to follow a truly reactive paradigm.

<blockquote class="twitter-tweet" data-lang="en"><p lang="en" dir="ltr">In <a href="https://twitter.com/hashtag/rxjs?src=hash&amp;ref_src=twsrc%5Etfw">#rxjs</a> .subscribe() is where reactive programming ends</p>&mdash; Michael Rx Hladky (@Michael_Hladky) <a href="https://twitter.com/Michael_Hladky/status/1180316203937681410?ref_src=twsrc%5Etfw">October 5, 2019</a></blockquote>
<script async src="https://platform.twitter.com/widgets.js" charset="utf-8"></script>

Using observables comes with a greater responsibility of always unsubscribing. Fortunately Angular provides an `async` pipe that handles the subscription and un-subscription logic for us. This worked great until I had to make a http post request. 

There was no obvious way to use the `async` pipe in this context, as this request is made upon a user action. And I really, really, did not want to use `.subscribe()`.

Cue the action stream:

```html
<!-- some.component.html -->

<ng-container *ngIf="{ postAction: postStream$ | async } as postStream">
```

We subscribe to a `postStream$` using the `async` pipe. The observable will not emit any values until the http request has returned, therefore, we wrap the subscription in an object. This will be a truthy value, so the `*ngIf` condition will be met, which allows our template to always be rendered.

The complete template:

```html
<!-- some.component.html -->

<ng-container *ngIf="{ postAction: postStream$ | async } as postStream">
  <p>Click the button to see the magic happen :)</p>
  <button (click)="onBtnClick()">Post data to back end</button>

  <!-- You can use this to show a success or error message, or loading state -->
  <p *ngIf="postStream.postAction">Saved!</p>
</ng-container>
```

In our component file:

```ts
// app.component.ts

// create your action stream
private readonly postAction$ = new Subject();
```

We create an action stream; simply a subject that we'll call `.next()` on when an event in our template has taken place.

```ts
// app.component.ts

// subscribe to this post stream (high order observable) in your template to start listening 
// for values emitted by the action stream
postStream$ = this.postAction$.pipe(
  // pick the correct operator for your use case
  exhaustMap(()=> {
    return of('my post'); // this could be your http post
  })
)
```

We create a higher order observable `postStream$`, that has an outer observable, our action stream we created earlier, and an inner observable that we map to. For this example we use a creation operator `of()` but this could be a http post request.

```ts
// app.component.ts

onBtnClick() {
  this.postAction$.next()
}
```

We then have a method on the class that will trigger our action stream.

The complete component:

```ts
// app.component.ts

export class AppComponent  {
  // create your action stream
  private readonly postAction$ = new Subject();
  
  // subscribe to this post stream (high order observable) in your template to start listening 
  // for values emitted by the action stream
  postStream$ = this.postAction$.pipe(
    // pick the correct operator for your use case
    exhaustMap(()=> {
      return of('my post'); // this could be your http post
    })
  )

  onBtnClick() {
    this.postAction$.next()
  }
}
```

With this approach, we can avoid having to manually call `.subscribe()` when making http post requests.

You can find the full code on [StackBlitz]

[StackBlitz]: https://stackblitz.com/edit/rxjs-action-stream