## Anatomy of an observable

<pre><code class="typescript" data-trim data-noescape>
            <span class="fragment highlight-current-red" data-fragment-index="4">const subscription</span> =
                <span class="fragment highlight-current-red" data-fragment-index="1">Observable.interval(1000)</span>
                <span class="fragment highlight-current-red" data-fragment-index="2">.pipe(
                    map(x => Math.random() * x),
                    filter(x => x < 1),
                )</span>
                .subscribe(<span class="fragment highlight-current-red" data-fragment-index="3">x => console.log(`Logged ${x}`)</span>);
            </code></pre>

<span>Observable</span>

<span>Operators</span>

<span>Observer</span>

<span>Subscription</span>

</div>

<aside class="notes">
              Notes:
              - 1: the observable itself, represents the stream of values
              - 2: operators, transform the stream
              - 3: observer, receives emissions from the observable
              - 4: subscription, represents the execution of the observable, can be unsubscribed from
                - effectively "calls" the observable
            </aside>

## `of`, `interval`, `from`, `fromPromise`

```typescript
            // Outputs: 1, 2, 3, 4
            Observable.of(1, 2, 3, 4)
                .subscribe(x => console.log(x));

// Outputs: 0, 1, 2, 3, ...
            Observable.interval(500)
                .subscribe(x => console.log(x));

// Converts the array to an observable of its items.
            Observable.from([1, 2, 3])
                .subscribe(x => ...);

// Converts the Promise to an observable.
            // Could also have just used `from`.
            Observable.fromPromise(fetch('spacetime.jpg'))
                .subscribe(blob => ...);
            ```
            <aside class="notes">
              Notes:
              - `of` just outputs what is passed to it
              - synchronous upon subscription
              - `interval` outputs the index
              - `from` can be used for promises, arrays, etc
              - `fromPromise` may be more readable
              - generally useful when converting non-observables into the observable space
            </aside>

## Using Subjects to multicast

```typescript
            const subject = new Subject<number>();
            Observable.interval(1000).pipe(
                take(3),
                map(v => {
                  // Sneaky side effect.
                  console.log(`Emit: ${v}`);
                  return v;
                }),
            )
            // This subscription creates one new execution of
            // the observable.
            .subscribe(subject);

// Even though there are multiple subscriptions on
            // the subject, each 'Emit:' will only print once.
            const sub1 = subject.subscribe(v => { ... });
            const sub2 = subject.subscribe(v => { ... });
            ```
            <aside class="notes">
              Notes:
              - we have a sneaky side effect in the map (could've also been a `tap`)
              - can act as a multicast point (bridge/proxy)
              - will be useful in some of the upcoming patterns
            </aside>

## Using AsyncPipe

```typescript
            @Component({
              template: `<div>{{curvature | async}}<div>`,
            })
            export class UniverseComponent implements OnInit {
              curvature: Observable<number>;
              constructor(private service: SpaceTimeService) {}
              ngOnInit() {
                this.curvature = this.service.calcCurvature();
              }
            }
            ```
            <aside class="notes">
              Notes:
              - async subscribes to an observable and returns the latest value it emits
              - automatically unsubscribes via Angular's component lifecycle
              - still an observable, so can be easily manipulated
              - caveats:
                - returns `null` until the observable actually emits, so blindly accessing properties on the result is dangerous
                - if it makes sense, that could be worked around via `startWith()`
                - each use of `async` creates a separate subscription! multiple executions if not multicasted
            </aside>

## Multiple AsyncPipes

```typescript
            @Component({
              template: `
                <div>
                  Universe curvature is {{curvature | async}}!
                  {{curvature | async}}!!!1
                <div>`,
            })
            export class UniverseComponent implements OnInit {
              ...
              ngOnInit() {
                this.curvature = this.service.calcCurvature().pipe(
                    // Ensures that source observable is only
                    // executed once.
                    shareReplay(1),
                );
              }
            }
            ```
            <aside class="notes">
              Notes:
              - async subscribes to an observable and returns the latest value it emits
              - automatically unsubscribes via Angular's component lifecycle
              - still an observable, so can be easily manipulated
              - caveats:
                - returns `null` until the observable actually emits, so blindly accessing properties on the result is dangerous
                - each use of `async` creates a separate subscription! multiple executions if not multicasted
            </aside>

## Managing subscriptions

If an explicit `.subscribe()` is needed

```typescript
            @Component({ ... })
            export class UniverseComponent implements OnDestroy {
              private onDestroy = new Subject<void>();
              ngOnDestroy() {
                // Attach to the component's lifecycle.
                this.onDestroy.next();
              }
              createGalaxy(): void {
                this.http.post('/api/galaxy', this.data).pipe(
                    switchMap(...),
                    filter(...),
                    // Ensure that this subscription is cleaned
                    // up on component destruction.
                    takeUntil(this.onDestroy),
                )
                .subscribe();
              }
            }
            ```
            <aside class="notes">
              Notes:
              - if `async` isn't appropriate, or an explicit .subscribe() is needed for some other reason...
              - can use an `onDestroy` Subject with `takeUntil` to ensure that subscriptions are cleaned up
              - orphaned subscriptions can cause memory leaks
              - also good to choose the minimize observable lifetime in other ways
                - `first()`, `take(n)`, etc
            </aside>

## The "imperative" approach

```typescript
            @Component({ ... })
            export class UniverseComponent implements OnInit {
              // Hard to trace how this state will change over time.
              galaxies: Galaxy[] = [];
              ngOnInit() {
                this.route.data.galaxies.subscribe(galaxies => {
                  this.galaxies = galaxies;
                });
              }
              deleteGalaxy(galaxy: Galaxy) {
                this.universeService.deleteGalaxy(galaxy)
                    .subscribe(() => {
                      // Harder to test this logic in isolation.
                      // Mutation also makes it more difficult to
                      // use OnPush.
                      const index = this.galaxies.indexOf(galaxy);
                      this.galaxies.splice(index, 1);
                    });
              }
            ```
            <aside class="notes">
              Notes:
              - omitting all the cleanup logic as well
              - could work, but gets harder as more sources of events are added
            </aside>

## Multiple transforms

```typescript
            @Component({ ... })
            export class UniverseComponent implements OnInit {
              // Use in the template with `async`.
              galaxies: Observable<Galaxy[]>;
              deletedGalaxies = new Subject<Galaxy>();
              ngOnInit() {
                // `applyStateTransforms` applies the given transforms
                // to the observable of starting data, over time.
                this.galaxies = applyStateTransforms(
                    this.route.data.galaxies,
                    // `stateTransforms` generates the transforms based
                    // on an observable of events and a reducer.
                    stateTransforms(this.deletedGalaxies, deleteReducer),
                    ..., // More transforms can be added here.
                );
              }
              deleteGalaxy(galaxy: Galaxy) {
                this.universeService.deleteGalaxy(galaxy)
                    .subscribe(this.deletedGalaxies);
              }
            }
            ```
            <aside class="notes">
              Notes:
              - `galaxies` can be used in the template directly, with async
              - initial state loaded via a Route Resolver
              - the transforms are applied over time to the initial state
              - when calling `deleteGalaxies` on the service, could also do a `takeUntil(this.onDestroy)`
              - easier to understand/reason about the flow of data in the component
                - also easier to reason about time / asynchronicity
              - easy to add new transforms in a structured way
            </aside>

## `applyStateTransforms`

```typescript
            // Composes outputs of `stateTransforms()` with
            // an observable of initial state.
            function applyStateTransforms<S>(
                state: Observable<S>,
                ...transforms: Array<Observable<(state: S) => S>>
                ): Observable<S> {
              const mergedTransforms = merge(...transforms);
              return state.pipe(
                switchMap(startState => transforms.pipe(
                  scan((state, trans) => trans(state), startState),
                  startWith(startState),
                )),
              );
            }
            ```
            <aside class="notes">
              Notes:
              - merges all the transforms (so there can be many)
              - switchMap's the state observable, meaning it will take precedence
              - can almost think of the `scan()` as a loop
              - important for there to be a single scan()!
                - if this took just a single transform observable, and composed with itself, scan internal state could get out of sync
                - only the downstream scan would see updates from the upstream on, so potential of mismatch
            </aside>