RxJS

参考

create observable

// create
const stream$ = Observable.create((observer) => {
  observer.next(1);
  observer.error('error message');
  observer.complete();

  return function () {
    someCleanUp();
  };
});

// producer
class Producer {
  i = 0;
  nextValue() {
    return this.i++;
  }
}

// range
let stream$ = range(1, 3);

// from
let stream$ = from(/* Array, Promise */);

// of
let stream$ = of(1, 2, 3, 4);

observe

// subscribe
const subscription = stream$.subscribe(onSuccess, onError, onComplete);

// unsubscribe
subscription.unsubscribe();

// cold observable => 1,2,3,1,2,3
let stream$ = of(1, 2, 3);
stream$.subscribe((data) => console.log(data));
stream$.subscribe((data) => console.log(data));

// hot observable => 0,1,2,2,3,3,4,4
let stream$ = interval(1000).pipe(
  take(5),
  publish(), // important
);
stream$.subscribe(
  (data) => console.log(`subscriber from first minute: ${data}`),
  (err) => console.log(err),
  () => console.log('completed'),
);
setTimeout(() => {
  stream$.subscribe(
    (data) => console.log(`subscriber from 2nd minute: ${data}`),
    (err) => console.log(err),
    () => console.log('completed'),
  );
}, 2100);
stream$.connect(); // important

// warm observable => 0,1,2,3,3,4,4
// (wait until a subscriber comes at least 1)
let stream$ = interval(1000).pipe(
  take(5),
  publish(),
  refCount(), // important
);
setTimeout(() => {
  stream$.subscribe((data) => console.log(data));
}, 2000);
setTimeout(() => {
  stream$.subscribe((data) => console.log(data));
}, 5100);

Operators

// of
const stream$ = of(1, 2, 3, 4, 5);

// tap
tap((value) => console.log(value));

// filter
filter((value) => value % 2 == 0);

// creating observable in observable
const stream$ = of(1, 2, 3).pipe(
  // we need to make it 'flat' because 'map' returns array of observable
  flatMap((val) =>
    of(val)
      .pipe(ajax({ url: url + val }))
      .pipe(map((e) => e.response)),
  ),
);

// fetch
from(fetch('https://jsonplaceholder.typicode.com/posts/1/'))
  .pipe(flatMap((res) => from(res.json())))
  .subscribe(console.log);

// to promise
const promise = of(1, 2, 3).toPromise();

combination

// return the array of the latest values  (1,2,3) (1,2) => [3,2]
let stream$ = combineLatest(source1, source2);

// run each source sequentially  (1,2,3) (1,2) => (1,2,3,1,2)
let stream$ = concat(source1, source2);

// merge: run each source at a time  (1,2,3) (1,2) => (1,1,2,2,3)
let stream$ = merge(source1, source2);

// zip: joins values on column basis.  (1,2,3) (1,2) => ([1,1], [2,2])
let stream$ = zip(source1, source2);

math

max();
min();
sum();

time

// reating observer
interval(100); // each 100ms
timer(1000); // after 1sec
timer(5000, 1000); // wait 5sec, then each 1sec

// operator
delay(100); // delay 1sec
debounceTime(500); // wait 0.5s and pass to next. timer and values is throw away if new event comes.

grouping

// buffer (0,1,2,3,4,5,6,7,8,9) => ([0,1,2,3,4],[5,6,7,8,9])
let breakWhen$ = interval(500);
let stream$ = interval(100).pipe(buffer(breakWhen$));
stream$.subscribe((data) => console.log('values', data));

// buffertime
bufferTime(500);
// is equal to...
buffer(interval(500));

error handling

// retry
// retry all th stream from the beginning
// (1, 2) => (1,1,1,1,1,1,2(error))
let stream$ = of(1, 2).pipe(
  map((value) => {
    if (value > 1) throw 'error';
    return value;
  }),
  retry(5),
);

// retryWhen
retryWhen((stream) => stream.delay(200));

// catchError(エラーを正常系に変換して流す)
let error$ = throwError('crash').pipe(catchError((err) => of('patched', err)));
error$.subscribe(
  (data) => console.log(data),
  (err) => console.error(err),
  () => console.log('complete'),
);
// ('patched', 'crash', 'complete')　すべて正常系