お久しぶりです。 pregum_foxです。
前回からほぼ1か月空いてしまいましたが、引き続き書いていこうと思います。
前回までの記事は以下をご覧ください。
- 【Dart】rxdartについて調べてみる 第1回目 -概要-【rxdart】 - 狐好きぷろぐらまー
- 【Dart】rxdartについて調べてみる 第2回目 -ObservableとSubject系クラス- 【rxdart】 - 狐好きぷろぐらまー
- 【Dart】rxdartについて調べてみる 第3回目 -concatとconcatEager- 【rxdart】 - 狐好きぷろぐらまー
- 【Dart】rxdartについて調べてみる 第4回目 -Defer, Merge- 【rxdart】 - 狐好きぷろぐらまー
- 【Dart】rxdartについて調べてみる 第5回目 -Periodic, Never- 【rxdart】 - 狐好きぷろぐらまー
- 【Dart】rxdartについて調べてみる 第6回目 -Race, Retry- 【rxdart】 - 狐好きぷろぐらまー
- 【Dart】rxdartについて調べてみる 第7回目 -RetryWhen, SequenceEqual- 【rxdart】 - 狐好きぷろぐらまー
- 【Dart】rxdartについて調べてみる 第8回目 -SwitchLatest, CombineLatest- 【rxdart】 - 狐好きぷろぐらまー
- 【Dart】rxdartについて調べてみる 第9回目 -ForkJoin, Zip- 【rxdart】 - 狐好きぷろぐらまー
目次です。
動作環境
動作環境は以下の通りです。
項目 | バージョン |
---|---|
Dart | 2.5.1 |
rxdart | 0.22.2 |
以下にForkJoinオペレータとZipオペレータについて記載します。 オペレータの名称は、rxdart、Stream APIの順で記載します。
今回からは変換系のオペレータのメモになります。
前回までのストリームに対してC#のLinqやJavaのStreamのようなメソッドチェインでつなげて書いていくことができます。
メソッドチェインで掛けることで、各処理の意図が明確になるというメリットがあります。
便利なのでなんでもチェインで書いてしまう弊害もありますが...
というわけで今回から変換系のオペレータを調べていきます。
Bufferオペレータ
- Rx: Buffer
- rxdart: buffer メソッド
- Stream API: BufferStreamTransformer<T>クラス
シグネチャ
rxdart
Observable<List<T>> buffer(Stream window)
Stream API
BufferStreamTransformer<T>(Stream window(T event))
処理の概要
Observableなストリームから発行されたアイテムを定期的にまとめ、まとまり事に発行するオペレータです。
とりあえずコードを以下に示します。
import 'package:rxdart/rxdart.dart'; // rxdart Future studyBuffer() async { Observable<int>.periodic(const Duration(milliseconds: 100), (i) { // 100msecごとに値を生成する print('periodic: $i -- ${DateTime.now()}'); return i; }) // 150msecごとに購読したストリームをまとめる (型はObservable<List<int>>) .buffer(Observable<int>.periodic(const Duration(milliseconds: 150), (j) { print('buffer: ${j * 150}msec -- ${DateTime.now()}'); // dummy data return j; })) // bufferでまとめたストリームを5要素取得する .take(5) // 購読する .listen((data) => print('listen: $data'), onDone: () => print('done. ')); // 区切るために少し待つ await Future.delayed(const Duration(milliseconds: 1000)); print('-----------------------------------------'); Observable<int>.periodic(const Duration(milliseconds: 100), (i) { print('periodic: $i -- ${DateTime.now()}'); return i; }) // 購読したストリームを8要素取得する .take(8) // 300msecごとにまとめる .buffer(Observable.periodic(const Duration(milliseconds: 300))) // List<int>を購読する .listen((data) => print('listen: $data'), onDone: () => print('done.')); } // 実行結果 // periodic: 0 -- 2019-11-30 21:33:11.732326 // periodic: 1 -- 2019-11-30 21:33:11.830984 // buffer: 0msec -- 2019-11-30 21:33:11.887837 // listen: [0, 1] // periodic: 2 -- 2019-11-30 21:33:11.930723 // periodic: 3 -- 2019-11-30 21:33:12.031452 // buffer: 150msec -- 2019-11-30 21:33:12.038436 // listen: [2, 3] // periodic: 4 -- 2019-11-30 21:33:12.131357 // buffer: 300msec -- 2019-11-30 21:33:12.187466 // listen: [4] // periodic: 5 -- 2019-11-30 21:33:12.230695 // periodic: 6 -- 2019-11-30 21:33:12.331394 // buffer: 450msec -- 2019-11-30 21:33:12.338383 // listen: [5, 6] // periodic: 7 -- 2019-11-30 21:33:12.431134 // buffer: 600msec -- 2019-11-30 21:33:12.488004 // listen: [7] // done. // ----------------------------------------- // periodic: 0 -- 2019-11-30 21:33:12.735647 // periodic: 1 -- 2019-11-30 21:33:12.834066 // periodic: 2 -- 2019-11-30 21:33:12.934208 // periodic: 3 -- 2019-11-30 21:33:13.034972 // listen: [0, 1, 2, 3] // periodic: 4 -- 2019-11-30 21:33:13.134700 // periodic: 5 -- 2019-11-30 21:33:13.234820 // periodic: 6 -- 2019-11-30 21:33:13.335909 // listen: [4, 5, 6] // periodic: 7 -- 2019-11-30 21:33:13.434182 // listen: [7] // done. // Stream API void studyBufferStream() async { // transform用に作成 BufferStreamTransformer<int> transform = BufferStreamTransformer<int>((window) { return Stream<int>.periodic(const Duration(milliseconds: 150), (j) { print('buffer: ${j * 150}msec -- ${DateTime.now()}'); // dummy data return j; }); }); // 100msecごとにストリームを生成する Stream<int>.periodic(const Duration(milliseconds: 100), (i) { print('periodic: $i -- ${DateTime.now()}'); return i; }) // 変換オペレータ(BufferStreamTransformer)で購読したストリームを .transform(transform) // bufferでまとめたストリームを5要素取得する .take(5) // List<int>を購読する .listen((data) => print('listen: $data'), onDone: () => print('done.')); // 区切る為に少し待つ await Future.delayed(const Duration(milliseconds: 1000)); print('-----------------------------------------'); Stream<int>.periodic(const Duration(milliseconds: 100), (i) { print('periodic: $i -- ${DateTime.now()}'); return i; }) // 購読したストリームを8要素取得する .take(8) // 300msecごとにまとめる .transform(BufferStreamTransformer<int>((window) { return Stream<int>.periodic(const Duration(milliseconds: 300)); })) // List<int>を購読する .listen((data) => print('listen: $data'), onDone: () => print('done.')); } // 実行結果 // periodic: 0 -- 2019-11-30 21:35:49.727264 // periodic: 1 -- 2019-11-30 21:35:49.825523 // buffer: 0msec -- 2019-11-30 21:35:49.884487 // listen: [0, 1] // periodic: 2 -- 2019-11-30 21:35:49.925599 // periodic: 3 -- 2019-11-30 21:35:50.026411 // buffer: 150msec -- 2019-11-30 21:35:50.034391 // listen: [2, 3] // periodic: 4 -- 2019-11-30 21:35:50.126242 // buffer: 300msec -- 2019-11-30 21:35:50.183365 // listen: [4] // periodic: 5 -- 2019-11-30 21:35:50.225443 // periodic: 6 -- 2019-11-30 21:35:50.326175 // buffer: 450msec -- 2019-11-30 21:35:50.334158 // listen: [5, 6] // periodic: 7 -- 2019-11-30 21:35:50.425243 // buffer: 600msec -- 2019-11-30 21:35:50.483406 // listen: [7] // done. // ----------------------------------------- // periodic: 0 -- 2019-11-30 21:35:50.728290 // periodic: 1 -- 2019-11-30 21:35:50.828272 // periodic: 2 -- 2019-11-30 21:35:50.928764 // periodic: 3 -- 2019-11-30 21:35:51.028473 // listen: [0, 1, 2, 3] // periodic: 4 -- 2019-11-30 21:35:51.128276 // periodic: 5 -- 2019-11-30 21:35:51.229244 // periodic: 6 -- 2019-11-30 21:35:51.328975 // listen: [4, 5, 6] // periodic: 7 -- 2019-11-30 21:35:51.429701 // listen: [7] // done.
サンプルの実行結果から、発行された値がbufferメソッドまたはBufferStreamTransformerクラス内に定義されている間隔ごとにまとめられていることが分かります。
Stream APIでも書けますが、変換系のオペレータはrxdartのObservable<T>のメソッドを使用した方が処理の意図が分かりやすいと思います。(buffer() vs transform())
一定間隔でまとめてデータの更新を行う時に便利だと思います。
ただ、今回の使い方だとrxdartのオペレータはbufferではなく、bufferTimeの方がより明確に意図が伝わると思います。
buffer系オペレータとwindow系オペレータは各目的に応じてオペレータが用意されています。
- 個数でまとめる: xxxCount
- buffer: bufferCount
- window: windowCount
- 時間でまとめる: xxxTime
- buffer: bufferTime
- window: windowTime
- 条件でまとめる: xxxTest
- buffer: bufferTest
- window: windowTest
今回はbufferとbufferCountオペレータのメモですので、次回にbufferTimeとbufferTestを書こうと思います。
BufferCountオペレータ
- Rx: なし(Bufferの派生オペレータ)
- rxdart: bufferCountメソッド
- Stream API: BufferCountStreamTransformer<T>クラス
シグネチャ
rxdart
Observable<List<T>> bufferCount(int count, [int startBufferEvery = 0])
Stream API
BufferCountStreamTransformer<T> BufferCountStreamTransformer(int count, [int startBufferEvery = 0])
処理の概要
Bufferオペレータのまとめる条件を個数に制限したオペレータです。
購読したストリームがcountに達した段階でまとめたストリームを発行します。
以下にコードを示します。
import 'package:rxdart/rxdart.dart'; Future studyBufferCount() async { // 1から4の値を発行するストリームを生成する Observable.range(1, 4) // 購読したストリームから2要素取得する .bufferCount(2) // 購読する .listen(print); // 区切る為に少し待つ await Future.delayed(const Duration(milliseconds: 500)); print('--------------------------------------------------'); // 1から10の値を発行するストリームを生成する Observable.range(1, 10) // 購読したストリームから5要素取得し、古いストリームから2要素だけ進めたストリームを生成する .bufferCount(5, 2) // 購読する .listen(print); // 区切る為に少し待つ await Future.delayed(const Duration(milliseconds: 500)); print('--------------------------------------------------'); // 1から10の値を発行するストリームを生成する Observable.range(1, 10) // 購読したストリームから2要素取得し、古いストリームから5要素進めたストリームを生成する .bufferCount(2, 5) // 購読する .listen(print); } // 実行結果 // [1, 2] // [3, 4] // -------------------------------------------------- // [1, 2, 3, 4, 5] // [3, 4, 5, 6, 7] // [5, 6, 7, 8, 9] // [7, 8, 9, 10] // -------------------------------------------------- // [1, 2] // [6, 7] Future studyBufferCountStreamTransformer() async { // 1から4の値を発行するストリームを生成する RangeStream(1, 4) // 購読したストリームから2要素取得する .transform(BufferCountStreamTransformer(2)) // 購読する .listen(print); // 区切る為に少し待つ await Future.delayed(const Duration(milliseconds: 500)); print('--------------------------------------------------'); // 1から10の値を発行するストリームを生成する RangeStream(1,10) // 購読したストリームから5要素取得し、古いストリームから2要素だけ進めたストリームを生成する .transform(BufferCountStreamTransformer(5, 2)) // 購読する .listen(print); // 区切る為に少し待つ await Future.delayed(const Duration(milliseconds: 500)); print('--------------------------------------------------'); // 1から10の値を発行するストリームを生成する RangeStream(1, 10) // 購読したストリームから2要素取得し、古いストリームから5要素進めたストリームを生成する .transform(BufferCountStreamTransformer(2, 5)) // 購読する .listen(print); } // 実行結果 // [1, 2] // [3, 4] // -------------------------------------------------- // [1, 2, 3, 4, 5] // [3, 4, 5, 6, 7] // [5, 6, 7, 8, 9] // [7, 8, 9, 10] // -------------------------------------------------- // [1, 2] // [6, 7]
サンプルの実行結果から、値が個数によってまとめられていることが分かります。
また、2つ目の処理ブロックは5要素ずつまとめて2要素間隔で進めているので、最初の要素が1, 3, 5, 7と2要素ずつずれていっていることが分かります。
また要素が足りなくなったタイミングで購読が終了しています。
3つ目の処理ブロックは2要素ずつまとめて5要素間隔で進めている為、最初の要素が1, 6と5要素ずつずれていることが分かります。
雑感
生成系のオペレータ(RangeStream、Stream.periodic)はrxdartとStream API共に名前が違う程度でほぼ同じように書けていたので、特にハマらずに書けていましたが、変換系のオペレータだとStream APIでの書き方がObservableとの書き方が違う為、ハマりました。
書き心地できには、順番に書けてタイプ数もそこまで多くならないrxdartの方が書きやすいなと思いました。
ただ、dartは標準でstreamが用意されている為、rxdartが使用できない状況下でも使用するオペレータだけ自作すれば、ある程度は似たようなことができるのはすごいと思いました。
firebase系のプラグインとかも少し試したりしていましたのでまたそのあたり記事にできたらなと思います。
更新もなるべく頑張ります。
ご覧頂きありがとうございました。
参考サイト
- ReactiveX - Buffer operator
- https://pub.dev/documentation/rxdart/latest/rx/Observable/buffer.html
- BufferStreamTransformer class - rx library - Dart API
- https://pub.dev/documentation/rxdart/latest/rx/Observable/bufferCount.html
- https://pub.dev/documentation/rxdart/latest/rx/Observable/windowCount.html
- https://pub.dev/documentation/rxdart/latest/rx/Observable/bufferTime.html
- https://pub.dev/documentation/rxdart/latest/rx/Observable/windowTime.html
- https://pub.dev/documentation/rxdart/latest/rx/Observable/bufferTest.html
- https://pub.dev/documentation/rxdart/latest/rx/Observable/windowTest.html
- https://pub.dev/documentation/rxdart/latest/rx/Observable/bufferCount.html
- BufferCountStreamTransformer class - rx library - Dart API