Async in Dart (3) ตอบค่าเป็นกระแสข้อมูลด้วย Stream
Tags: beginners, programming, dart, flutter
Published: 2020-05-11 14:35:15

---
canonical_url: https://www.centrilliontech.co.th/blog/2811/async-in-dart-3-stream/
---

ในการเขียนโปรแกรมทั่วๆ ไปเรามักจะส่งข้อมูลกันในรูปแบบของ **`sync`** เช่น
```dart
int getNumber() {
    return 1;
}
```
การเรียกใช้งานก็ง่ายๆ แบบนี้
```dart
print(getNumber());
```
แต่ในบทก่อนๆ เราเคยสอนเรื่องของ Generator ไปแล้ว นั่นคือการที่ส่งข้อมูลแบบหลายๆ ตัวกลับมาในรูปแบบของ **`sync*`** 

*ทบทวนแบบเร็วๆ! คือแทนที่จะตอบข้อมูลกลับแค่ครั้งเดียวด้วย `return` ก็เปลี่ยนเป็น `yield` แทน (yield ตอบได้หลายครั้งมากกว่า return)*
```dart
Iterable<int> getNumbers() sync* {
    yield 1;
    yield 2;
    yield 3;
}
```
เวลารับข้อมูลจากฟังก์ชันพวกนี้ไปใช้ ก็เลยต้องใช้การวนลูปนั่นเอง
```dart
for(var num in getNumbers()) {
    print(num);
}
```
ถ้าสรุปเป็นรูป ก็จะได้แบบข้างล่างนี่
![](https://3.bp.blogspot.com/-aa6jHdeKwVw/XrRAwjUKbWI/AAAAAAAASU8/oH8GjOMcl4IEi6vwLtGhljolskntBKyewCNcBGAsYHQ/s1600/dart-async-3-01.png)

แต่ทั้ง 2 วิธีนี้มันเป็นการส่งข้อมูลแบบ Synchronous คือโค้ดยังทำงานเรียงๆ กันตามลำดับ และโค้ดทั้งหมดจะไม่ถูกแตกเป็นส่วนๆ เข้า Event Loop แบบ Asynchronous

ดังนั้นต่อไป... เรามาดู 2 กรณีนี้ **แต่อยู่ในรูปของ Asynchronous กันบ้างดีกว่า!**

![](https://1.bp.blogspot.com/-4rmEzIxuXDo/XrRAwhuRHmI/AAAAAAAASVA/XUtkxsfxfxAyjzvF1hm5dz7gzkii4auPwCNcBGAsYHQ/s1600/dart-async-3-02.png)
ในบทที่แล้วเราพูดถึง `Future` กันไปแล้ว ซึ่งมันคือการเปลี่ยนจาก **`sync`** ให้กลายเป็น **`async`** นั่นเอง 

แปลว่าเรายังขาดอยู่หนึ่งตัว คือ **`async*`** นั่นเอง

| | Single Value | Zero or more Values |
|---|---|---|
| Sync | `int` | `Iterable<int>` |
| Async | `Future<int>` | `Stream<int>` |

> สรุปง่ายๆ คือ `Stream` ก็คือ `Iterable` ที่อยู่ในรูปของ Asynchronous นั่นเอง

## วิธีการดึงค่าออกมาจาก Stream
การใช้งาน Stream ก็คล้ายๆ กับ Future นั่นแหละ แต่มีรายละเอียดเยอะกว่า เพราะมันตอบได้หลายค่ามากกว่า Future ยังไงล่ะ

ขอสมมุติว่าเรามีฟังก์ชันหนึ่ง ที่ตอบค่าเป็น `Stream<int>` ชื่อ **`getNumberStream()`** นะ

### รอฟังค่าเรื่อยๆ ด้วย listen
สำหรับ Future เราจะใช้คำสั่ง `then()` ในการรอค่า แต่ Stream จะใช้ `listen()` แทน
```dart
Stream<int> numberStream = getNumberStream();

var subscription = numberStream.listen((num){
    print('Receive: $num');
});
```
*`listen()` จะตอบกลับเป็น subscription ซึ่งเดี๋ยวจะพูดถึงอีกที*
แน่นอนว่าเวลาเอาไปรัน ผลที่ได้อาจจะเป็นแบบนี้
```
output:
Receive: 1
Receive: 2
Receive: 3
Receive: ...
```
ก็คือเราได้รับตัวเลขหลายๆ ตัวนั่นเอง **แต่อาจจะได้ตอนไหนก็ไม่รู้** แบบนี้
```
time            │     │    │
0|             1sec.  │    │
1| Receive: 1  ─┘    2sec. │
2| Receive: 2  ───────┘    │
3|                        4sec.
4| Receive: 3  ────────────┘
5|
```
สำหรับตัว `listen()` นั้นมี options เสริมให้เราใช้งานได้หลายตัว คือ
- `onError`: เมื่อเกิด error ให้ทำอะไร
- `onDone`: เมื่อได้รับข้อมูลครบทุกตัวแล้วให้ทำอะไร (ทำงานเมื่อ Stream ตอบข้อมูลครบหมดทุกตัวแล้ว มันจะเรียก onDone ให้ทำงาน)
- `cancelOnError`: เป็นตัวกำหนดว่าถ้าเกิด error ขึ้นแล้ว (ในกรณีที่ข้อมูลยังส่งกลับมาไม่ครบทุกตัว) จะเลือกที่จะหยุดการทำงานของ Stream ไปเลย หรือจะยังให้ทำงานต่อก็ได้
```dart
var subscription = numberStream.listen(
    (num){
        print('ได้รับข้อมูล $num');
    }, 
    onError: (err){
        print('มีข้อผิดพลาดเกิดขึ้นล่ะ $err');
    },
    onDone: (){
        print('ได้รับข้อมูลครบแล้วจ้า');
    },
    cancelOnError: false,
);
```
### ลดรูป Stream ด้วย await
เช่นเดียวกับตอน Future นะ คือเราสามารถเปลี่ยน `listen()` ให้ไปอยู่ในรูปแบบการเขียนแบบ Synchronous ได้ แต่ก็จะไม่ได้ตรงๆ แบบ Future นะเพราะเราต้องรับข้อมูลด้วยลูป
```dart
Stream<int> numberStream = getNumberStream();

await for(var number in numberStream) {
    print('Receive: $number');
}
```
## Broadcast Stream
ตามปกติแล้วการใช้งาน Stream จะมีการ `listen()` ได้แค่ครั้งเดียวเท่านั้น 

เมื่อเรา subscription ไปครั้งหนึ่งแล้ว ถ้าจะ subscription ซึ่งกับ Stream ตัวเดิมมันจะเกิด **Exception** ขึ้นนะ!!
```dart
Stream<int> numberStream = getNumberStream();

var subscription1 = numberStream.listen((num){
    print('Receive: $num');
});

//second subscribe -> Exception!!
var subscription2 = numberStream.listen((num){
    print('Receive: $num');
});
```

วิธีการแก้ (ถ้าอยาก subscription หลายครั้งจริงๆ) ให้เปลี่ยน Stream ตัวนั้นให้เป็นสิ่งที่เรียกว่า **"Broadcast Stream"** แทน

วิธีเปลี่ยนก็ง่ายๆ คือใช้ get property ที่ชื่อว่า `asBroadcastStream` แบบนี้
```dart
Stream<int> numberStream = getNumberStream().asBroadcastStream;

var subscription1 = numberStream.listen((num){
    print('First receive:  $num');
});

var subscription2 = numberStream.listen((num){
    print('Second receive: $num');
});
```
ทีนี้ ถ้า Stream ของเรามีการส่งตัวเลข [**1** เมื่อผ่านไป1วินาที] และ [**3** เมื่อผ่านไป3วินาที] ก็จะได้ผลลัพธ์แบบนี้ (สังเกตดูว่าเราได้ข้อมูลตัวละ 2 ครั้ง เพราะมี subscription 2 ตัวนั่นเอง)
```
time                   │     │
0|                   1sec.   │
1| First receive:  1  ─┤     │
 | First receive:  1  ─┘    3sec.
2|                           │
3| Second receive: 3  ───────┤
 | Second receive: 3  ───────┘
4|
```





## Subscription
กลับมาที่สิ่งที่เราพูดค้างไว้ นั่นคือตอบที่เราสั่ง `listen()` เราจะได้สิ่งที่เรียกว่า `subscription` กลับมา

เพราะ Stream คือ **"กระแสข้อมูล"** ที่ไหลมาเรื่อยๆ มาตอนไหนก็ไม่รู้ เลยมี subscription เอาไว้ควบคุมกระแสนั้นอีกที
```dart
//หยุดการรับข้อมูล
subscription.pause();

//กลับรับข้อมูลใหม่
subscription.resume();

//แคนเซิล Stream ตัวนั้น
subscription.cancel();
```
แต่มีข้อควรระวังอย่างนึง คือการ `pause()` ไม่ใช้การไม่รับข้อมูลในจังหวะนั้น แต่เป็นการห**ยุดรับข้อมูลชั่วคราว**เท่านั้น แปลว่า **"ข้อมูลไม่ได้หายไปนะ มันแค่เข้าคิวรอเรา `resume()` อีกทีนั่นเอง!"**

ลองดูตัวอย่างข้างล่างประกอบ ขอสมมุติว่าเรามี stream อยู่หนึ่งตัวที่จะส่งตัวเลขกลับมาเรื่อยๆ ทุกๆ 1 วินาที [1, 2, 3, 4, 5, ...] แบบนี้นะ
```dart
var subscription = stream.listen((x){
    print('Receive: $x');
});
Future.delayed(Duration(seconds: 3), (){
    subscription.pause();
});
Future.delayed(Duration(seconds: 6), (){
    subscription.resume();
});
Future.delayed(Duration(seconds: 8), (){
    subscription.cancel();
});
```
แล้วเราก็ตั้งค่า (โดยใช้ Future.delayed) ให้มัน..
- pause: เมื่อผ่านไป 3 วินาที
- resume: เมื่อผ่านไป 6 วินาที (จากตอนเริ่มโปรแกรม)
- cancel: เมื่อผ่านไป 8 วินาที (จากตอนเริ่มโปรแกรม)

เราอาจจะคิดว่าระหว่างวินาทีที่ 3-6 เราจะไม่ได้รับข้อมูลช่วงนั้น (คือข้อมูล [4,5] หายไปเลย)

**แต่ไม่ใช่นะ!** เพราะเลข [4,5] นั้นยังอยู่นะ แค่มันเข้าคิวรอที่จะออกมาอยู่

เมื่อเรา resume ในวินาทีที่ 6 มันก็จะออกมาทีเดียวหมดเลย [4,5,6]

ดูอธิบายด้วย timeline ข้างล่างนี่น่าจะเข้าใจมากกว่า
```
time  
1| Receive: 1
2| Receive: 2
3| Receive: 3
4|            ├─ waiting until (6)
5|            │
6| Receive: 4 ┐
 | Receive: 5 ├─ 3 values in 1 sec.
 | Receive: 6 ┘
7| Receive: 7
8| Receive: 8
9|
```











## มาลองสร้าง Stream กันบ้าง

การสร้าง Stream ทำได้หลายวิธีมากๆ แต่แบบง่ายที่สุดคือใช้วิธีแบบ Generator


## แปล Iterable เป็น Stream
เราเคยสอนการสร้าง Generator ไปแล้วในบท [Dart 105: มันวนซ้ำได้นะ! มาสร้าง Generator และใช้งาน Iterable กันเถอะ](https://dev.to/centrilliontech/dart-106-generator-iterable-2oof) แต่ตอนนั้นเราเขียนมันในรูปแบบของ sync 

อย่างที่พูดไปตอนต้นว่า `Stream` ก็คือ `Iterable` ที่อยู่ในรูปของ Asynchronous เราจะมาแสดงให้เห็นกัน

ขอเปิดด้วยโค้ดแบบ Iterable ในรูปของ Synchronous

โจทย์คือ... เราต้องการสร้างฟังก์ชันสำหรับดึง **Data** ออกมาจำนวนหนึ่งตั้งแต่ *from* ถึง *to* 
```dart
Data fetchOneData(int id){
    ...
}

Iterable<int> getData(int from, int to) sync* {
    var dataList = [];
    for(var i=from; i<=to; i++){
        dataList.add(fetchOneData(i))
    }
    return dataList;
}
```
*(ฟังก์ชัน `fetchOneData()` ทำอะไรไม่ต้องสนใจนะ ไม่ใช่ประเด็นของเรื่อนี้)*

แต่เพื่อ performance ที่ดี เลยเราไม่โหลดข้อมูลในครั้งเดียว แต่เลือกที่จะเขียนมันในรูปแบบ Generator แทน ก็จะได้โค้ดแบบนี้..
```dart
Data fetchOneData(int id){
    ...
}

Iterable<int> getData(int from, int to) sync* {
    for(var i=from; i<=to; i++){
        yield fetchOneData(i);
    }
}
```
เวลาเราจะใช้งาน ก็เอาไปวนลูปได้ธรรมดาๆ แบบนี้
```dart
var dataList = getData(1, 10);

for(var data in dataList) {
    print(data);
}
```
ขึ้นต่อไปคือ แล้วถ้าข้อมูลของเราไม่ได้มาแบบ sync ล่ะ?

ถ้าฟังก์ชัน `fetchOneData()` ใช้เวลาโหลดข้อมูลนานขึ้น เราคงต้องเปลี่ยนมันเป็น **Future** แบบนี้
```dart
Future<Data> fetchData(int id) {
    ...
}
```

เมื่อเป็นแบบนี้ เราก็มีปัญหาซะแล้ว!

เพราะฟังก์ชัน `getData()` ที่เป็น sync ไม่สามารถเรียก `fetchOneData()` ที่เป็น async ได้ล่ะ!!

วิธีการแก้ก็คือเปลี่ยนฟังก์ชัน  `fetchOneData()` ให้หลายเป็น async ยังไงล่ะ

**"ด้วยการเปลี่ยน Iterable ให้กลายเป็น Stream"**

สิ่งที่เราต้องทำ มีอยู่ 3 อย่าง
1. จากเดิมที่รีเทิร์นค่าเป็น `Iterable<Data>` เราต้องเปลี่ยนมันเป็น `Stream<Data>` แทน
2. จากเดิมที่ฟังก์ชันเป็นชนิด `sync*` เราต้องเปลี่ยนมันเป็น `async*`
3. จากเดิมที่เรา `yield` ค่าทันทีได้เลย แต่เมื่อมันเป็น Future แล้วเราก็ต้องสั่งให้มันรอด้วย `await`
```dart
Stream<Data> getData(int from, int to) async* {
    for(var id=from; id<=to; id++) {
        yield await fetchOneData(id);
    }
}

void main() {
    fetchAllData(1, 10).listen(print);
}
```
จะเห็นว่า Iterable นั้นแปลงเป็น Stream ได้ง่ายๆ เลย

ลองมาดูกันอีกตัวอย่างกัน

```dart
int sumIt(List<int> items) {
    int sum = 0;
    for(var item in items) {
        sum += item;
    }
    return sum;
}
```
เรามีฟังก์ชัน `sumIt()` สำหรับหาผลรวมทั้งหมดใน List แต่ถ้าลิสต์ตัวนี้ไม่ได้ค่ามาแบบ sync ล่ะ? จะทำยังไง?

คราวนี้ให้สังเกตว่า parameter นั้นรับ List (หรือก็คือ Iterable นั่นแหละนะ) เข้ามา แล้วมันรีเทิร์นค่าแบบสเกล่าธรรมดา (คือ `int` ธรรมดาๆ นี่แหละ)

สำหรับเคสนี้ สิ่งที่เราต้องทำก็คือ
1. เปลี่ยนค่าแบบสเกล่าธรรมดาให้กลายเป็น Future
2. เปลี่ยนให้ฟังก์ชันเป็น `async` (ระวัง! ไม่ได้เปลี่ยนเป็น `async*` นะ เพราะมันไม่ได้รีเทิร์น Stream)
3. จุดที่เป็นปัญหาคือ for loop ที่วนค่าตัว Stream .. ให้เราเติม `await` ลงไปข้างหน้าเป็นอันจบ
```dart
Future<int> sumIt(Stream<int> items) async {
    int sum = 0;

    await for(var item in items) {
        sum += item;
    }

    return sum;
}
```

เรื่องสุดท้ายคือในบางกรณี เราอาจจะมีฟังก์ชัน Stream ที่เรียกใช้งาน Stream อีกตัวก็เป็นได้นะ

```dart
Stream<Data> getFirstStream() async* {
    ...
}

Stream<Data> getSecondStream() async* {
    yield getFirstStream(); //Not Work!!
}
```
วิธีการนี้โค้ดอาจคอมไพล์ได้ แต่เราจะไม่ได้ข้อมูลอะไรเลย (ถ้าใครยังไม่เข้าใจว่าทำไมไม่เวิร์ค ลองกลับไปอ่านบทที่เราพูดถึง Generator อีกทีนะ)

วิธีการแก้ก็เหมือนกับฟังก์ชัน `sync*` นั่นแหละ คือเราจะต้องวนลูปทีละตัว
```dart
Stream<Data> getFirstStream() async* {
    ...
}

Stream<Data> getSecondStream() async* {
    await for(var s in getFirstStream()){
        tield s;
    }
}
```
หรือแบบง่ายกว่าคือการใช้ `yield*` ก็ได้
```dart
Stream<Data> getFirstStream() async* {
    ...
}

Stream<Data> getSecondStream() async* {
   yield* getFirstStream(); //Ok!!
}
```


---




นอกจากวิธีสร้างแบบ Generator ที่เราใช้แนวคิดเดียวกับ Generator ใน sync แล้วยังมีอีกวิธีหนึ่งที่ให้เราคุมการไหลของข้อมูลใน Stream ได้คล่องตัวขึ้น นั่นคือใช้ **StreamController** 

ซึ่งเราจะพูดถึงกันในบทต่อไปนะ (พร้อมกับเก็บตกเนื้อหาเกี่ยวกับ Stream กันอีกนิดหน่อยด้วย)