[Dart] Stream 사용법 -1

 

Dart에서는 Future와 Stream으로 비동기 프로그래밍을 구현할 수 있습니다. 이번 포스팅에서는 Stream에 대해 알아보겠습니다.

Stream의 사전적인 의미는 흐르다는 뜻이며, 프로그래밍에서는 데이터의 흐름을 의미합니다. Stream을 사용한다면 지속적으로 데이터의 흐름을 관찰할 수 있습니다.

 

 

 

1. Stream 함수의 구조

2. Stream 함수들

3. StreamBasic Widget 

 

 

 

1. stream 함수의 구조

Stream에서는 데이터를 전달하는 부분이 있고, 데이터를 받는 부분이 있습니다. 데이터 전달은 Stream 함수에서 하고, 데이터를 받는 것은 listen에서 합니다.

아래 코드를 보면 countStream이라는 Stream<int>타입의 함수를 만들었습니다. 구조를 보면 Stream는 Future와 흡사합니다.

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

Future에서 async를 썼다면 Stream에서는 async*를 사용합니다. Future와 마찬가지로 await도 사용할 수 있습니다. 그런데 Future에는 없는 yield라는 키워드를 사용할 수 있습니다. yield는 '생산하다'라는 의미로 countStream 함수의 listen에게 데이터를 전달합니다.

  countStream(10).listen((event) {
  	// something
  });

 

 

 

2. Stream 함수들

Stream에는 많은 함수들이 있습니다. 이번에는 take, takeWhile, where만 사용해 보고 다른 함수들을 추후에 더 자세히 알아보겠습니다.

take, takeWhile, where 함수들은 listen함수에 도달하기 까지의 관문이라고 생각하면 됩니다. 

• take : take는 입력한 횟수 만큼만 stream을 통과시킵니다. 입력한 값에 도달한다면 stream은 종료됩니다.
• takeWhile : takeWhile은 bool 값을 리턴해야 하며, false를 리턴한 경우 stream을 종료시킵니다.
• where : where는 takeWhile처럼 bool 값을 리턴해야합니다. true를 리턴하면 stream을 통과시키고, false를 리턴하면 통과시키지 않습니다. takeWhile과 다른 점은 false를 리턴한다 해도 stream을 종료시키진 않습니다. 

countStream(10)
.take(9)
.takeWhile((element) {
  return element <= 8;
}).where((event) {
  return !(event > 6);
}).listen((v) {
	// do something
});

 

 

 

3. StreamBasic Widget 

StreamBasic Widget 은 1번과 2번의 내용을 활용해서 만든 예제입니다. Stream 사용해서 숫자를 누적 합산하는 기능이 있습니다. 

_pageDispose라는 변수가 있는데 dispose할 때 true로 변경시키고 있습니다. 그리고 takeWhile 함수에서 _pageDispose값이 true인 경우 false를 리턴함으로써 stream을 종료시키고 있습니다. 만약 종료시키지 않는다면 'Unhandled Exception: setState() called after dispose()' 오류가 발생합니다. 스트림을 이런 방법으로만 cancel할 수 있는 것은 아닙니다. 다른 방법은 추후에 다시 다루겠습니다.

 

import 'package:flutter/material.dart';

class StreamBasic extends StatefulWidget {
  const StreamBasic({Key? key}) : super(key: key);

  @override
  State<StreamBasic> createState() => _StreamBasicState();
}

class _StreamBasicState extends State<StreamBasic> {
  int sum = 0;

  bool _pageDispose = false;

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: const Text("StreamBasic"),
      ),
      body: Column(
        children: [
          Row(
            mainAxisAlignment: MainAxisAlignment.spaceEvenly,
            children: [
              ///초기화
              _reset(),

              ///시작
              _start(),
            ],
          ),
          const SizedBox(height: 10),
          Text("value : $sum", style: const TextStyle(fontSize: 20)),
        ],
      ),
    );
  }

  ///초기화
  Widget _reset() {
    return ElevatedButton(
        onPressed: () {
          setState(() {
            sum = 0;
          });
        },
        child: const Text("초기화", style: TextStyle(fontSize: 20)));
  }

  @override
  void dispose() {
    _pageDispose = true;
    super.dispose();
  }

  Widget _start() {
    return ElevatedButton(
      onPressed: () async {
        ///take를 활용하면 take에 입력된 횟수 만큼만 listen함수로 들어온다.
        countStream(10).take(9).takeWhile((element) {
          ///takeWhile에서 false를 리턴하면 stream을 종료시킨다
          if (element == 8) {
            return false;
          } else if (_pageDispose) {
            return false;
          } else {
            return true;
          }
        }).where((event) {
          ///where에서 false를 return하면 stream의 listener는 계속 붙어있지만
          ///listen 함수로 진입하지는 않는다
          return !(event > 6);
        }).listen((v) {
          setState(() {
            sum += v;
          });
        });
      },
      child: Container(
        alignment: Alignment.center,
        child: const Text(
          "Stream basic",
          style: TextStyle(fontSize: 20),
        ),
      ),
    );
  }

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