In Dart, how would you use a StreamTransformer and transform() to turn a socket's raw bytes into a stream of chat messages?
answer
- transform(t) is just t.bind(this)
- decode bytes, split lines, parse
- fromHandlers gets a sink per event
- fromBind composes existing transformers
- Socket is Stream<Uint8List>
basics
~10 sChain transformers: decode the socket's bytes to text, split into lines, then parse each line with a StreamTransformer built by StreamTransformer.fromHandlers, whose handleData adds a ChatMessage to its sink. transform(t) simply calls t.bind(stream).
solid answer
~30 sA `StreamTransformer<S, T>` turns a `Stream<S>` into a `Stream<T>` through its `bind` method, and `stream.transform(t)` is just `t.bind(stream)`, which makes pipelines read left to right. For a chat socket: `utf8.decoder.bind(socket)` turns bytes into text, `.transform(const LineSplitter())` yields one line per frame, and a `StreamTransformer<String, ChatMessage>.fromHandlers(handleData: ...)` parses each line and calls `sink.add(message)` — or `sink.addError` for a bad frame. `fromBind` wraps an existing stream-to-stream function, and subclassing `StreamTransformerBase` suits a stateful transformer. Transformers forward pause and cancel upstream, so the pipeline stays lazy. Note that `socket.transform(utf8.decoder)` does not compile: `Socket` is a `Stream<Uint8List>` and the decoder accepts `List<int>`.
code
dart · 25 linesimport 'dart:async';
import 'dart:convert';
import 'dart:io';
final class ChatMessage {
ChatMessage(this.author, this.text);
final String author;
final String text;
}
final chatFrames = StreamTransformer<String, ChatMessage>.fromHandlers(
handleData: (line, sink) {
if (line.isEmpty) return;
if (jsonDecode(line) case {'author': String author, 'text': String text}) {
sink.add(ChatMessage(author, text));
} else {
sink.addError(FormatException('Bad chat frame', line));
}
},
);
Stream<ChatMessage> chatMessages(Socket socket) => utf8.decoder
.bind(socket)
.transform(const LineSplitter())
.transform(chatFrames);go deeper
Know that transform() chains a reusable step onto a stream and that a transformer can emit more or fewer events than it receives.
Build one with fromHandlers, explain the sink, the omitted-handler defaults and error events, and why utf8.decoder.bind is needed on a Socket.
Choose between fromHandlers, fromBind and a StreamTransformerBase subclass based on state, and rely on transformers forwarding pause and cancel instead of a hand-fed controller.
Treat protocol decoding as a tested pipeline of small transformers, so a new frame type or a codec change touches one step rather than every listener.
## What a transformer is A **`StreamTransformer<S, T>`** (library `dart:async`) is an object with one essential method, `Stream<T> bind(Stream<S> stream)`. It describes a reusable step that turns one stream into another. **`Stream.transform(t)`** is a convenience that returns `t.bind(this)`, so steps chain fluently: ```dart socketText.transform(const LineSplitter()).transform(chatFrames) ``` Unlike `map`, a transformer is not limited to one output per input. It can emit zero, one or many events per input event, hold state between events, and emit extra events when the source finishes. That is exactly what a wire protocol needs, where a frame can be split across several network reads or several frames can arrive in one. ## Three ways to build one | Constructor | You provide | Good for | |---|---|---| | `StreamTransformer.fromHandlers` | `handleData(data, sink)`, `handleError(error, stackTrace, sink)`, `handleDone(sink)` | Stateless per-event work: parse, filter, fan out | | `StreamTransformer.fromBind` | A function `Stream<T> Function(Stream<S>)` | Packaging an existing chain as one named transformer | | Subclass `StreamTransformerBase<S, T>` | Your own `bind` | Stateful or configurable transformers | The default `StreamTransformer(onListen)` constructor, which builds the output subscription by hand, exists too but is rarely needed. Rules for `fromHandlers` worth knowing: - Each handler receives an `EventSink<T>`; use it only during that call, never store it. - An omitted `handleData` forwards data unchanged, an omitted `handleError` forwards errors, an omitted `handleDone` closes the output. - Calling `sink.close()` from a handler closes the output and cancels the source subscription. - An exception thrown by a handler becomes an **error event** on the output stream rather than crashing the pipeline. - The handlers should be **stateless**: the same transformer object may be bound to several streams, or to one broadcast stream listened to several times, and they would share any captured state. For per-stream state, create the handlers inside `fromBind` or a `StreamTransformerBase` subclass. ## The chat pipeline For newline-delimited JSON over a TCP socket: 1. **Bytes to text.** `dart:io`'s `Socket` implements `Stream<Uint8List>`. `utf8.decoder` is a `Converter<List<int>, String>`, and converters are transformers, but `socket.transform(utf8.decoder)` fails to type-check because a transformer from `List<int>` is not a transformer from `Uint8List`. The SDK's documented fix is to call `utf8.decoder.bind(socket)` directly. 2. **Text to frames.** `.transform(const LineSplitter())` emits one string per line, however the text was chunked. 3. **Frames to messages.** A `fromHandlers` transformer decodes each line with `jsonDecode`, pattern-matches the map, and calls `sink.add(ChatMessage(...))`, or `sink.addError(FormatException(...))` for a malformed frame so the listener can report it without losing the connection. ## Why transformers beat a hand-fed controller here - **Laziness.** Nothing reads the socket until the final stream is listened to. - **Flow control.** The SDK's transformers forward the listener's `pause`, `resume` and `cancel` to the upstream subscription, so a slow message list pauses the socket instead of filling a buffer. - **Broadcast-ness follows the source.** A transformed broadcast stream stays broadcast; a single-subscription source stays single-subscription. - **Testability.** `chatFrames` can be tested with `Stream.fromIterable(['{...}'])` and no socket at all. If the parsing step needs to be asynchronous — say, looking up the author in a local database — that is a job for `asyncMap` rather than a synchronous `handleData`.
- Why should a transformer built with fromHandlers not keep state in captured variables?The same transformer instance can be bound to several streams, or to one broadcast stream listened to several times; every binding calls the same handlers, so captured counters or partial buffers are shared and corrupted. For per-stream state, create the handlers inside `StreamTransformer.fromBind` or subclass `StreamTransformerBase` and allocate state in `bind`.
- What happens if jsonDecode throws inside handleData?The exception is caught by the transformer and emitted as an error event on the output stream. The stream keeps running unless the listener subscribed with `cancelOnError: true`. Adding a `FormatException` explicitly with `sink.addError` makes the intent clearer and lets you attach the offending line.
saying these in an interview costs you the question
- transform() and map() are interchangeable because both emit one event per input.
- socket.transform(utf8.decoder) compiles fine on a dart:io Socket.
- It is fine to store the EventSink from handleData and add to it later.
- An exception in handleData tears down the whole stream.
- A transformer starts reading its source as soon as transform() is called.