Stream과 이벤트 처리

Dart의 비동기 프로그래밍에서 Stream은 데이터의 연속된 흐름을 처리하는 데 중요한 역할을 한다. 비동기적으로 발생하는 데이터 이벤트를 처리하기 위해서는 Stream을 적절히 다루어야 하며, Dart는 이를 효과적으로 관리할 수 있는 다양한 기능을 제공한다. 이 섹션에서는 Stream의 기본 개념부터 이벤트 처리 방법, 그리고 더 나아가서 Stream을 활용한 복잡한 비동기 데이터 흐름 처리까지 설명한다.

Stream의 기본 개념

Stream은 Dart에서 일련의 비동기 이벤트를 전달하는 객체이다. 여기서 이벤트는 데이터를 나타낼 수 있으며, 오류나 작업 완료와 같은 상황도 포함될 수 있다. Stream은 여러 개의 이벤트를 순차적으로 전달하며, 각 이벤트는 비동기적으로 처리된다.

Dart의 Stream은 두 가지 유형으로 나눌 수 있다.

Single subscription Stream: 한 번에 하나의 리스너만 등록할 수 있는 Stream이다. 주로 UI 이벤트나 사용자 상호작용과 같은 상황에서 사용된다.
Broadcast Stream: 여러 리스너가 동시에 동일한 Stream을 청취할 수 있는 Stream이다. 서버의 메시지를 여러 클라이언트에게 동시에 전달해야 하는 상황에서 주로 사용된다.

Stream의 생성과 사용

Dart에서 Stream을 생성하는 방법은 다양한다. 가장 기본적인 방법은 StreamController를 사용하는 것이다. StreamController는 Stream과 이를 제어할 수 있는 기능을 제공한다.

StreamController<int> controller = StreamController<int>();
Stream<int> stream = controller.stream;

위 코드에서 controller.stream을 통해 Stream 객체가 생성되었다. 이후 controller를 통해 이벤트를 추가하거나 오류를 전달할 수 있다.

Stream의 주요 메소드

listen(): Stream을 구독하고 이벤트를 처리한다.

stream.listen((event) {
  print('New event: $event');
});

**pause()**와 resume(): Stream을 일시정지하거나 다시 재개한다.

stream.listen((event) {
  print('New event: $event');
}).pause(); // 스트림을 일시정지

cancel(): Stream의 구독을 취소한다.

Stream과 이벤트 처리의 흐름

Stream은 이벤트 기반의 데이터를 다룰 때 매우 유용하다. 예를 들어, 사용자가 특정 동작을 수행할 때마다 이벤트가 발생하는 경우, 이러한 이벤트를 Stream을 통해 처리할 수 있다. 이벤트는 보통 다음과 같이 세 가지 유형으로 구분된다.

데이터 이벤트: 새로운 데이터가 들어왔을 때 발생한다.
오류 이벤트: 비동기 작업 중 오류가 발생했을 때 발생한다.
완료 이벤트: 스트림이 더 이상 이벤트를 전송하지 않을 때 발생한다.

각 이벤트는 Stream 구독자의 콜백 함수에서 처리된다. Stream의 구독자는 보통 다음과 같은 구조를 갖는다:

stream.listen(
  (data) {
    // 데이터 이벤트 처리
  },
  onError: (error) {
    // 오류 이벤트 처리
  },
  onDone: () {
    // 완료 이벤트 처리
  },
);

이처럼 Stream 구독자는 data를 처리하는 메인 콜백과 오류 및 완료 이벤트를 처리하는 추가적인 콜백을 갖는다.

Stream에서의 비동기 이벤트 흐름

Stream은 동기적 또는 비동기적으로 이벤트를 전달할 수 있다. Dart에서 비동기 스트림을 사용하는 경우 await for 구문을 통해 간편하게 비동기 데이터를 처리할 수 있다. 이 구문은 Stream에서 데이터를 비동기적으로 가져오고 이를 순차적으로 처리한다.

await for (var event in stream) {
  print(event);
}

await for는 비동기 데이터를 순차적으로 처리할 때 매우 유용하며, 이벤트가 발생하는 순서대로 데이터를 처리할 수 있게 한다.

StreamController와 StreamTransformer

Stream을 생성할 때 일반적으로 사용하는 객체가 StreamController이다. StreamController는 Stream을 제어할 수 있는 기능을 제공한다. 이를 통해 데이터 이벤트를 추가하거나, 오류를 발생시키거나, 스트림을 종료할 수 있다. StreamController는 단순한 데이터 이벤트 흐름뿐만 아니라 복잡한 비동기 작업을 처리하는 데 중요한 역할을 한다.

StreamController의 주요 메소드

add(): 새로운 데이터를 스트림에 추가한다.
```
controller.add(10);  // 새로운 데이터 10을 추가
```

addError(): 오류 이벤트를 스트림에 추가한다.

controller.addError('오류 발생');  // 오류 이벤트 발생

close(): 스트림을 더 이상 사용할 수 없도록 종료한다.
```
controller.close();  // 스트림 종료
```

StreamController는 Dart에서 스트림을 다룰 때 기본적으로 많이 사용되는 컨트롤러로, 데이터를 제어하는 역할을 담당한다. 또한, StreamController를 활용하면 사용자 정의 스트림을 쉽게 생성하고 제어할 수 있다.

StreamTransformer

StreamTransformer는 입력 스트림을 출력 스트림으로 변환하는 기능을 제공한다. 이를 통해 스트림의 데이터를 원하는 형식으로 변환하거나 중간에서 데이터를 필터링하는 등의 작업을 할 수 있다.

StreamTransformer<int, String> transformer = StreamTransformer.fromHandlers(
  handleData: (int data, EventSink<String> sink) {
    sink.add('변환된 데이터: $data');
  }
);

Stream<String> transformedStream = stream.transform(transformer);
transformedStream.listen((event) {
  print(event);  // "변환된 데이터: ..."
});

위 예제에서는 StreamTransformer를 통해 정수형 데이터를 문자열로 변환하고 있다. 이처럼 StreamTransformer는 입력 스트림에서 데이터를 가져와서 특정 규칙에 따라 변환한 후 새로운 스트림으로 전달할 수 있다.

Stream과 이벤트 루프

Dart에서 Stream은 이벤트 루프를 기반으로 작동한다. 이벤트 루프는 Dart의 비동기 작업을 처리하는 기본 메커니즘으로, Stream의 이벤트 또한 이 이벤트 루프에 의해 관리된다. Dart의 이벤트 루프는 각 이벤트를 큐(queue)에 저장하고, 비동기적으로 이를 처리한다.

마이크로태스크 큐와 이벤트 큐

이벤트 루프에는 두 가지 큐가 존재한다:

마이크로태스크 큐(Microtask Queue): 우선적으로 처리되어야 하는 짧은 작업이 저장된다. 보통 Future.microtask()로 생성된 작업이 여기에 저장된다.
이벤트 큐(Event Queue): 비동기 작업이 이 큐에 저장되며, 일반적인 Future와 Stream 이벤트가 이 큐에 저장된다.

Stream의 이벤트는 보통 이벤트 큐에서 처리되며, 각 이벤트는 순차적으로 처리된다. 이벤트가 처리되는 동안 새로운 이벤트가 큐에 추가되면, 이벤트 루프는 이를 다음 차례로 처리한다.

이러한 Dart의 이벤트 기반 아키텍처 덕분에 Stream은 비동기 데이터 흐름을 효율적으로 처리할 수 있다.

Broadcast Stream

기본적으로 Stream은 한 번에 하나의 구독자만 허용하는 single subscription 방식이다. 하지만 여러 구독자가 동일한 Stream을 청취해야 하는 경우에는 Broadcast Stream을 사용해야 한다.

Broadcast Stream은 여러 구독자를 동시에 허용하며, 주로 이벤트가 여러 대상에게 동시에 전달되어야 할 때 사용된다. 예를 들어, 서버에서 클라이언트로 메시지를 전송할 때 동일한 이벤트가 여러 클라이언트에게 동시에 전달되어야 한다면 Broadcast Stream이 유용하다.

Stream<int> broadcastStream = stream.asBroadcastStream();
broadcastStream.listen((event) {
  print('첫 번째 구독자: $event');
});
broadcastStream.listen((event) {
  print('두 번째 구독자: $event');
});

asBroadcastStream() 메소드를 사용하여 일반 스트림을 Broadcast Stream으로 변환할 수 있으며, 여러 구독자를 동시에 처리할 수 있다.

Stream의 지연 평가와 lazy loading

Stream은 Dart에서 **지연 평가(lazy evaluation)**를 사용하여 비동기 데이터를 처리할 수 있다. 지연 평가는 필요할 때만 데이터를 처리하여 자원을 절약하는 방식이다. Dart의 Stream은 데이터를 즉시 소비하지 않고, 구독자가 Stream을 구독할 때 이벤트를 처리한다.

이 방식은 특히 대용량 데이터를 다루거나 시간이 지남에 따라 점진적으로 데이터를 처리할 때 유용하다. Stream이 지연 로딩 방식으로 동작하기 때문에 구독이 이루어질 때까지는 아무런 데이터도 처리하지 않는다.

Stream<int> stream = Stream.fromIterable([1, 2, 3, 4, 5]);

await for (var value in stream) {
  print(value);  // 구독 시점에서만 데이터가 처리됨
}

위의 예제에서는 Stream이 fromIterable로부터 데이터를 제공하지만, 실제로 구독자가 데이터를 처리할 때까지 지연되며, await for를 통해 구독이 시작되면 데이터가 순차적으로 처리된다.

Stream에서 데이터 필터링

Stream에서 데이터를 필터링할 수 있는 다양한 메소드가 제공된다. 이를 통해 스트림에서 특정 조건을 만족하는 데이터만 처리할 수 있다.

where(): 특정 조건에 맞는 데이터만 필터링한다.

stream.where((event) => event.isEven).listen((event) {
  print('짝수만 처리: $event');
});

map(): 데이터를 변환한다.

stream.map((event) => event * 2).listen((event) {
  print('데이터 변환 후: $event');
});

skip(): 처음의 n개의 데이터를 건너뛰고 그 이후의 데이터를 처리한다.
```
stream.skip(2).listen((event) {
  print('처음 2개 데이터 건너뜀: $event');
});
```

이러한 필터링 기능을 통해 Stream에서 불필요한 데이터를 제외하고 필요한 데이터만 처리할 수 있다.

Stream을 통한 데이터 병합과 분리

여러 개의 Stream을 하나로 병합하거나, 하나의 Stream을 여러 개로 분리하는 작업도 가능한다. 이 작업은 복잡한 비동기 작업을 처리할 때 유용하다.

Stream 병합

Dart에서 제공하는 Stream 관련 유틸리티 함수는 여러 개의 Stream을 하나로 병합할 수 있는 기능을 제공한다. 예를 들어, Stream 두 개를 하나로 결합하여 동시에 두 Stream에서 발생하는 데이터를 모두 처리할 수 있다.

Stream<int> stream1 = Stream.fromIterable([1, 2, 3]);
Stream<int> stream2 = Stream.fromIterable([4, 5, 6]);

Stream<int> mergedStream = Stream.concat([stream1, stream2]);
mergedStream.listen((event) {
  print('병합된 스트림 데이터: $event');
});

위 코드에서 Stream.concat()을 사용하여 두 Stream의 데이터를 병합하여 하나의 Stream으로 처리하고 있다.

Stream 분리

하나의 Stream을 여러 개로 분리하는 것도 가능한다. Dart에서는 Stream을 특정 조건에 따라 여러 하위 Stream으로 나누어 처리할 수 있다.

Stream<int> evenStream = stream.where((event) => event.isEven);
Stream<int> oddStream = stream.where((event) => event.isOdd);

evenStream.listen((event) {
  print('짝수 스트림: $event');
});

oddStream.listen((event) {
  print('홀수 스트림: $event');
});

이처럼 Stream을 원하는 조건에 따라 분리하여 각각의 데이터 흐름을 별도로 처리할 수 있다.

Stream의 Buffering

Stream의 **버퍼링(buffering)**은 일정 시간 동안 발생한 이벤트를 모아서 한꺼번에 처리하는 방식이다. 버퍼링은 네트워크 요청이나 대량의 데이터를 처리할 때 유용하다.

Dart에서는 RxDart와 같은 라이브러리를 사용하여 Stream의 버퍼링 기능을 구현할 수 있다.

stream.bufferTime(Duration(seconds: 2)).listen((bufferedEvents) {
  print('2초 동안의 이벤트: $bufferedEvents');
});

위 코드는 2초 동안 발생한 모든 이벤트를 모아 한꺼번에 처리하는 예제이다. 이 방식은 네트워크나 I/O 작업에서 대량의 이벤트를 처리할 때 매우 유용하다.

Previousasync, await 키워드 Nextchapter_09

Last updated 1 year ago

hashtagStream의 기본 개념

hashtagStream의 생성과 사용

hashtagStream의 주요 메소드

hashtagStream과 이벤트 처리의 흐름

hashtagStream에서의 비동기 이벤트 흐름

hashtagStreamController와 StreamTransformer

hashtagStreamTransformer

hashtagStream과 이벤트 루프

hashtagBroadcast Stream

hashtagStream의 지연 평가와 lazy loading

hashtagStream에서 데이터 필터링

hashtagStream을 통한 데이터 병합과 분리

hashtagStream의 Buffering