[ Flutter ] 스트림

Stream이란?

💬  스트림은 데이터나 이벤트가 들어오는 통로다.

💬  앱을 만들다보면 데이터를 처리할 일이 많은데, 어느 타이밍에 데이터가 들어올 지 알기 어려운 것을 비동기 작업으로 처리할 때 스트림이 사용된다.

💬  간단한 예를 통해 스트림을 이해해보자.

import 'dart:async';

void main() {
  // 1초마다 데이터 1개를 최대 5개까지 만듦
  Stream stream = Stream.periodic(Duration(seconds: 1), (x) => x).take(5);
  // 이벤트 처리
  stream.listen(print);
}

◽  스트림은 한쪽 끝에서 데이터를 보내고 다른 쪽에서는 보낸 데이터를 받을 수 있는 파이프와 같다.

◽  스트림은 기본적으로 Single Subscription, 즉 하나의 스트림에는 하나의 리스너만 등록이 가능하다.

 

Stream 값 변경하기

import 'dart:async';

void main() {
  Stream stream = Stream.periodic(Duration(milliseconds: 200), (x) => x)
    .take(3)
    .map((x) => x + 10);
  
  stream.listen(print);
}

/* 
  10
  11
  12
*/

◽  간단한 처리는 map을 통해 가능하지만, 복잡한 처리는 StreamTransformer를 통해 가능하다.

import 'dart:async';

void main() {
  StreamTransformer transform = new StreamTransformer.fromHandlers(
    handleData : (data, sink) {
      sink.add('first $data');
      sink.add('second $data');
    },
  );
  
  Stream stream = new Stream.fromIterable(<dynamic>['hello', 1, 2]);
  stream.transform(transform).listen((event) => print('listen: $event'));
}

/*
  listen: first hello
  listen: second hello
  listen: first 1
  listen: second 1
  listen: first 2
  listen: second 2
*/

 

사용자 정의 스트림

import 'dart:async';

// async* 키워드는 yield 키워드를 사용하겠다는 뜻이다.
Stream<int> createStream(List<int> numbers) async* {
  for (int element in numbers) {
    yield element;
  }
}

void main() {
  Stream<int> stream = crateStream(<int>[1, 2, 3, 4, 5]);
  stream.listen(print);
}

/*
  1
  2
  3
  4
  5
*/

◽  async*는 generator를 만든다는 뜻이며, 이는 요청이 있을 때까지 연산을 미루다가 필요할 때 처리한다는 뜻이다.

◽  yeild는 한 번 반환하면 함수가 종료되는 return과 다르게, 열린 채로 존재하여 필요할 때 다른 연산을 할 수 있다.

 

StreamSubscription이란?

💬  위의 예제에서는 listen()에서 이벤트 처리를 할 것 같았으나, 실제로는 StreamSubscription에서 이벤트 콜백을 다룬다.

💬  listen 콜백 함수는 StreamSubscription을 반환하며, StreamSubscription은 앱이 종료되기 전까지 항상 열려 있으므로, 메모리 누수 위험이 있어 닫아줘야 한다.

💬  또한, onData, onError, onDone을 통해 이벤트에 대한 처리를 할 수 있다.

import 'dart:async';

void main() {
  Stream<int> stream =
    new Stream.periodic(new Duration(milliseconds: 200), (x) => x).take(5);
  
  StreamSubscription subscription = stream.listen(
    (data) => print('data: $data'),  // 데이터를 하나씩 처리
    onDone: () {                     // 더 이상 데이터가 없을 때
      print('Stream Completed');
    },
    onError: (err) {                 // 에러가 발생했을 때
      print('Error : $err');
    },
  );
}

void dispose() {
  subscription.cancel();
  super.dispose();
}

/*
  data: 0
  data: 1
  data: 2
  data: 3
  data: 4
  Stream Completed
*/

 

StreamController란?

💬  StreamController는 여러 개의 스트림을 관리하기 위해 만들어졌다.

💬  StreamController를 통해서 값을 넣을 수 있는 스트림을 만들 수 있고, 이는 stream 속성을 통해 접근할 수 있다.

 

Broadcast

💬  기본적으로 만들어지는 스트림은 한 곳에서만 listen 할 수 있다. 따라서 하나의 stream에 listen을 두 개 이상 연결시키면 오류가 발생한다.

💬  하지만, broadcast를 통해서 하나의 stream에 여러 개의 listen을 할당할 수 있다.

import 'dart:async';

void main() {
  StreamController<int> controller = new StreamController.broadcast();
  Stream<int> stream = controller.stream;                   // Stream 접근
  
  stream.listen((event) => print('broadcast 1 : $event'));  // Stream 사용
  stream.listen((event) => print('broadcast 2 : $event'));  // Stream 사용
  
  controller.sink.add(10);                                  // Stream에 값 추가
  controller.sink.add(20);                                  // Stream에 값 추가
  
  controller.close();                                       // Stream 종료
}

/*
  broadcast 1 : 10
  broadcast 2 : 10
  broadcast 1 : 20
  broadcast 2 : 20
*/