armanozak

import { timer } from "rxjs";
import { timeoutWith } from "rxjs/operators";

timer(5000, 1000)
  .pipe(timeoutWith(2000, of("timeout")))
  .subscribe(console.log);
// (after ~2s) timeout

armanozak

import { timer } from "rxjs";
import { timeout } from "rxjs/operators";

timer(5000, 1000)
  .pipe(timeout({ first: 2000, with: _ => of("timeout") }))
  .subscribe(console.log);
// (after ~2s) timeout

armanozak

import { defer, from } from "rxjs";
import { retry, tap } from "rxjs/operators";

const values = ["_", 0, 1, 0, 2, 0, 3, 0, 0, 0, 4];

defer(() => {
  values.shift();
  return from(values);
})
  .pipe(

armanozak

import { defer, from } from "rxjs";
import { retry, tap } from "rxjs/operators";

const values = ["_", 0, 1, 0, 2, 0, 3, 0, 0, 0, 4];

defer(() => {
  values.shift();
  return from(values);
})
  .pipe(

armanozak

import { interval, of, zip } from "rxjs";
import { map, shareReplay } from "rxjs/operators";

const shared$ = zip(interval(1000), of("A", "B", "C", "D", "E")).pipe(
  map(([, char]) => char),
  shareReplay({ refCount: true, bufferSize: 3 })
);

shared$.subscribe(console.log);
// (~1s apart) A, B, C, D, E

armanozak

import { interval, of, ReplaySubject, zip } from "rxjs";
import { map, share } from "rxjs/operators";

const shared$ = zip(interval(1000), of("A", "B", "C", "D", "E")).pipe(
  map(([, char]) => char),
  share({
    connector: () => new ReplaySubject(3),
    resetOnComplete: false,
    resetOnError: false,
    resetOnRefCountZero: false

armanozak

import { merge, of } from "rxjs";
import { connect, filter, map } from "rxjs/operators";

const chars$ = of("A", "b", "C", "D", "e", "f", "G");

chars$
  .pipe(
    connect(shared$ =>
      merge(
        shared$.pipe(

armanozak

import { connectable, merge, of } from "rxjs";
import { filter, map } from "rxjs/operators";

const chars$ = of("A", "b", "C", "D", "e", "f", "G");
const connectableChars$ = connectable(chars$);

const lower$ = connectableChars$.pipe(
  filter(x => x.toLowerCase() === x),
  map(x => `lower ${x.toUpperCase()}`)
);

armanozak

import { animationFrames, combineLatest, concat } from "rxjs";
import { endWith, map, takeWhile } from "rxjs/operators";

const h1 = document.querySelector("h1")!;

combineLatest({
  x: tween(0, 200, 3600),
  y: wave(25, 1200, 3)
}).subscribe(({ x, y }) => {
  h1.style.transform = `translate3d(${x}px, ${y}px, 0)`;

armanozak

import { config, throwError } from "rxjs";

config.onUnhandledError = console.warn;
throwError(() => "TEST ERROR 1").subscribe();
throwError(() => "TEST ERROR 2").subscribe({ error: console.warn });
// (synchronously) TEST ERROR 2
// (asynchronously) TEST ERROR 1