RxJS snippets based on https://github.com/brechtbilliet/ngxReactiveToolkit

rxjs_snippets.md

Pipeable operators

import {filter, map} from 'rxjs/operators';
const interval$ = interval(1000);
const new$ = interval$
    .pipe(
        filter(v => v % 2 === 0),
        map(v => v *2)
    )

Subscriptions

class AppComponent implements OnInit, OnDestroy {
    destroy$ = new Subject();
    ngOnInit() {
        // interval$: 0--1--2--3--4--5--6--...
        // destroy$:  -------------true|
        // result:    0--1--2--3--4|
        const interval$ = interval(1000);
        interval$
            // let the interval$ stream live 
            // until the destroy$ Subject gets a value
            .pipe(takeUntil(this.destroy$))
            .subscribe(r => console.log(r));
    }

    ngOnDestroy() {
        // when the component get's destroyed, pass something to the
        // destroy$ Subject
        this.destroy$.next(true);
    }
}

Avoiding nested subscribes

class AppComponent {
    user: User;
    constructor(
        private route: ActivatedRoute, 
        private userService: UserService) 
    {
        // when the params of the route changes,
        // we want to fetch the user and set the user property
        //
        // GOOD: we have created a single subscribe which makes
        // the flow way easier and gives us the control we need
        // It’s recommended to use higher-order streams like mergeMap or switchMap
        this.route.params
            .pipe(
                map(v => v.id),
                switchMap(id => this.userService.fetchById(id))
            )
            .subscribe(user => this.user = user)
    }
}

Avoiding manual subscribes in Angular

// Angular has this super cool feature called the async pipe. It’s used to consume streams directly in the template The async pipe does 3 things for us:
// It subscribes to the stream and passes the value to a component
// It unsubscribes automatically when the component gets destroyed (removes a lot of unsubscribe logic)
// Triggers change detection automatically
@Component({
    ...
    template: `
        <user-detail [user]="user$|async"></user-detail>
    `
})
class AppComponent {
    // expose a user$ stream that will be 
    // subscribed in the template with the async pipe
    user$ = this.route.params.pipe(
        map(v => v.id),
        switchMap(id => this.userService.fetchById(id))
    );

    constructor(
        private route: ActivatedRoute, 
        private userService: UserService) {
    }
}

Don't pass streams to components directly

// GOOD
// app.component.ts
@Component({
    selector: 'app',
    template: `
        <user-detail [user]="user$|async"></user-detail>
    `
})
class AppComponent implements OnInit {
    users$: Observable<User[]> = this.http.get(...);
    user: User;
    ngOnInit(){
        // the app component (smart) subscribes to the user$ which will
        // do an XHR call here
        this.users$ = this.http.get(...);
    }
    ...
}

// user-detail.component.ts
@Component({
    selector: 'user-detail',
    template: `
        
    `
})
class UserDetailComponent {
    // This component doesn't even know that we are using RxJS which
    // results in better decoupling
    @Input() user: User;
}

Don’t pass streams to services

// GOOD Use switchMap over mergeMap if possible, since it will unsubscribe the previous stream
// app.component.ts
class AppComponent {
    users$ = this.http.get(...)
    filteredusers$ = this.users$
        .pipe(switchMap(users => this.fooService.filterUsers(users)));
    ...
}

// foo.service.ts
class FooService {
    // this is way cleaner: this service doesn't even know
    // about streams now
    filterUsers(users: User): User[] {
        return users.pipe(filter(user => user.age >= 18);
    }
}

Sharing subscriptions

//  most streams are cold by default, every subscription will trigger the producer of these streams
// share the subscription using the share() operator
// Sharing a stream makes it hot. This means that if we subscribe after the value is produced, 
// we will miss that value. In that case we might want to use shareReplay(1) or publishReplay(1).refCount() instead of share(). 
// This will keep the last value in memory for us.
@Component({
    selector: 'app',
    template: `
        Number of users: {{numberOfUsers$|async}}
        <users-grid [users]="users$|async"></users-grid>
    `
})
// GOOD
class AppComponent {
    users$ = this.http.get(...).pipe(share());
    // the subscription on this stream will execute the xhr call again
    numberOfUsers$ = this.users$.pipe(map(v => v.length); 
}

// alternative way
@Component({
    selector: 'app',
    template: `
        <div *ngIf="users$|async as users; else loading">
            Number of users: 
            <users-grid [users]="users"></users-grid>
        </div>
        <ng-template #loading>Loading...</ng-template>
    `
})
class AppComponent {
    // This stream will only subscribed to once
    users$ = this.http.get(...);
}