Stream bridges

fromStream, fxStream, and .toStream() — cross freely between Dart's Stream and FxDart's FxAsyncIterable.

FxAsyncIterable<T> fromStream<T>(Stream<T> stream) // lossless FIFO + pause FxAsyncIterable<T> fromStreamLatest<T>(Stream<T> stream) // drop superseded FxAsyncIterable<List<T>> fromStreamChunked<T>(Stream<T> stream) // batches FxAsyncIterable<T> fromStreamNext<T>(Stream<T> stream) // demand-gated drop FxAsync<T> fxStream<T>(Stream<T> stream) // chain, wraps fromStream Stream<T> FxAsyncIterableToStream.toStream<T>() // extension on FxAsyncIterable Stream<T> FxAsync.toStream() // chain FxAsync<T> FxEvents.pull() // = fromStream FxAsync<T> FxEvents.pullLatest() // = fromStreamLatest FxAsync<List<T>> FxEvents.pullChunked() // = fromStreamChunked FxAsync<T> FxEvents.pullNext() // = fromStreamNext

Lecture

fromStream converts any Stream — single- or broadcast-subscription — into an FxAsyncIterable, so you can run the whole FxDart operator set (map, filter, concurrent, …) over data that's arriving from a socket, a file, a widget's event stream, or anywhere else Dart hands you a Stream. fxStream(stream) is the same thing, but returns a chainable FxAsync directly instead of a raw FxAsyncIterable — the async counterpart of fx and fxAsync.

Going the other way, .toStream() drives an FxAsyncIterable (or FxAsync chain) to completion and re-emits its values as a plain Stream — handy when some other API (a StreamBuilder, for example) expects one. One caveat: toStream() always pulls sequentially, ignoring any concurrent(n) upstream of it — apply concurrent/concurrentPool to the chain before calling .toStream() if you need the parallelism to actually happen; the stream conversion itself won't add it.

Crossing from push to pull is not one operation — it is four, because a stream may keep emitting while the consumer is busy. RxJS 9 names the four iterateEach, iterateLatest, iterateBuffered and iterateNext. FxDart maps them onto fromStream* (raw iterable) and FxEvents.pull* (chain):

RxJS 9	FxDart	while you are busy
`iterateEach`	`fromStream` / `.pull()`	lossless FIFO — pause the source, queue every value
`iterateLatest`	`fromStreamLatest` / `.pullLatest()`	drop superseded — keep only the newest unread value
`iterateBuffered`	`fromStreamChunked` / `.pullChunked()`	batch — yield the arrivals as one list
`iterateNext`	`fromStreamNext` / `.pullNext()`	demand-gated drop — ignore anything that arrived with no pull waiting

fromStream is the default and the one Demo 1 uses, because a file or a socket should not lose bytes. Reach for latest when a UI only cares about the current reading, chunked when the consumer wants work in batches, and next when stale events are worse than gaps.

Demo 1 · fromStream and fxStream

Both wrap a Stream.fromIterable so you can pipe an existing stream through FxDart operators:

Demo 2 · Round-trip, with a finite periodic stream

Stream.periodic never ends on its own, so .take(n) keeps the demo finite. The second half shows the reverse direction — building an FxAsync chain, then handing it back out as a plain Stream with .toStream():

Demo 3 · Four ways to pull a stream

A sync burst of 1, 2, 3 arrives while a pull is already waiting. fromStream keeps every value; fromStreamLatest keeps only the newest; fromStreamChunked yields them as one list; fromStreamNext keeps only the value that met the waiting pull. The events-chain spellings are .pull(), .pullLatest(), .pullChunked(), .pullNext().

import 'dart:async';

import 'package:fxdart/fxdart.dart';

/// Start a pull, then push 1, 2, 3 on a sync controller. The four
/// policies disagree about what that burst becomes.
Future<List<Object?>> drainBurst<T>(
  FxAsyncIterable<T> Function(Stream<int>) wrap,
) async {
  final c = StreamController<int>(sync: true);
  final it = wrap(c.stream).iterator;
  final first = it.next();
  c
    ..add(1)
    ..add(2)
    ..add(3)
    ..close();
  final out = <Object?>[];
  var r = await first;
  if (!r.done) out.add(r.value);
  while (true) {
    r = await it.next();
    if (r.done) break;
    out.add(r.value);
  }
  return out;
}

Future<void> main() async {
  print('fromStream        ${await drainBurst(fromStream)}');
  // [1, 2, 3] — lossless FIFO
  print('fromStreamLatest  ${await drainBurst(fromStreamLatest)}');
  // [3] — latest-wins
  print('fromStreamChunked ${await drainBurst(fromStreamChunked)}');
  // [[1, 2, 3]] — batched
  print('fromStreamNext    ${await drainBurst(fromStreamNext)}');
  // [1] — demand-gated

// Same four, from the events chain:
  //   fxEvents(s).pull() / pullLatest() / pullChunked() / pullNext()
}

Two chains, one Stream

A Stream is the one source that belongs to both halves of FxDart, so it carries two getters. They are not variants of each other — they are different models:

	`stream.fx`	`stream.fxEvents`
gives	`FxAsync<T>`	`FxEvents<T>`
same as	`fxStream(stream)`	`fxEvents(stream)`
model	pull — data over demand	push — events over time
who sets the pace	the consumer, one `next()` at a time	the stream; operators reshape the timing
you reach for	`map`, `filter`, `concurrent`, `toList`	`debounce`, `throttle`, `switchMap`, `combineLatest`
backpressure	yes — nothing is pulled until asked	no — a stream emits when it emits

The rule of thumb: if the question is “how many at once?” you want the pull chain, because concurrent(n) only means something when the consumer controls demand. If the question is “how often, and which one wins?” you want the event chain. Start in either and cross over — .pull() / .pullLatest() / .pullChunked() / .pullNext() turn an FxEvents into the pull chain, .toStream() turns an FxAsync back into a Stream.

Full lessons: fxEvents for the push chain, fx for the chain model and the getter spellings, concurrent for the thing only the pull side can do.

Try it yourself

Exercise: keep only the values >= 10 from this stream.

Related: toAsync — lift a plain Iterable instead · async variants — the *Async naming convention · concurrent — apply before toStream() for real parallelism · concurrentPool — completion-order variant · fxEvents — the push chain; .pull() / .pullLatest() / .pullChunked() / .pullNext() cross back