ムーディー・ブルースとは

ムーディー・ブルースについてピクシブ百科事典には

護衛チームの一員であるレオーネ・アバッキオのスタンド能力。
過去の出来事を再生（リプレイ）する能力を持つ。

このように書いてあります。

この 過去の出来事を再生（リプレイ）する というのがRxJSにぴったりなテーマである & 現在ジョジョ第5部のアニメが放送中なのでこれを書きました。

RxJS とは

知らない方のために誇張した表現で説明すると、Rxとは時空を操る概念です。RxJSはRxのJavaScript実装となります。

過去のイベントデータがあるとする

キーボード入力のイベントデータを持っているとします。

const events = [
  { ts: 100000, value: 'a' },
  { ts: 100100, value: 'b' },
  { ts: 101100, value: 'c' },
  { ts: 101120, value: 'd' },
  { ts: 101150, value: 'e' },
  { ts: 101250, value: 'f' },
  { ts: 101300, value: 'g' },
  { ts: 102000, value: 'h' },
  { ts: 102010, value: 'i' },
  { ts: 103010, value: 'j' }
];

このような感じです。tsはtimestampです。

timestampを100000と書いていますが、イメージとしてはDate.now()で得られる1540643168056のような値が本来なら入ると思ってください。

RxJSを使う

Node.js

npm install rxjs

const { of, from, timer } = require('rxjs');
const { scan, concatMap, delay, mapTo } = require('rxjs/operators');

Web

<script src="https://unpkg.com/rxjs@6.3.3/bundles/rxjs.umd.min.js"></script>

const { of, from, timer } = rxjs;
const { scan, concatMap, delay, mapTo } = rxjs.operators;

このように必要なものだけをデストラクチャリングで取り出すのが作法です。

リプレイする！

const head = (events.length > 0 && events[0]) || {};

from(events)
  .pipe(
    scan((p, event) => ({ ...event, diff: event.ts - p.ts }), head),
    concatMap(event => of(event).pipe(delay(event.diff)))
  )
  .subscribe({
    next: event => console.log(event),
    complete: () => console.log('complete')
  });

from ... events配列をストリームに変換する。
scan ... eventオブジェクトにdiffプロパティを追加している。前回の実行結果がpに入る。
concatMap ... Observableから新しいObservableを作る。flatMapに近いが、非同期処理の完了を待つという特性がある。
of ... 値をObservableに変換する。
delay ... 一定時間待ってからストリームを次に流す。
subscribe ... nextではストリームを流れてきた値を使って何かする。completeはストリームが完了したときに呼ばれる。

ちなみにconcatMapの行は下記のように書いても同じように動きます。

concatMap(event => timer(event.diff).pipe(mapTo(event)))

timer ... 一定時間待ってからストリームを次に流すObservableを作る。
mapTo ... 次に流す値を指定したもの差し替える。

実行結果

ただの配列だったeventsがRxJSを使うと時間の流れを持ったストリームとなり、コード上でも上から下に流れるように表現されます。

以上、『RxJSで実装するムーディー・ブルース』でした。

(JSBinにサンプルコードを書きました => https://jsbin.com/wilakaj/5/edit?html,js,console )