skipLast, takeLastオペレータはデータを流すタイミングをずらす

ちょっと面白かったのでメモ。

例えば複数のSubscriberを並行して処理したいなと思った時に、autoConnectで流すタイミングを合わせたとする。

Observable<Integer> foo = Observable.just(3, 4, 5).publish().autoConnect(2);

foo.subscribe(data -> System.out.println("A:" + String.valueOf(data)));
foo.subscribe(data -> System.out.println("B:" + String.valueOf(data)));

A:3
B:3
A:4
B:4
A:5
B:5

もちろん順番に値が流れることになる。

この時にskipLastで片方だけ流す量を調整してみる。

Observable<Integer> foo = Observable.just(3, 4, 5).publish().autoConnect(2);

foo.subscribe(data -> System.out.println("A:" + String.valueOf(data)));
// 5だけskipされる
foo.skipLast(1).subscribe(data -> System.out.println("B:" + String.valueOf(data)));

するとこんな感じになる。