ข้อมูล Stream การสร้าง และใช้งาน Stream ในภาษา Dart เบื้องต้น

Stream<String> genStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    yield 'This is Line $i\n';
  }
}

This is Line 1\n
This is Line 2\n
This is Line 3\n
This is Line 4\n
This is Line 5\n

/// แยกข้อมูล stream ของข้อความ ออกเป็นบรรทัดๆ
/// ข้อมูลจาก Stream Source ถูกแบ่งเป็นส่วนๆ ย่อย
Stream<String> lines(Stream<String> source) async* {
  // สร้างตัวแปร เก็บข้อมูลบรรทัด บางส่วน จากส่วนของข้อมูล
  var partial = '';
  // รอข้อมูลแต่ละส่วน ส่วนไหนพร้อม ก็นำไปใช้งานต่อ
  await for (var chunk in source) { // วนลูปใช้งาน ส่วนที่พร้อม
    var lines = chunk.split('\n'); // แยกเป็นข้อมูล array จากการขึ้นบรรทัดใหม่
    // ถ้าข้อมูลมี \n จะได้ lines เป็น ['This is Line 1','']
    // แต่ถ้าไม่มี \n จะได้ lines เป็น ['This is Line 1']
    lines[0] = partial + lines[0]; // เก็บข้อมูลบรรทัดแรก
    partial = lines.removeLast(); // เก็บค่าบางส่วน
    // ถ้าข้อมูลมี \n partial จะเก็บค่า lines[1] ซึ่งเป็นค่า ว่าง
    // แต่ถ้าไม่มี \n partial จะเก็บค่าข้อความ lines[0]
    for (var line in lines) { // วนลูปส่งออกข้อมูล stream จากในลูป
      yield line; // ส่งออก output stream คล้ายคำสั่ง return
    }
  }
  // กรณีไม่มี \n จะทำให้ไม่มีการคืนค่าด้วย yield จากในลูป เพราะ lines ไม่มีข้อมูล
  // ข้อมูลถูกเก็บไว้ใน partial เราก็เช็คค่า partial ถ้ามีก็ให้ส่งออกไป เป็นข้อมูลทั้งหมด
  // จากค่า partial แทน
  if (partial.isNotEmpty) yield partial;
}

Future<String> concatStream(Stream<String> stream) async {
  var concatStr = '';
  await for (var value in stream) {
    concatStr += "\"$value\" |";
  }
  return concatStr;
}

"This is Line 1" |"This is Line 2" |"This is Line 3" |"This is Line 4" |"This is Line 5" |

"This is Line 1This is Line 2This is Line 3This is Line 4This is Line 5" |

void main() async {
  // สร้างข้อมูล Stream โดยใช้ async* ฟังก์ชั่น
  var stream = genStream(5);
  // สร้างข้อมูล Stream ดยใช้ Stream เดิม
  var newStream = lines(stream); 
  // เรียกใช้งาน Stream
  var output = await concatStream(newStream);
  print(output);
}

Stream<String> genStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    yield 'This is Line $i\n';
  }
}

Stream<String> lines(Stream<String> source) async* {
  var partial = '';
  await for (var chunk in source) { 
    var lines = chunk.split('\n'); 
    lines[0] = partial + lines[0]; 
    partial = lines.removeLast(); 
    for (var line in lines) {  
      yield line; 
    }
  }
  if (partial.isNotEmpty) yield partial;
}

Future<String> concatStream(Stream<String> stream) async {
  var concatStr = '';
  await for (var value in stream) {
    concatStr += "\"$value\" |";
  }
  return concatStr;
}

var counterStream =
    Stream<int>.periodic(Duration(seconds: 1), (x) => x).take(15);

void main() async {
  
  var counterStream =
    Stream<int>.periodic(Duration(seconds: 1), (x) => x).take(15);  
  await for(var data in counterStream){
    print(data);
  }

}

void main() async {
  
  var counterStream =
    Stream<int>.periodic(Duration(seconds: 1), (x) => x).take(15);  
  counterStream.forEach(print);

}

void main() async {
  
  var counterStream =
    Stream<int>.periodic(Duration(seconds: 1), (x) => x).take(15);  
  counterStream = counterStream.map((x) => x*2 );
  counterStream.forEach(print); // ค่าจะแสดงทุก 1 วินาทีเป้น 0.. 2.. 4 ...จะถึง 28
  
}

void main() async {
  
  var counterStream =
    Stream<int>.periodic(Duration(seconds: 1), (x) => x).take(15);  
  counterStream = counterStream
    .where((int x) => x.isEven) // เช็คที่ event ค่า x เป็นเลขคู่
    .expand((var x) => [x, x]) // เพิ่มหรือทำซ้ำ event นี้
    .take(5); // หยุดที่ 5 รายการแรก
  counterStream.forEach(print); ค่าจแสดง แค่ 5 รายการ ใน 3 วินาที เพราะ ที่วินาทีแรก จะแสดง
    // 0 และ 0 นับเป็น 2 events ที่แสดงซ้ำเพราะเราใช้ expand() และวินาทีที่สองแสดงเป็น 2 และ 2
    // วินนาทีที่ 3 แสดง 4 แค่ค่าเดียว เพราะ ครบ 5 จากคำสั่ง take(5)
  
}

void main() {

  // สร้าง Stream ตัวเลข
 var stream = timedCounter(Duration(seconds:2),5);
  stream.listen(print); // เรียกใช้งาน
  // จะแสดงตัวเลข 0..1..2..3 และ 4 แต่ละค่าใน ทุกๆ 2 วินาที
  // จนครบตามจำนวน ที่กำหนดใน maxCount ถ้ามีกำหนด
  
}

Stream<int> timedCounter(Duration interval, [int maxCount]) async* {
  int i = 0;
  while (true) { // สร้างลูปแบบ infinite
    // รอข้อมูลโดยให้ delay ตามเวลาที่กำหนด ในที่นี้คือ 2 วินาที
    await Future.delayed(interval);
    yield i++; // เพิ่มค่า i และส่งออกเป็น output stream
    if (i == maxCount) break; // ออกจากลูปถ้าเข้าเงื่อนไข
  }
}

Stream<T> streamFromFutures<T>(Iterable<Future<T>> futures) async* {
  for (var future in futures) { // วนลูป futures  
    var result = await future; // รอผลลัพธ์จาก future
    yield result;  // ส่งออกเป็น Stream 
  }
}

import 'dart:async';

void main() async {
  
  // สร้าง Steam โดยใช้ StreamController
  var stream = timedCounter(Duration(seconds:1),10);
  // เราจะหน่วงเวลา 5 วินาที ก่อนเรียกใช้งาน เพื่อดูว่า Stream ทำงานทันที
  // ทั้งที่ยังไม่ได้เรียกใช้งาน
  await Future.delayed(const Duration(seconds: 5));
  stream.listen(print); 

}

// NOTE: รูปแบบการใช้งานนี้ยังไม่สมบูรณ์
// ทำงานทันที ก่อนเรียกใช้งาน และยังไม่มีกำหนด กรณีทำการหยุดชั่วคราว 
Stream<int> timedCounter(Duration interval, [int maxCount]) {
  var controller = StreamController<int>(); // สร้างตัวแปรเก็บ StreamController
  // หรือใช้เป็น StreamController<int> controller; ก็ได้
  int counter = 0;
  
  void tick(Timer timer) {
    counter++; // บวกค่าเพิ่ม
    controller.add(counter); // ส่งข้อมูล stream ออกเป็น events
    if (maxCount != null && counter >= maxCount) { // ถึงค่าที่กำหนด ถ้ามี
      timer.cancel(); // ยกเลิการทำงานของเวลา
      controller.close(); // ปิดการใช้งานข้อมูล Stream
    }
  }

  Timer.periodic(interval, tick); // BAD: ทำงานก่อนที่จะเรียกใช้งาน
  return controller.stream;
}

import 'dart:async';

void main() async {
  
  // สร้าง Steam โดยใช้ StreamController
  var stream = timedCounter(Duration(seconds:1),10);
  StreamSubscription<int> subscription; // กำหนดตัวเรียกใช้งาน
  
  // เราจะหน่วงเวลา 3 วินาที ก่อนเรียกใช้งาน 
  await Future.delayed(const Duration(seconds: 3));
  
  // เรียกใช้ง่านโดยใช้ subscription เพิ่อที่เราะสามารถ หยุดชั่วคราวได้ หรือใช้คำสั่งอื่นๆ
  // จัดการกับข้อมูล Stream ได้
  subscription = stream.listen((int counter) {
    print(counter); // แสดงข้อมูล ทุกๆ 1 วินาที
    if (counter == 5) { // ถ้าข้อมูลเท่ากับ 5
      // จำลองถ้าข้อมูลเท่ากับ 5 เราจะหยุดชั่วคราว 2 วินาที แล้วค่อยทำงานต่อ
      subscription.pause(Future.delayed(const Duration(seconds: 2)));
    }
  });

}


Stream<int> timedCounter(Duration interval, [int maxCount]) {
  // กำหนด property หรือตัวแปรสำหรับใช้งาน
  // ข้อมูล StreamController และ Timer อยู่ใน async* library 
  StreamController<int> controller;
  Timer timer;
  int counter = 0;

  // ส่วนของคำสั่งกำหนดการทำงาน
  void tick(_) {
    counter++;
    controller.add(counter); // Ask stream to send counter values as event.
    if (counter == maxCount) {
      timer.cancel();
      controller.close(); // Ask stream to shut down and tell listeners.
    }
  }

  // ส่วนของคำสั่ง เริ่มทำงาน
  void startTimer() {
    // เริ่มเวลา และเรียกใช้งาน ข้อมูล Stream จากคำสั่ง tick()
    timer = Timer.periodic(interval, tick);
  }

  // ส่วนของคำสั่ง หยุดการทำงาน
  void stopTimer() {
    if (timer != null) { // เช็คว่าเวลายังเดินอยู่ไหม ถ้ายังเดินอยู่ ให้ยกเลิก
      timer.cancel(); // ยกเลิกเวลา
      timer = null; // กำหนดเป็น null
    }
  }

  // ส่วนของการควบคุมการทำงาน
  controller = StreamController<int>(
      onListen: startTimer, // เมื่อเรีียกใช้งาน
      onPause: stopTimer, // เมื่อหยุดชั่วคราว
      onResume: startTimer, // เมื่อกลับมาเริ่มต่อ
      onCancel: stopTimer); // เมื่อยกเลิก ไม่ใช้งานแล้ว

  // return ข้อมูล Stream ที่สามารถควบคุมการทำงานได้
  return controller.stream;
}

    if( counter == 9){
      subscription.cancel();
    }

void main() async { 
  var stream = genStream(5); // สร้างข้อมูล Stream
  var sum = await sumStream(stream); // ใช้งานข้อมูล Stream
  print(sum); //output: 15
}

// ฟังก์ชั่น จำลองสร้างข้อมูล Stream
Stream<int> genStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    yield i;
  }
}

// ฟังก์ชั่น ใช้งานข้อมูล Events ของ Stream โดยหาค่าผลบวก
Future<int> sumStream(Stream<int> stream) async {
  var sum = 0;
  await for (var value in stream) {
    sum += value;
  }
  return sum;
}

void main() async { 
  var stream = genStream(5); // สร้างข้อมูล Stream
  var sum = await sumStream(stream); // ใช้งานข้อมูล Stream
  print(sum); //output: -1
}

// ฟังก์ชั่น จำลองสร้างข้อมูล Stream
Stream<int> genStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    if (i == 4) { // จำลอง Error ข้อมูลที่ 4
      throw new Exception('Intentional exception');
    } else {
      yield i;
    }
  }
}

// ฟังก์ชั่น ใช้งานข้อมูล Events ของ Stream โดยหาค่าผลบวก
Future<int> sumStream(Stream<int> stream) async {
  var sum = 0;
  try {
    await for (var value in stream) {
      sum += value;
    }
  } catch (e) {
    return -1;
  }
  return sum;
}

void main() async { 
  var stream = genStream(5); // สร้างข้อมูล Stream
  var sum = await lastPositive(stream); // ใช้งานข้อมูล Stream
  print(sum); //output: 5
}

// ฟังก์ชั่น จำลองสร้างข้อมูล Stream
Stream<int> genStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    yield i;
  }
}

// รูปแบบ Fat Arrow
Future<int> lastPositive(Stream<int> stream) =>
    stream.lastWhere((x) => x >= 0);

// รูปแบบเต็ม
// Future<int> lastPositive(Stream<int> stream) {
//   return stream.lastWhere((x){
//     return x>=0;
//   });
// }

Future<T> get first;
Future<bool> get isEmpty;
Future<T> get last;
Future<int> get length;
Future<T> get single;
Future<bool> any(bool Function(T element) test);
Future<bool> contains(Object needle);
Future<E> drain<E>([E futureValue]);
Future<T> elementAt(int index);
Future<bool> every(bool Function(T element) test);
Future<T> firstWhere(bool Function(T element) test, {T Function() orElse});
Future<S> fold<S>(S initialValue, S Function(S previous, T element) combine);
Future forEach(void Function(T element) action);
Future<String> join([String separator = ""]);
Future<T> lastWhere(bool Function(T element) test, {T Function() orElse});
Future pipe(StreamConsumer<T> streamConsumer);
Future<T> reduce(T Function(T previous, T element) combine);
Future<T> singleWhere(bool Function(T element) test, {T Function() orElse});
Future<List<T>> toList();
Future<Set<T>> toSet();

// สร้างฟังก์ชั่น หาข้อมูลที่ต้องการใน Stream
Future<bool> contains(Object needle) async {
  await for (var event in this) {
    if (event == needle) return true;
  }
  return false;
}

// สร้างฟังก์ชั่น ทำคำสั่งจัดการข้อมูล Stream อีกที
Future forEach(void Function(T element) action) async {
  await for (var event in this) {
    action(event);
  }
}

// สร้างฟังก์ชั่นแปลงชนิดข้อมูล
Future<List<T>> toList() async {
  final result = <T>[];
  await this.forEach(result.add);
  return result;
}

// สร้างฟังก์ชั่น เชื่อมข้อมูล
Future<String> join([String separator = ""]) async =>
    (await this.toList()).join(separator);

Stream<R> cast<R>();
Stream<S> expand<S>(Iterable<S> Function(T element) convert);
Stream<S> map<S>(S Function(T event) convert);
Stream<T> skip(int count);
Stream<T> skipWhile(bool Function(T element) test);
Stream<T> take(int count);
Stream<T> takeWhile(bool Function(T element) test);
Stream<T> where(bool Function(T event) test);

Stream<E> asyncExpand<E>(Stream<E> Function(T event) convert);
Stream<E> asyncMap<E>(FutureOr<E> Function(T event) convert);
Stream<T> distinct([bool Function(T previous, T next) equals]);

Stream<T> handleError(Function onError, {bool test(error)});
Stream<T> timeout(Duration timeLimit,
    {void Function(EventSink<T> sink) onTimeout});
Stream<S> transform<S>(StreamTransformer<T, S> streamTransformer);

// ฟังก์ชั่น ใช้งานข้อมูล Events ของ Stream โดยหาค่าผลบวก
Future<int> sumStream(Stream<int> stream) async {
  var sum = 0;
  var streamWithoutErrors = stream.handleError((e) => null);
  await for (var value in streamWithoutErrors) {
    sum += value;
  }
  return sum;
}

import 'dart:convert';
import 'dart:io';

Future<void> main(List<String> args) async {
  var file = File(args[0]);
  var lines = utf8.decoder
      .bind(file.openRead())
      .transform(LineSplitter());
  await for (var line in lines) {
    if (!line.startsWith('#')) print(line);
  }
}

StreamSubscription<T> listen(void Function(T event) onData,
    {Function onError, void Function() onDone, bool cancelOnError});