Dart 42 🎯 Stream Transformers (StreamTransformer) and Async Operators
A stream transformation is a function that takes a stream and returns a stream. The output stream emits values derived from the input stream’s values, and the relationship between the two is defined by the transformation. map transforms each value, where filters values, take limits the number, expand flattens each value into multiple values, and asyncMap transforms each value with an asynchronous function. These operators are the vocabulary of stream processing, and they are all built on the same primitive: the StreamTransformer.
A StreamTransformer<S, T> is an object that binds to a source stream of type S and produces a new stream of type T. The bind method connects the transformer to the source and returns the output stream. Every operator — map, where, take, expand, asyncMap, asyncExpand, handleError, timeout, and dozens more — is a method that constructs a StreamTransformer and binds it to the source. The operators are the convenience API; the StreamTransformer is the underlying mechanism. Understanding the mechanism is what makes it possible to write a custom transformer when the built-in operators do not fit.
The operators fall into two categories. Synchronous operators like map, where, take, and expand transform values without awaiting anything. Asynchronous operators like asyncMap, asyncExpand, and asyncWhere transform values with a function that returns a future. The asynchronous operators are more powerful but also more subtle: the source stream is paused while the asynchronous function is running, so the order of the values is preserved, but the throughput is limited by the slowest transformation. Choosing the right operator is a matter of understanding what the transformation needs to do.
This chapter covers three areas. First, why StreamTransformer exists — the problem of composing stream transformations and the relationship between the operators and the underlying primitive. Second, the synchronous operators — map, where, take, skip, expand, distinct, and the patterns for combining them. Third, the asynchronous operators — asyncMap, asyncExpand, asyncWhere, and the difference between synchronous and asynchronous transformation. The chapter ends with a complete example session, a quick reference, best practices, common pitfalls, real-world examples, and diagrams showing the transformer pipeline.
Key point: A StreamTransformer<S, T> binds to a source stream and produces a new stream. Every stream operator — map, where, take, asyncMap, and the rest — constructs a transformer and binds it to the source. Synchronous operators transform values without awaiting; asynchronous operators transform values with a function that returns a future and pause the source while the future is pending.
Why StreamTransformer exists
The pipeline problem. A stream transformation is rarely a single operation. A typical pipeline is: read the stream, filter out the values that do not match a condition, transform each remaining value, limit the number, and handle errors. Each of these is a separate operator, and they compose. The output of one operator is the input of the next. The composition is the point: each operator is small and focused, and the pipeline is the combination. The StreamTransformer is the type that makes the composition possible. Each operator returns a new stream, and the next operator binds to that stream.
The custom-transformer problem. The built-in operators cover the common cases, but not every case. A transformation that needs to buffer values, reorder them, or coordinate multiple streams cannot be expressed with the built-in operators alone. The StreamTransformer API is the escape hatch. A custom transformer is a class that implements the bind method, takes a source stream, and returns a new stream. The transformer can use a StreamController internally, listen to the source, and add transformed values to the controller’s stream. This is how the built-in operators are implemented, and it is how a custom transformation is written.
The type-transformation problem. A stream of one type can be transformed into a stream of another type. map takes a stream of S and returns a stream of T, where T is the return type of the mapping function. expand takes a stream of S and returns a stream of T, where T is the element type of the iterable returned by the expansion function. The StreamTransformer<S, T> type captures this: the source is S, the output is T. The type system tracks the transformation, so the output stream’s type is known to the compiler.
The error-handling problem. A stream can emit errors, and the operators have to decide what to do with them. map passes errors through without calling the mapping function. where passes errors through without calling the predicate. handleError intercepts errors and can replace them with values or other errors. asyncMap passes errors through without calling the asynchronous function. The error behavior is part of the operator’s contract, and the StreamTransformer is the mechanism that implements it. A custom transformer must decide how to handle errors, and the built-in transformers provide the patterns.
The cancellation problem. A stream subscription can be canceled, and the transformer must propagate the cancellation to the source. When the output stream’s subscription is canceled, the transformer must cancel its subscription to the source. When the output stream is paused, the transformer must pause the source. The StreamTransformer API handles this coordination: the bind method returns a stream that manages the source subscription, and the cancellation and pause signals propagate correctly.
The trade-off. The operators are convenient and composable, but each one adds a layer of indirection. A pipeline of five operators creates five intermediate streams, and each one has its own subscription, its own error handling, and its own cancellation logic. For a long pipeline, the overhead is measurable. A single custom transformer that performs all five transformations can be more efficient, but it is also more code and more complexity. The trade-off is between the composability of the operators and the efficiency of a hand-written transformer. For most pipelines, the operators are the right choice, and the custom transformer is reserved for the cases where the built-in operators cannot express the transformation.
a. The synchronous operators
The synchronous operators transform values without awaiting anything. The transformation function returns a value directly, and the operator applies it immediately.
map transforms each value with a function. The function receives a value of the source type and returns a value of the output type.
final stream = Stream.fromIterable([1, 2, 3]);
final doubled = stream.map((n) => n * 2);
await doubled.forEach(print); // 2, 4, 6The map operator is the most common transformation. It is synchronous: the mapping function returns a value, and the operator emits it immediately.
where filters values with a predicate. The predicate receives a value and returns a boolean. Values for which the predicate returns true are emitted; values for which it returns false are dropped.
final stream = Stream.fromIterable([1, 2, 3, 4, 5]);
final evens = stream.where((n) => n.isEven);
await evens.forEach(print); // 2, 4The where operator is the filter. It does not transform the values; it selects a subset of them.
take and skip limit the stream by position. take(n) emits the first n values and then closes the stream. skip(n) drops the first n values and emits the rest.
final stream = Stream.fromIterable([1, 2, 3, 4, 5]);
await stream.take(3).forEach(print); // 1, 2, 3
await stream.skip(2).forEach(print); // 3, 4, 5takeWhile and skipWhile limit the stream by predicate. takeWhile emits values while the predicate is true and closes the stream when it becomes false. skipWhile drops values while the predicate is true and emits the rest.
final stream = Stream.fromIterable([1, 2, 3, 4, 5]);
await stream.takeWhile((n) => n < 3).forEach(print); // 1, 2
await stream.skipWhile((n) => n < 3).forEach(print); // 3, 4, 5expand transforms each value into multiple values. The function receives a value and returns an iterable. The operator emits each element of the iterable in order.
final stream = Stream.fromIterable([1, 2, 3]);
final expanded = stream.expand((n) => [n, n * 10]);
await expanded.forEach(print); // 1, 10, 2, 20, 3, 30The expand operator is the inverse of where: instead of dropping values, it multiplies them. It is the flat-map operator for synchronous transformations.
distinct drops consecutive duplicate values. The operator compares each value with the previous one using the equality operator, and drops the value if it is a duplicate.
final stream = Stream.fromIterable([1, 1, 2, 2, 3, 1]);
final distinct = stream.distinct();
await distinct.forEach(print); // 1, 2, 3, 1The distinct operator only drops consecutive duplicates. The final 1 is emitted because it follows a 3, not another 1.
b. The asynchronous operators
The asynchronous operators transform values with a function that returns a future. The source stream is paused while the future is pending, so the order of the values is preserved.
asyncMap transforms each value with an asynchronous function. The function receives a value and returns a Future of the output type. The operator awaits the future and emits the result.
final stream = Stream.fromIterable([1, 2, 3]);
final fetched = stream.asyncMap((n) async {
await Future.delayed(Duration(milliseconds: 100));
return n * 2;
});
await fetched.forEach(print); // 2, 4, 6The asyncMap operator is the asynchronous version of map. The transformation takes time, and the operator waits for each future before emitting the result. The source stream is paused during the wait, so the order is preserved, but the throughput is limited by the slowest transformation.
asyncExpand transforms each value into a stream and flattens the results. The function receives a value and returns a stream. The operator listens to each stream in order and emits its values.
final stream = Stream.fromIterable([1, 2, 3]);
final expanded = stream.asyncExpand((n) {
return Stream.fromIterable([n, n * 10]);
});
await expanded.forEach(print); // 1, 10, 2, 20, 3, 30The asyncExpand operator is the asynchronous version of expand. The difference is that the function returns a stream, not an iterable, and the operator listens to each stream in sequence.
asyncWhere filters values with an asynchronous predicate. The function receives a value and returns a Future<bool>. The operator awaits the future and emits the value if the result is true.
final stream = Stream.fromIterable([1, 2, 3, 4, 5]);
final filtered = stream.asyncWhere((n) async {
await Future.delayed(Duration(milliseconds: 10));
return n.isEven;
});
await filtered.forEach(print); // 2, 4The asyncWhere operator is the asynchronous version of where. It is less common than asyncMap and asyncExpand, but it is useful when the predicate requires an asynchronous check.
The order-preservation property. The asynchronous operators preserve the order of the source values. The source stream is paused while the asynchronous function is pending, so the next value is not processed until the previous one has completed. This is the correct behavior for a pipeline where the order matters. If the order does not matter and the throughput needs to be higher, a different pattern is required: the values are mapped to futures, and the futures are awaited with Future.wait.
final futures = [1, 2, 3].map((n) => fetchData(n));
final results = await Future.wait(futures);The Future.wait pattern processes all the futures concurrently and collects the results. It is faster than asyncMap when the order of the results does not need to match the order of the inputs, but it loses the stream semantics and the backpressure.
c. Custom transformers
A custom transformer is a class that implements StreamTransformer<S, T>. The bind method receives the source stream and returns the output stream. The transformer uses a StreamController internally to create the output stream and listens to the source.
class PairwiseTransformer<T> extends StreamTransformerBase<T, List<T>> {
@override
Stream<List<T>> bind(Stream<T> source) {
final controller = StreamController<List<T>>();
T? previous;
bool hasPrevious = false;
source.listen(
(value) {
if (hasPrevious) {
controller.add([previous as T, value]);
}
previous = value;
hasPrevious = true;
},
onError: controller.addError,
onDone: controller.close,
);
return controller.stream;
}
}The PairwiseTransformer emits a list of two consecutive values for each pair. The first value is stored, and when the second value arrives, the pair is emitted. The transformer listens to the source, handles errors and the done event, and returns the output stream.
The transformer is used with the transform method:
final stream = Stream.fromIterable([1, 2, 3, 4]);
final pairs = stream.transform(PairwiseTransformer<int>());
await pairs.forEach(print); // [1, 2], [2, 3], [3, 4]The transform method binds the transformer to the source and returns the output stream. The method is the entry point for custom transformers, just as map and where are the entry points for the built-in operators.
The cancellation and pause propagation. A custom transformer must propagate the cancellation and pause signals to the source. The StreamController‘s onListen, onPause, onResume, and onCancel callbacks are the hooks. A well-behaved transformer pauses the source when the output stream is paused and cancels the source subscription when the output stream is canceled.
class PairwiseTransformer<T> extends StreamTransformerBase<T, List<T>> {
@override
Stream<List<T>> bind(Stream<T> source) {
late StreamSubscription<T> subscription;
T? previous;
bool hasPrevious = false;
final controller = StreamController<List<T>>(
onListen: () {
subscription = source.listen(
(value) {
if (hasPrevious) {
controller.add([previous as T, value]);
}
previous = value;
hasPrevious = true;
},
onError: controller.addError,
onDone: controller.close,
);
},
onPause: () => subscription.pause(),
onResume: () => subscription.resume(),
onCancel: () => subscription.cancel(),
);
return controller.stream;
}
}The onListen callback starts the source subscription. The onPause, onResume, and onCancel callbacks propagate the signals to the source subscription. This is the pattern for a well-behaved transformer: the output stream’s lifecycle is tied to the source stream’s lifecycle, and the signals propagate correctly.
Complete Example Session
// ============================================
// PART 1: map — TRANSFORM EACH VALUE
// ============================================
import 'dart:async';
void main() async {
final stream = Stream.fromIterable([1, 2, 3]);
final doubled = stream.map((n) => n * 2);
await doubled.forEach(print);
}
// Output: 2, 4, 6
// ============================================
// PART 2: where — FILTER VALUES
// ============================================
void main() async {
final stream = Stream.fromIterable([1, 2, 3, 4, 5]);
final evens = stream.where((n) => n.isEven);
await evens.forEach(print);
}
// Output: 2, 4
// ============================================
// PART 3: take AND skip
// ============================================
void main() async {
final stream = Stream.fromIterable([1, 2, 3, 4, 5]);
await stream.take(3).forEach(print); // 1, 2, 3
await stream.skip(2).forEach(print); // 3, 4, 5
}
// ============================================
// PART 4: expand — ONE VALUE TO MANY
// ============================================
void main() async {
final stream = Stream.fromIterable([1, 2, 3]);
final expanded = stream.expand((n) => [n, n * 10]);
await expanded.forEach(print);
}
// Output: 1, 10, 2, 20, 3, 30
// ============================================
// PART 5: distinct — DROP CONSECUTIVE DUPLICATES
// ============================================
void main() async {
final stream = Stream.fromIterable([1, 1, 2, 2, 3, 1]);
final distinct = stream.distinct();
await distinct.forEach(print);
}
// Output: 1, 2, 3, 1
// ============================================
// PART 6: asyncMap — ASYNCHRONOUS TRANSFORM
// ============================================
void main() async {
final stream = Stream.fromIterable([1, 2, 3]);
final fetched = stream.asyncMap((n) async {
await Future.delayed(Duration(milliseconds: 100));
return n * 2;
});
await fetched.forEach(print);
}
// Output: 2, 4, 6 (with a 100ms delay between each)
// ============================================
// PART 7: asyncExpand — ASYNCHRONOUS FLATTEN
// ============================================
void main() async {
final stream = Stream.fromIterable([1, 2, 3]);
final expanded = stream.asyncExpand((n) {
return Stream.fromIterable([n, n * 10]);
});
await expanded.forEach(print);
}
// Output: 1, 10, 2, 20, 3, 30
// ============================================
// PART 8: CHAINING OPERATORS
// ============================================
void main() async {
final stream = Stream.fromIterable([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
final result = stream
.where((n) => n.isEven)
.map((n) => n * n)
.take(3);
await result.forEach(print);
}
// Output: 4, 16, 36
// Even numbers: 2, 4, 6, 8, 10
// Squared: 4, 16, 36, 64, 100
// Take 3: 4, 16, 36
// ============================================
// PART 9: A CUSTOM TRANSFORMER
// ============================================
class PairwiseTransformer<T> extends StreamTransformerBase<T, List<T>> {
@override
Stream<List<T>> bind(Stream<T> source) {
late StreamSubscription<T> subscription;
T? previous;
bool hasPrevious = false;
final controller = StreamController<List<T>>(
onListen: () {
subscription = source.listen(
(value) {
if (hasPrevious) {
controller.add([previous as T, value]);
}
previous = value;
hasPrevious = true;
},
onError: controller.addError,
onDone: controller.close,
);
},
onPause: () => subscription.pause(),
onResume: () => subscription.resume(),
onCancel: () => subscription.cancel(),
);
return controller.stream;
}
}
void main() async {
final stream = Stream.fromIterable([1, 2, 3, 4]);
final pairs = stream.transform(PairwiseTransformer<int>());
await pairs.forEach(print);
}
// Output: [1, 2], [2, 3], [3, 4]
// ============================================
// PART 10: THE COMPLETE PIPELINE
// ============================================
void main() async {
final stream = Stream.periodic(
Duration(milliseconds: 100),
(count) => count,
).take(10);
final result = stream
.where((n) => n.isEven)
.asyncMap((n) async {
await Future.delayed(Duration(milliseconds: 50));
return n * n;
})
.take(3);
await result.forEach(print);
}
// Output: 0, 4, 16
// Values: 0, 1, 2, 3, 4, 5, 6, 7, 8, 9
// Even: 0, 2, 4, 6, 8
// Squared: 0, 4, 16, 36, 64
// Take 3: 0, 4, 16The ten parts show map, where, take and skip, expand, distinct, asyncMap, asyncExpand, chaining operators, a custom transformer, and the complete pipeline.
Quick Reference
Synchronous Operators
| Operator | Effect |
|---|---|
map(f) | Transform each value |
where(f) | Filter values by predicate |
take(n) | Emit the first n values |
skip(n) | Drop the first n values |
takeWhile(f) | Emit while predicate is true |
skipWhile(f) | Drop while predicate is true |
expand(f) | Transform each value to many |
distinct() | Drop consecutive duplicates |
Asynchronous Operators
| Operator | Effect |
|---|---|
asyncMap(f) | Transform each value with a future |
asyncExpand(f) | Transform each value to a stream |
asyncWhere(f) | Filter with a future predicate |
The Transformer API
| Method | Purpose |
|---|---|
StreamTransformer<S, T> | Transforms a stream of S to a stream of T |
bind(source) | Connects the transformer to the source |
stream.transform(t) | Applies a transformer to a stream |
StreamTransformerBase | Base class for custom transformers |
Error and Cancellation
| Behavior | Synchronous | Asynchronous |
|---|---|---|
| Error pass-through | map, where pass errors | asyncMap passes errors |
| Error handling | handleError intercepts | handleError intercepts |
| Pause propagation | Source paused | Source paused during future |
| Cancel propagation | Source canceled | Source canceled |
Best Practices
✅ Do This:
// Use map for synchronous transformations
stream.map((n) => n * 2); // ✅
// Use where for filtering
stream.where((n) => n.isEven); // ✅
// Use asyncMap for asynchronous transformations
stream.asyncMap((n) async => await fetch(n)); // ✅
// Chain operators for a pipeline
stream.where(...).map(...).take(3); // ✅
// Implement onPause, onResume, and onCancel in custom transformers
onPause: () => subscription.pause(); // ✅
// Use distinct to drop consecutive duplicates
stream.distinct(); // ✅❌ Don’t Do This:
// Don't use asyncMap when a synchronous map works
stream.asyncMap((n) async => n * 2); // unnecessary async // ❌
// Don't forget to handle errors
stream.map((n) => n / 0); // errors pass through // ❌
// Don't create a custom transformer for a built-in operation
class MyMapTransformer<T> ... // use map instead // ❌
// Don't ignore pause and cancel in a custom transformer
// The source subscription must be paused and canceled // ❌
// Don't use expand for asynchronous transformations
stream.expand((n) async => [n]); // returns a future, not an iterable // ❌Common Pitfalls
| Pitfall | Why It Happens | Fix |
|---|---|---|
| Wrong output type | map returns a value of the wrong type | Check the function’s return type |
| Errors pass through silently | No handleError | Add handleError or handle in listen |
asyncMap is slow | Each value awaits the previous | Use Future.wait for concurrency |
| Custom transformer leaks | No onCancel handler | Cancel the source subscription |
| Pause not propagated | No onPause handler | Pause the source subscription |
expand with async | The function returns a future | Use asyncExpand |
| Order not preserved | Used Future.wait | Use asyncMap for order preservation |
Real-World Examples
1. Map
stream.map((n) => n * 2);2. Where
stream.where((n) => n.isEven);3. Take
stream.take(3);4. Expand
stream.expand((n) => [n, n * 10]);5. Distinct
stream.distinct();6. AsyncMap
stream.asyncMap((n) async => await fetch(n));7. AsyncExpand
stream.asyncExpand((n) => Stream.fromIterable([n, n * 10]));8. Chained Pipeline
stream.where(...).map(...).take(3);9. Custom Transformer
class PairwiseTransformer<T> extends StreamTransformerBase<T, List<T>> { ... }10. Transform Method
stream.transform(PairwiseTransformer<int>());Visual
The Transformer Pipeline
┌──────────────────────────────────────────────────────────────┐
│ THE TRANSFORMER PIPELINE │
│ │
│ Source stream: [1, 2, 3, 4, 5, 6] │
│ │ │
│ ▼ │
│ .where((n) => n.isEven) │
│ │ [2, 4, 6] │
│ ▼ │
│ .map((n) => n * n) │
│ │ [4, 16, 36] │
│ ▼ │
│ .take(2) │
│ │ [4, 16] │
│ ▼ │
│ Output stream: [4, 16] │
│ │
│ Each operator constructs a StreamTransformer and binds it │
│ to the previous stream. The result is a pipeline. │
│ │
└──────────────────────────────────────────────────────────────┘Synchronous vs Asynchronous Operators
┌──────────────────────────────────────────────────────────────┐
│ SYNCHRONOUS vs ASYNCHRONOUS │
│ │
│ SYNCHRONOUS (map): │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ value 1 ──► map ──► result 1 │ │
│ │ value 2 ──► map ──► result 2 │ │
│ │ value 3 ──► map ──► result 3 │ │
│ │ │ │
│ │ Each value is transformed immediately. │ │
│ └──────────────────────────────────────────────────────┘ │
│ │
│ ASYNCHRONOUS (asyncMap): │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ value 1 ──► asyncMap ──► await ──► result 1 │ │
│ │ (source paused) │ │
│ │ value 2 ──► asyncMap ──► await ──► result 2 │ │
│ │ (source paused) │ │
│ │ value 3 ──► asyncMap ──► await ──► result 3 │ │
│ │ (source paused) │ │
│ │ │ │
│ │ The source is paused while each future is pending. │ │
│ │ The order is preserved. │ │
│ └──────────────────────────────────────────────────────┘ │
│ │
└──────────────────────────────────────────────────────────────┘The Custom Transformer
┌──────────────────────────────────────────────────────────────┐
│ THE CUSTOM TRANSFORMER │
│ │
│ class PairwiseTransformer<T> │
│ extends StreamTransformerBase<T, List<T>> { │
│ │
│ Stream<List<T>> bind(Stream<T> source) { │
│ final controller = StreamController<List<T>>( │
│ onListen: () { │
│ subscription = source.listen( │
│ (value) { │
│ if (hasPrevious) { │
│ controller.add([previous, value]); │
│ } │
│ previous = value; │
│ }, │
│ onError: controller.addError, │
│ onDone: controller.close, │
│ ); │
│ }, │
│ onPause: () => subscription.pause(), │
│ onResume: () => subscription.resume(), │
│ onCancel: () => subscription.cancel(), │
│ ); │
│ return controller.stream; │
│ } │
│ } │
│ │
│ The transformer listens to the source and adds transformed │
│ values to a controller. The pause and cancel signals are │
│ propagated to the source subscription. │
│ │
└──────────────────────────────────────────────────────────────┘The Complete Pipeline
┌──────────────────────────────────────────────────────────────┐
│ THE COMPLETE PIPELINE │
│ │
│ Stream.periodic(100ms, (c) => c).take(10) │
│ │ [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] │
│ ▼ │
│ .where((n) => n.isEven) │
│ │ [0, 2, 4, 6, 8] │
│ ▼ │
│ .asyncMap((n) async => await transform(n)) │
│ │ [0, 4, 16, 36, 64] │
│ ▼ │
│ .take(3) │
│ │ [0, 4, 16] │
│ ▼ │
│ Output: 0, 4, 16 │
│ │
│ The pipeline combines synchronous and asynchronous │
│ operators. The order is preserved. │
│ │
└──────────────────────────────────────────────────────────────┘Summary
| Item | Value |
|---|---|
StreamTransformer<S, T> | Transforms a stream of S to a stream of T |
bind(source) | Connects the transformer to the source |
transform(t) | Applies a transformer to a stream |
map(f) | Synchronous transform |
where(f) | Synchronous filter |
expand(f) | Synchronous flatten |
asyncMap(f) | Asynchronous transform |
asyncExpand(f) | Asynchronous flatten |
| Custom transformer | Implements StreamTransformerBase |
| Cancellation | Propagated via onCancel |
Key takeaways:
- A
StreamTransformer<S, T>binds to a source stream and produces a new stream. Every operator —map,where,take,asyncMap, and the rest — constructs a transformer and binds it to the source. The operators are the convenience API; the transformer is the underlying mechanism. - The synchronous operators transform values without awaiting.
maptransforms each value,wherefilters,takeandskiplimit by position,takeWhileandskipWhilelimit by predicate,expandmultiplies, anddistinctdrops consecutive duplicates. - The asynchronous operators transform values with a function that returns a future.
asyncMaptransforms each value,asyncExpandflattens each value into a stream, andasyncWherefilters with an asynchronous predicate. The source stream is paused while the future is pending, so the order is preserved. - Chaining operators composes a pipeline. The output of one operator is the input of the next. The pipeline is the composition, and each operator is small and focused.
- A custom transformer is a class that implements
StreamTransformerBase. Thebindmethod listens to the source, transforms the values, and adds them to aStreamController. The controller’s stream is the output. - A well-behaved transformer propagates pause and cancel. The
onPause,onResume, andonCancelcallbacks of the controller pause, resume, and cancel the source subscription. Without this, the transformer leaks the source subscription when the output is canceled. - Errors pass through the operators.
mapandwherepass errors through without calling the transformation function.handleErrorintercepts errors and can replace them. A custom transformer forwards errors to the controller’saddError. asyncMappreserves order but limits throughput. The source is paused while each future is pending, so the values are processed one at a time. For concurrent processing without order preservation, useFuture.waiton a list of futures.
Remember: A StreamTransformer is the primitive that all stream operators are built on. map, where, take, expand, asyncMap, and the rest are methods that construct a transformer and bind it to the source. The synchronous operators transform values immediately; the asynchronous operators transform values with a future and pause the source while the future is pending. A custom transformer implements the bind method, listens to the source, and adds transformed values to a controller. The pause and cancel signals must be propagated to the source, and the errors must be forwarded. The result is a composable pipeline where each operator is a small, focused transformation, and the pipeline is the combination.
Stop using slow, ad-bloated tool sites! 🤮
🔎 Search “KandZ Tools” on Google to use many professional utilities for free.
KandZ.me is the ultimate minimalist hub for:
✅ Finance (Mortgage, Interest, Inflation)
✅ Tech (Base64, JSON, Dev Suite, IP)
✅ Health (BMI, BMR, TDEE)
✅ Productivity (Timer, Workspace, QR)
⚡️ Fast & Private
🔒 No data leaves your device
💎 100% Free
🔗 Use it now: https://tools.kandz.me
🔖 Bookmark it—you’ll need it later!