`toAsync`

값이든 Future든, 평범한 Iterable을 FxDart 비동기 파이프라인의 출발점인 FxAsyncIterable로 끌어올립니다.

FxAsyncIterable<T> toAsync<T>(Iterable<FutureOr<T>> iterable) FxAsync<T> Fx.toAsync<T>() // chain, on a sync Fx FxAsync<T> Iterable<FutureOr<T>>.fxAsync // method (resolves the futures)

강의

FxDart의 모든 비동기 파이프라인은 toAsync에서 시작합니다. 평범한 Iterable<FutureOr<T>> — 값의 리스트든, Future의 리스트든, 둘이 섞여 있든 — 를 받아서 모든 *Async 연산자와 FxAsync 체인이 이해하는 타입인 FxAsyncIterable로 감쌉니다. 원소가 Future인 경우에는 끌어당기는 시점에 자동으로 await됩니다.

FxAsyncIterable은 pull 기반입니다. 종결 연산자(toListAsync, eachAsync, FxAsync 체인의 .toList() 등)가 next()를 호출하기 전까지는 아무것도 실행되지 않고, 한 번에 한 단계씩만 진행됩니다 — 평범한 Iterable과 똑같되 비동기라는 점만 다릅니다. 이는 Dart의 Stream과 의도적으로 갈라지는 지점입니다. 스트림은 push 기반이라, 일단 값을 내보내기 시작하면 스스로 페이스를 정하며, 하류 소비자가 "이 중 3개를 한 번에 평가해 달라"고 말할 방법이 없습니다. FxDart의 next([Concurrent? concurrent]) 프로토콜이 바로 그 역채널을 더해 줍니다. concurrent(n) 같은 하류 연산자가 매 pull마다 마커를 상류로 실어 보내 소스에게 n개를 병렬로 실행하라고 요청할 수 있습니다. 스트림에는 이에 해당하는 훅이 없으며, FxDart가 Stream 위에 얹는 대신 자체 비동기 이터러블을 정의한 이유도 오로지 이것입니다.

날것의 Iterable<FutureOr<T>>에는 최상위 함수 toAsync(iterable)을, 이미 만들어 둔 Fx 체인을 FxAsync 체인으로 전환할 때는 체인 메서드 fx(iterable).toAsync()를 쓰세요. 둘 다 지연 평가되므로, 파이프라인을 구성하는 것만으로는 무엇도 실행되지 않고 누군가 값을 끌어당겨야 비로소 움직입니다.

데모 1 · 값, Future, 그리고 체인 형태

toAsync는 평범한 값도, Future도, 둘이 섞인 것도 받습니다 — 체인 형태는 이미 있는 Fx에서 같은 일을 합니다.

데모 2 · pull 기반 모델이 중요한 이유

Dart에서 Future는 await하는 시점이 아니라 생성되는 즉시 실행을 시작합니다. 그래서 리스트 리터럴 안에서 미리 만들어진 Future 세 개는 toAsync가 손대기도 전에 이미 함께 달리고 있습니다. 반면 mapAsync(또는 체인의 .map)는 원소를 실제로 끌어당길 때 비로소 원소마다 새 Future를 하나씩 만듭니다 — concurrent(n)을 붙이지 않는 한, 지연 평가로 하나씩 말이죠.

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  final sw = Stopwatch()..start();

// These three Futures are created right here, eagerly — Dart starts a
  // Future's work the moment it's constructed, not when it's awaited.
  // By the time toAsync/toListAsync touch them, all three are already
  // racing in parallel, so the total time is ~200ms, not ~600ms.
  final alreadyStarted = toAsync([
    delay(const Duration(milliseconds: 200), 'a'),
    delay(const Duration(milliseconds: 200), 'b'),
    delay(const Duration(milliseconds: 200), 'c'),
  ]);
  print(await toListAsync(alreadyStarted)); // [a, b, c]
  print('eager futures: ${sw.elapsedMilliseconds}ms'); // ~200ms

// Contrast: mapAsync creates one Future per element, lazily, only when
  // pulled — so without concurrent(n) they run one at a time.
  sw.reset();
  final lazyPerPull = await fx(['x', 'y', 'z'])
      .toAsync()
      .map((a) => delay(const Duration(milliseconds: 200), a))
      .toList();
  print(lazyPerPull); // [x, y, z]
  print('lazy pulls: ${sw.elapsedMilliseconds}ms'); // ~600ms
}

메서드 표기

xs.fxAsync가 이 함수의 getter 형태입니다. 맨 FxAsyncIterable이 아니라 FxAsync를 돌려주므로 체인을 바로 이어 갈 수 있고, xs.fx와 달리 Future를 풀어 줍니다.

await responses.fxAsync.map(parse).concurrent(4).toList();

// responses.fx would be an Fx<Future<T>> — a chain over the
// futures rather than their values.

이름에 fx가 들어간 것은 의도한 것입니다. toAsync는 너무 일반적인 말이라 맨 Iterable 위에 올라가면 어느 라이브러리로 들어가는지를 전혀 말해 주지 못합니다. getter 표기는 fx에 모아 두었습니다.

직접 해 보기

연습: 아래 리스트를 필터링해서 합격 점수(>= 60)만 결과에 남도록 해 보세요.

관련 항목: *Async 명명 규칙 — mapAsync, filterAsync, … · Stream 브리지 — fromStream, fxStream, toStream · concurrent — 역채널이 실제로 동작하는 모습 · delay & sleep — 비동기 데모 만들기