🎲

【Dart】Streamを簡単に学ぶ

2020/12/11に公開

Flutter

Dart

tech

全てをカバーしているわけではありませんが、基本的な部分を簡単順に記載します。

【Dart】Streamの簡単な使い方

参考

公式サイト
- チュートリアル
  - Asynchronous programming: streams | Dart
  - Creating streams in Dart | Dart > Using a StreamController
- クラスの説明
  - dart:async library - Dart API
    - Stream class - dart:async library - Dart API
      - Streamを作成するのに便利なConstructorsや、Streamを利用する際に便利なPropertiesが記載
    - StreamController class - dart:async library - Dart API

はじめに

Streamとは
- ストリーム - Wikipedia
  - ストリーム（Stream）は原語の川、流れといった意味の派生的用法で用いられる。
Dartでもデータを流す川のイメージ
- 新しいデータが流れた際に受け取りたい人に自動で教えてくれる
  - 例：チャットの文字情報
    - A君のメッセージがStreamに流れ、B君側ではそれをすぐに見れる
実行環境
- DartPadやAndroid Studio等で実行
  - Dart SDK 2.10.4
Streamの説明の流れ

同期実行（Iterable<T>）と非同期実行（Stream<T>）の基礎
非同期実行（Stream<T>）のちょっとした応用
Stream<T>の便利なコンストラクタ
Streamを実際に扱うのに便利なコントローラ

説明として、チンチロリンを想定し、サイコロ3つを一つずつ投げるイメージで記載
- 同期
  - 自分で片手で一つ一つ投げ続ける
- 非同期
  - 他人と交互に投げるかのイメージ

同期（`Iterable<T>`）の基礎

目的
- 全て逐次的な処理の場合
ポイント
- Streamに変化を加える関数（chinchiroSync）
  - Iterable<int>を返す
  - sync*をつける
備考
- yieldは式を計算し、結果の値を渡す（複数回実行できるreturnのようなもの）
- DartPadでの実行では、import 'dart:io';を利用したsleep関数に対応していないため、以下のエラーとなる。
  
  Uncaught Error: Unsupported operation: ProcessUtils._sleep

import 'dart:io';
import 'dart:math';

Iterable<int> chinchiroSync() sync* {
  final random = Random();

  for (var i = 0; i < 3; ++i) {
    sleep(Duration(seconds: 1));
    yield random.nextInt(6) + 1;
  }
}

void main() {
  final stream = chinchiroSync();
  for (int i in stream) {
    print(i);
  }
}

実行結果

4
6
2

☝1秒経過毎に1行出力される

上記のmain関数内のfor文をforEach文にした際も同様

import 'dart:io';
import 'dart:math';

Iterable<int> chinchiroSync() sync* {
  final random = Random();

  for (var i = 0; i < 3; ++i) {
    sleep(Duration(seconds: 1));
    yield random.nextInt(6) + 1;
  }
}

void main() {
  final stream = chinchiroSync();
  stream.forEach((int i) => print(i));
}

実行結果

5
4
6

非同期（`Stream<T>`）の基礎

目的
- 何かデータを外部から取得する等の処理をしたい場合
ポイント
- Streamに変化を加える関数（chinchiroAsync）
  - Stream<int>を返す
  - async*をつける
- main関数
  - main関数自体にasyncをつけ、値を待つためにawaitを利用する
    - awaitが無い場合は、for(int i in stream)のstream変数に以下エラー
      - The type 'Stream<int>' used in the 'for' loop must implement Iterable

import 'dart:math';

Stream<int> chinchiroAsync() async* {
  final random = Random();

  for (var i = 0; i < 3; ++i) {
    await Future.delayed(Duration(seconds: 1));
    yield random.nextInt(6) + 1;
  }
}

void main() async {
  final stream = chinchiroAsync();
  await for(int i in stream){
    print(i);
  }
}

実行結果

3
3
5

上記のmain関数内のfor文をforEach文にした際は、awaitが無くても問題なく動作
- 以下どちらでも結果は同様※何か違いがあるかもしれない
  - await stream.forEach((int i) => print(i));
  - stream.forEach((int i) => print(i));

import 'dart:math';

Stream<int> chinchiroAsync() async* {
  final random = Random();

  for (var i = 0; i < 3; ++i) {
    await Future.delayed(Duration(seconds: 1));
    yield random.nextInt(6) + 1;
  }
}

void main() async {
  final stream = chinchiroAsync();
  stream.forEach((int i) => print(i));
}

実行結果

3
3
3

非同期（`Stream<T>`）の応用

複数回連続でダイスロールを行う場合
- 「0」でダイスロールし、「-1」で終了の意味とする

import 'dart:math';

Stream<int> chinchiroAsync([int times = 3]) async* {
  final random = Random();
  int count = 0;

  while (true) {
    if (count == times) {
      yield -1; // = 'Finish!'
      break;
    }
    yield 0; // = 'Roll'
    for (var i = 0; i < 3; ++i) {
      await Future.delayed(Duration(seconds: 1));
      yield random.nextInt(6) + 1;
    }
    ++count;
  }
}

void main() async {
  final stream = chinchiroAsync();
  await for (int i in stream) {
    print(i);
  }
}

実行結果

Streamの関数を複数に分ける場合（他のStreamを呼び出す場合）
- yield*を用いて、Streamへの値渡しを一時停止することが可能
- また、分割することで、可読性向上

import 'dart:math';

Stream<int> chinchiroAsync([int times = 3]) async* {
  int count = 0;

  while (true) {
    if (count == times) {
      yield -1; // = 'Finish!'
      break;
    }
    yield 0; // = 'Roll'
    yield* diceRoll();
    ++count;
  }
}

Stream<int> diceRoll() async* {
  final random = Random();
  for (var i = 0; i < 3; ++i) {
    await Future.delayed(Duration(seconds: 1));
    yield random.nextInt(6) + 1;
  }
}

void main() async {
  final stream = chinchiroAsync();
  await for (int i in stream) {
    print(i);
  }
}

実行結果

非同期（`Stream<T>`）の便利なコンストラクタ

`Stream<T>.value`

引数とした一つの情報（'Chinchiro'）を渡す（yieldする）Streamを作成

void main() async {
  final stream = Stream<String>.value('Chinchiro');
  print(await stream.first);
}

実行結果

Chinchiro

Stream<String>.value('Chinchiro')を引数としたFuture関数を自作し、受け取ったデータを処理

Future<void> printThings(Stream<String> data) async {
  await for (var x in data) {
    print(x);
  }
}

void main() async {
  await printThings(Stream<String>.value('Chinchiro'));
  print('TEST');
}

実行結果

Chinchiro
TEST

awaitをprintThings関数呼び出し時に利用しないと、先に次のコードが実行される

Future<void> printThings(Stream<String> data) async {
  await for (var x in data) {
    print(x);
  }
}

void main() async {
  printThings(Stream<String>.value('Chinchiro'));
  print('TEST');
}

実行結果

TEST
Chinchiro

`Stream<T>.periodic`

1秒ごとにStreamに0から始まる値（period）を渡す

void main() async {
  final stream = Stream<int>.periodic(
    const Duration(seconds: 1),
    (int period) => period,
  );
  await stream.forEach((int i) => print(i));
}

実行結果

0
1
2
3
4
5
6
7
8
9
10
・
・
・
【省略】

1秒ごとにStreamに1から6までのランダム値を渡す
- ※この場合、period変数は特に利用していない

import 'dart:math';

void main() async {
  final random = Random();
  final stream = Stream<int>.periodic(
    const Duration(seconds: 1),
    (period) => random.nextInt(6) + 1,
  );
  await stream.forEach((int i) => print(i));
}

実行結果

1
4
4
1
2
1
6
・
・
・
【省略】

`Stream<T>.fromIterable`

リスト内の数値をStreamに渡す

import 'dart:math';

void main() async {
  int random() => Random().nextInt(6) + 1;
  final stream = Stream<int>.fromIterable(
    <int>[
      random(),
      random(),
      random(),
    ],
  );
  await stream.forEach((int i) => print(i));
}

実行結果

5
3
6

`Stream<T>.fromFuture`

FutureオブジェクトをStreamに渡す

import 'dart:math';

void main() async {
  String random() => (Random().nextInt(6) + 1).toString();
  final stream = Stream<String>.fromFuture(
      Future<String>.value(random() + random() + random()));
  print(await stream.first);
}

実行結果

以下、おまけ程度

Streamを扱うのに便利な`StreamController<T>`

実際にStreamを上手くに利用するためには、StreamControllerが便利
- final stream = chinchiroAsync();のようにStreamを手動で開始するより便利なのが、StreamController<T>でStreamを制御する方法
  - Streamをlistenさせることで、データを簡単に受け取れる
  - StreamControllerによって、必要な種類と数のStreamを分かりやすく扱える
サンプル①
- StreamController
  - プレイヤーが二人と想定し、二つ用意
- Input
  - サイコロの結果をaddにより、Streamにデータを渡す
- Output
  - listen関数を利用することで、Streamからデータを都度受け取れる

import 'dart:async';
import 'dart:math';

void main() {
  String random() => (Random().nextInt(6) + 1).toString();
  String diceData() => random() + random() + random();

  // StreamController
  final controllerPlayer1 = StreamController<String>();
  final controllerPlayer2 = StreamController<String>();

  // Input(add or sink.add)
  controllerPlayer1.add(diceData());
  controllerPlayer1.add(diceData());
  controllerPlayer2.add(diceData());
  controllerPlayer2.add(diceData());

  // Output(listen)
  controllerPlayer1.stream.listen((data) {
    print('Player1:Roll');
    print(data);
  });
  controllerPlayer2.stream.listen((data) {
    print('Player2:Roll');
    print(data);
  });
}

実行結果

Player1:Roll
264
Player2:Roll
511
Player1:Roll
456
Player2:Roll
216

サンプル②
- StreamController
  - StreamController<T>プロパティのstreamをreturnする関数を利用する場合
  - 【備考】StreamControllerのコンストラクタ
    - StreamController({void onListen(), void onPause(), void onResume(), FutureOr<void> onCancel(), bool sync: false})
- Input
  - サイコロの結果をaddにより、Streamにデータを渡す
    - streamをreturnする関数に与えた秒間隔毎に最大回数までadd
- Output
  - listen関数を利用することで、Streamからデータを都度受け取れる

import 'dart:async';
import 'dart:math';

Stream<String> chinchiro(Duration interval, [int times]) {
  String random() => (Random().nextInt(6) + 1).toString();
  String diceData() => random() + random() + random();

  // StreamController
  StreamController<String> controller;
  Timer timer;
  int count = 0;

  void tick(Timer timer) {
    count++;
    // Input(add or sink.add)
    controller.add(diceData());
    if (count == times) {
      timer.cancel();
      controller.close();
    }
  }

  void startTimer() {
    timer = Timer.periodic(interval, tick);
  }

  void stopTimer() {
    if (timer != null) {
      timer.cancel();
      timer = null;
    }
  }

  controller = StreamController<String>(
    onListen: startTimer,
    onPause: stopTimer,
    onResume: startTimer,
    onCancel: stopTimer,
  );

  return controller.stream;
}

void main() {
  final stream = chinchiro(const Duration(seconds: 1), 3);
  // Output(listen)
  stream.listen(print);
}

実行結果

243
211
145

サンプル③
- StreamController
  - StreamController<T>を持つ、クラスを利用する場合
  - controller.streamを渡せるようにgetter追加
- Input
  - サイコロの結果をaddにより、Streamにデータを渡す
    - １秒間隔毎に最大回数までadd
- Output
  - listen関数を利用することで、Streamからデータを都度受け取れる

import 'dart:async';
import 'dart:math';

class Chinchiro {
  final int times;
  String random() => (Random().nextInt(6) + 1).toString();
  String diceData() => random() + random() + random();

  Timer timer;
  int count;
  // StreamController
  StreamController<String> controller;

  Chinchiro({this.times = 3}) {
    count = 0;
    controller = StreamController<String>(
      onListen: startTimer,
      onPause: stopTimer,
      onResume: startTimer,
      onCancel: stopTimer,
    );
  }

  Stream<String> get stream => controller.stream;

  void tick(Timer timer) {
    count++;
    // Input(add or sink.add)
    controller.add(diceData());
    if (count == times) {
      timer.cancel();
      controller.close();
    }
  }

  void startTimer() {
    timer = Timer.periodic(const Duration(seconds: 1), tick);
    count = 0;
  }

  void stopTimer() {
    timer?.cancel();
    controller.close();
  }
}

void main() {
  final stream = Chinchiro().stream;
  // Output(listen)
  stream.listen(print);
}

実行結果

631
523
132

【Dart】Streamの簡単な使い方

参考

はじめに

同期（Iterable<T>）の基礎

非同期（Stream<T>）の基礎

非同期（Stream<T>）の応用

非同期（Stream<T>）の便利なコンストラクタ

Stream<T>.value

Stream<T>.periodic

Stream<T>.fromIterable

Stream<T>.fromFuture

Streamを扱うのに便利なStreamController<T>

Discussion

同期（`Iterable<T>`）の基礎

非同期（`Stream<T>`）の基礎

非同期（`Stream<T>`）の応用

非同期（`Stream<T>`）の便利なコンストラクタ

`Stream<T>.value`

`Stream<T>.periodic`

`Stream<T>.fromIterable`

`Stream<T>.fromFuture`

Streamを扱うのに便利な`StreamController<T>`