Asynchronous Programming : Streams

Stream은 데이터의 비동기 시퀀스를 제공합니다.

데이터 시퀀스는 사용자가 발생한 이벤트와 파일을 읽은 데이터등을 포함합니다.

await for 나 listen() 을 사용하여스트림을 처리할 수있습니다.

Stream은 에러에 응답하는 방법을 제공합니다.

스트림의 종류는 단일 구독(single subscription) 혹은 브로드캐스트가 있습니다.

 

Dart의 비동기 프로그래밍은 Future 및 Stream 클래스로 특징 지워집니다.

 

Future는 즉시 완료되지 않는 계산을 나타냅니다.

일반함수가 결과를 반환한다면 비동기함수는 결과를 포함하는 Future를 반환합니다.

결과가 준비가되면 Future가 알려줄 것입니다.

 

스트림은 일련의 비동기 이벤트입니다. 비동기 Iterable과 유사합니다.

Iterable은 요청하고 다음 이벤트를 받는데, 스트림은 준비가되었을 때 이벤트가 있음을 알려줍니다.

 

 

 

스트림 이벤트 받기

 

스트림은 여러 가지 방법으로 작성 될 수 있지만 다른 기사의 주제이지만 모든 메소드를 동일한 방식으로 사용할 수 있습니다.

비동기 for루프(보통 await for 라고 부름)은 마치 for 루프가 Iterable을 반복하는 것 처럼 스트림 이벤트를 반복합니다.

예를 들어 아래코드의 sumStream함수는 단순히 int 이벤트 스트림의 각 이벤트를 받고 이것들의 합계를 반환합니다.

함수는 await for 루프를 사용할 때 필요한 async 키워드로 표시됩니다.

 

 

 

import 'dart:async';

Future<int> sumStream(Stream<int> stream) async {

var sum = 0;

await for (var value in stream) {

sum += value;

}

return sum;

}

Stream<int> countStream(int to) async* {

for (var i = 0; i <= to; i++) {

yield i;

}

}

main() async {

var stream = countStream(10);

var sum = await sumStream(stream);

print(sum);

}

 

 

스트림 변환

import 'dart:async';

void main(List<String> args){
  Stream<int> stream = countStream(10);

  stream
    .transform(transformer)
    .listen((value) { 
      print(value);
    });
}

var transformer = StreamTransformer<int, String>.fromHandlers(
  handleData:(value, sink){
    sink.add("현재 값은 $value");
  }
);


Stream<int> countStream(int to) async*{
  for(int i=0; i<=to; i++){
    await Future.delayed(const Duration(seconds: 1));
    yield i;
  }
}

 

 

 

에러발생

import 'dart:async';

void main(List<String> args){
  Stream<int> stream = countStream(10);

  stream
    .transform(transformer)
    .listen((value) { 
      print(value);
    });
}

var transformer = StreamTransformer<int, String>.fromHandlers(
  handleData:(value, sink){
    sink.add("현재 값은 $value");
  }
);


Stream<int> countStream(int to) async*{
  for(int i=0; i<=to; i++){
    await Future.delayed(const Duration(seconds: 1));

    if(i==5){                           //<< 추가한 부분
      throw new Exception("MyError");   //<< 추가한 부분
    }else{                              //<< 추가한 부분
      yield i;
    }
    
  }
}

출력결과

MyError가 출력되고 종료됨

현재 값은 0
현재 값은 1
현재 값은 2
현재 값은 3
현재 값은 4
Unhandled exception:
Exception: MyError

 

 

 

브로드캐스트스트림

보통 스트림은 하나의 리스너만 있고,

2개 이상이의 리스터가 필요하다면 브로드캐스트 스트림을 사용하여야한다.

 

import 'dart:async';

void main(List<String> args){
  Stream<int> stream = countStream(10);
  var broadcastSteam = stream.asBroadcastStream(); //추가

  broadcastSteam
    .transform(transformer)
    .listen((value) { 
      print(value);
    });

  broadcastSteam.listen((value) {   //추가
    print("두번째 리스너 $value");   //추가
  });                               //추가
}

var transformer = StreamTransformer<int, String>.fromHandlers(
  handleData:(value, sink){
    sink.add("현재 값은 $value");
  }
);


Stream<int> countStream(int to) async*{
  for(int i=0; i<=to; i++){
    await Future.delayed(const Duration(seconds: 1));
     yield i;
    
  }
}