Dart 비동기 프로그래밍: Streams

Dart 비동기 프로그래밍: Streams

이 글은 반응형 프로그래밍(reactive programming)의 기본 요소 중 하나인 streams, 즉 Stream 타입의 객체를 다룹니다. futures에 관한 앞선 글을 읽었다면 기억하시겠지만, 각 future는 비동기로 전달되는 단일 값(오류 또는 데이터)을 나타내요. Streams도 비슷하게 동작하지만, 단일 항목 대신 시간에 따라 값과 오류를 0개 이상 전달할 수 있어요.

출처: Dart asynchronous programming: Streams

본문

참고: 이 글의 코드를 null safety에 맞게 갱신한 버전을 확인하는 걸 권해요. (비디오는 변경되지 않았어요.)

스트림에서 데이터를 표시하는 간단한 Flutter 앱을 보여드리면서 시작할게요.

이 글은 반응형 프로그래밍의 기본 요소 중 하나인 streams, 즉 Stream 타입의 객체를 다룹니다. futures에 관한 앞선 글을 읽었다면, 각 future가 비동기로 전달하는 단일 값(오류 또는 데이터)을 나타낸다는 걸 기억하실 거예요. Streams도 비슷하게 동작하지만, 단일 항목 대신 시간에 따라 값과 오류를 0개 이상 전달할 수 있어요.

이 글은 Flutter in Focus 비디오 시리즈 Asynchronous Programming in Dart를 바탕으로 한 세 번째 글이에요. 첫 번째 글인 Isolates and event loops는 백그라운드 작업을 지원하는 Dart의 기초를 다뤘고, 두 번째 글인 Futures는 Future 클래스를 다뤘어요. 보고 듣는 것으로 배우는 걸 더 좋아한다면, 이 글의 모든 내용이 다음 비디오에 담겨 있어요.

새 탭에서 YouTube 재생: "Dart Streams - Flutter in Focus"

단일 값이 같은 타입의 iterator와 어떤 관계인지 생각해 보면, 그게 바로 future가 stream과 맺는 관계예요.

futures에서처럼, 핵심은 미리 "데이터 조각이 준비되면 이렇게 하고, 오류가 있으면 이렇게 하고, 스트림이 완료되면 이렇게 한다."고 결정하는 거예요. futures에서와 마찬가지로 Dart 이벤트 루프가 여전히 진행을 관장하고 있어요.

Streams는 Dart 이벤트 루프와 함께 동작해요. 예를 들어 File 클래스의 openRead() 메서드로 파일에서 데이터를 읽는다면, 그 메서드는 스트림을 반환해요. 디스크에서 데이터 덩어리들이 읽혀 이벤트 루프에 도착해요. Dart 라이브러리가 그것들을 보고 "아, 이걸 기다리는 사람이 있네." 하며 데이터를 스트림에 추가하고, 그러면 그 데이터가 당신의 앱 코드 쪽으로 튀어나와요.

또 다른 데이터 조각이 도착하면 들어가고 나오고를 반복하죠. 타이머 기반 스트림, 네트워크 소켓에서 스트리밍하는 데이터도 시계 이벤트와 네트워크 이벤트를 사용해 이벤트 루프와 함께 동작해요.

스트림 듣기

스트림이 제공하는 데이터로 작업하는 방법을 이야기해 볼게요. 초당 한 번씩 새 정수(1, 2, 3, 4, 5…)를 내뱉는 스트림을 주는 클래스가 있다고 해볼게요.

listen() 메서드로 스트림을 구독할 수 있어요. 유일한 필수 파라미터는 함수예요.

final myStream = NumberCreator().stream;

final subscription = myStream.listen(
  (data) => print('Data: $data'),
);

스트림이 새 값을 내뱉을 때마다 함수가 호출되어 값을 출력해요.

Data: 1
Data: 2
Data: 3
Data: 4
...

이것이 listen()이 동작하는 방식이에요.

중요: 기본적으로 스트림은 단일 구독(single subscription)용으로 설정돼요. 누군가 구독할 때까지 값을 붙잡고 있고, 수명 전체에 걸쳐 리스너 하나만 허용해요. 스트림을 두 번 들으려고 하면 예외가 발생해요. 다행히 Dart에는 broadcast streams도 있어요. asBroadcastStream() 메서드로 단일 구독 스트림에서 브로드캐스트 스트림을 만들 수 있어요. 브로드캐스트 스트림은 단일 구독 스트림과 동일하게 동작하지만, 여러 리스너를 가질 수 있어요. 그리고 데이터 조각이 준비됐을 때 아무도 듣고 있지 않으면 그 데이터는 버려져요.

final myStream = NumberCreator().stream.asBroadcastStream();

final subscription = myStream.listen(
  (data) => print('Data: $data'),
);

final subscription2 = myStream.listen(
  (data) => print('Data again: $data'),
);

첫 번째 listen() 호출로 돌아가 볼게요. 이야기할 게 몇 가지 더 있거든요.

앞서 언급했듯이 스트림은 futures가 그렇듯 오류도 만들어낼 수 있어요. listen() 호출에 onError 함수를 추가하면 오류를 잡아 처리할 수 있어요. 기본값이 true인 cancelOnError 프로퍼티도 있는데, false로 설정하면 오류가 발생한 뒤에도 구독을 계속 유지할 수 있어요. 그리고 스트림이 데이터 전송을 끝냈을 때(예: 파일을 완전히 읽었을 때) 실행할 코드를 onDone 함수로 추가할 수도 있어요.

이 네 파라미터(onError, onDone, cancelOnError, 그리고 필수 파라미터인 onData)를 모두 조합하면 무슨 일이 생겨도 미리 준비된 상태가 돼요.

final subscription = myStream.listen(
  (data) {
    print('Data: $data');
  },
  onError: (err) {
    print('Error!');
  },
  cancelOnError: false,
  onDone: () {
    print('Done!');
  },
);

팁: listen()이 반환하는 작은 구독 객체에는 유용한 메서드가 몇 가지 있어요. 이 객체는 StreamSubscription이며, 데이터 흐름을 일시 중지(pause), 재개(resume), 취소(cancel)하는 데 쓸 수 있어요.

final subscription = myStream.listen(...);

subscription.pause();
subscription.resume();
subscription.cancel();

스트림 사용하고 조작하기

이제 listen()으로 스트림을 구독하고 데이터 이벤트를 받는 법을 알았으니, 스트림을 정말 멋지게 만드는 것, 바로 스트림 조작에 대해 이야기할 수 있어요. 스트림에 데이터가 들어오면 많은 연산이 유창하고 우아해져요.

앞서 쓴 숫자 스트림으로 돌아가서, map()이라는 메서드로 스트림의 각 값을 즉석에서 다른 것으로 변환할 수 있어요. map()에 변환을 수행할 함수를 주면, 함수의 반환 값에 맞춰 타입이 지정된 새 스트림을 반환해요. int 스트림 대신 이제 string 스트림을 갖게 되는 거죠. 끝에 listen() 호출을 붙이고 print() 함수를 주면, 이제 도착하는 대로 비동기로 스트림에서 직접 문자열을 출력해요.

NumberCreator().stream
    .map((i) => 'String $i')
    .listen(print);
/*
OUTPUT:
String 1
String 2
String 3
String 4
*/

이렇게 체인으로 이어 붙일 수 있는 메서드가 엄청 많아요. 예를 들어 짝수만 출력하고 싶다면 where()로 스트림을 필터링할 수 있어요. 각 요소에 대해 boolean을 반환하는 테스트 함수를 주면, 테스트를 통과하는 값만 포함하는 새 스트림을 반환해요.

NumberCreator().stream
    .where((i) => i % 2 == 0)
    .map((i) => 'String $i')
    .listen(print);
/*
OUTPUT:
String 2
String 4
String 6
String 8
*/

distinct() 메서드도 좋은 예시예요. Redux 스토어를 사용하는 앱이 있다면, 그 스토어가 onChange 스트림에서 새 앱 상태 객체를 내보내요. map()으로 그 상태 객체 스트림을 앱의 한 부분을 위한 뷰 모델 스트림으로 변환할 수 있어요. 그다음 distinct() 메서드로 연속된 동일 값을 걸러내는 스트림을 얻어요(스토어가 뷰 모델의 데이터 하위 집합에 영향을 주지 않는 변경을 내보내는 경우 대비). 그리고 나서 listen으로 새 뷰 모델이 나올 때마다 UI를 갱신하면 돼요.

myReduxStore.onChange
    .map((s) => MyViewModel(s))
    .distinct()
    .listen( /* update UI */ );

Dart에 내장된 추가 메서드도 스트림을 다듬고 수정하는 데 쓸 수 있는 게 많아요. 게다가 더 고급 기능을 쓸 준비가 되면, Dart 팀이 관리하며 pub.dev에서 제공하는 async 패키지가 있어요. 두 스트림을 병합하거나, 결과를 캐시하거나, 그 밖의 스트림 기반 마법을 수행하는 클래스가 들어 있죠. 스트림 마법이 더 필요하다면 stream_transform 패키지도 살펴보세요.

스트림 만들기

여기서 언급할 가치가 있는 고급 주제가 하나 있는데, 바로 자신만의 스트림을 만드는 방법이에요. futures에서와 마찬가지로 대부분은 네트워크 라이브러리, 파일 라이브러리, 상태 관리 등이 만들어 준 스트림으로 작업하게 돼요. 하지만 StreamController를 사용해 직접 만들 수도 있어요.

지금까지 써 온 NumberCreator로 돌아가 볼게요. 실제 코드는 이렇습니다.

class NumberCreator {
  NumberCreator() {
    Timer.periodic(Duration(seconds: 1), (t) {
      _controller.sink.add(_count);
      _count++;
    });
  }

  var _count = 1;
  final _controller = StreamController<int>();
  Stream<int> get stream => _controller.stream;
}

보시다시피 실행 중인 카운트를 유지하고, 타이머로 매 초마다 카운트를 증가시켜요. 흥미로운 부분은 스트림 컨트롤러예요. StreamController는 처음부터 완전히 새로운 스트림을 만들고, 그 양쪽 끝에 접근할 수 있게 해줘요. 데이터가 도착하는 스트림 끝이 있어요. 이 글 내내 이 끝을 사용해 왔죠.

Stream<int> get stream => _controller.stream;

그리고 새 데이터가 스트림에 추가되는 sink 끝이 있어요.

_controller.sink.add(_count);

여기 NumberCreator는 둘 다 사용해요. 타이머가 울리면 최신 카운트를 컨트롤러의 sink에 추가하고, 컨트롤러의 스트림을 공개 프로퍼티로 노출해서 다른 객체들이 구독할 수 있게 해요.

스트림으로 Flutter 위젯 만들기

스트림을 만들고, 조작하고, 듣는 법을 다뤘으니, 이제 Flutter에서 위젯을 만드는 데 활용하는 방법을 이야기해 볼게요.

futures에 관한 앞선 비디오를 봤다면 FutureBuilder를 기억하실 거예요. future와 builder 메서드를 주면, future의 상태에 따라 위젯을 만들어요. 스트림에는 StreamBuilder라는 비슷한 위젯이 있어요. 스트림과 builder 메서드를 주면, 스트림이 새 값을 내뱉을 때마다 자식 위젯을 다시 빌드해요.

StreamBuilder<String>(
  stream: NumberCreator().stream.map((i) => 'String $i'),
  builder: (context, snapshot) {
    /* Build widgets! */
  }
)

snapshot 파라미터는 FutureBuilder 때와 마찬가지로 AsyncSnapshot이에요. connectionState 프로퍼티를 확인해서 스트림이 아직 데이터를 보내지 않았는지, 아니면 완전히 끝났는지 볼 수 있어요. hasError 프로퍼티로 최신 값이 오류인지도 확인할 수 있고, 당연히 데이터 값을 처리할 수도 있어요.

StreamBuilder<String>(
  stream: NumberCreator().stream.map((i) => 'String $i'),
  builder: (context, snapshot) {
    if (snapshot.connectionState == ConnectionState.waiting) {
      return Text('No data yet.');
    } else if (snapshot.connectionState == ConnectionState.done) {
      return Text('Done!');
    }
  }
)

가장 중요한 것은 builder가 스트림의 가능한 모든 상태를 처리하는 방법을 알고 있게 하는 거예요. 그게 준비되면 스트림이 무엇을 하든 반응할 수 있죠.

정리

이 글에서는 스트림이 무엇을 나타내는지, 스트림에서 값을 얻는 방법, 그 값을 조작하는 여러 방식, 그리고 StreamBuilder가 Flutter 앱에서 스트림 값을 사용하는 데 어떻게 도움을 주는지 다뤘어요.

스트림에 대해 더 배우려면 Dart와 Flutter 문서를 참고하세요. dart.dev에서는 streams 튜토리얼, 명령줄 앱 튜토리얼, 그리고 라이브러리 투어의 stream 섹션을 확인하세요. flutter.dev에서는 StreamBuilder를 사용하는 예시가 있는 Work with WebSockets를 살펴보세요.

아니면 Asynchronous Programming in Dart 시리즈의 다음 비디오로 넘어가세요. 비동기 코드를 빡빡하고 읽기 쉽게 유지하도록 돕는 두 키워드 아니 async와 await에 관한 내용이에요.

새 탭에서 YouTube 재생: "Async/Await - Flutter in Focus"

이 글이 바탕으로 한 비디오를 만든 Andrew Brogdon에게 큰 감사를 전합니다.

더 알아보기