Irbisa · Шпаргалка
Потоки (Streams)
Streams, StreamController, StreamBuilder, широковещательные потоки
Чем
Futureотличается отStreamи когда что использовать?Что такое
Streamв Dart и чем single-subscription стрим отличается от broadcast?// Single-subscription: exactly one listenerfinal single = StreamController<int>();single.stream.listen((v) => print('A got $v'));// single.stream.listen(...); // Bad state: Stream has already been listened to // Broadcast: many listeners, no replay…Что напечатает этот код и что изменится, если убрать
.broadcast()?import 'dart:async'; Future<void> main() async { final controller = StreamController<String>.broadcast(); controller.add('hello'); controller.stream.listen(print); controller.add('world'); await controller.close();}Чем
FutureBuilderотличается отStreamBuilder?class _ProfileState extends State<Profile> { late final Future<User> _user = api.fetchUser(widget.id); // once late final Stream<int> _unread = chat.watchUnread(widget.id); // over time @override Widget build(BuildContext context) {…Что даёт
StreamControllerи как с его помощью превратить колбэк-API (датчик, сокет) вStream?Stream<int> watch(ValueListenable<int> source) { late final StreamController<int> controller; void push() => controller.add(source.value); controller = StreamController<int>.broadcast( onListen: () => source.addListener(push), // first listener: start…Что такое
StreamSubscriptionи как управлять её жизненным циклом в виджете?class _TickerState extends State<Ticker> { StreamSubscription<int>? _sub; int _last = 0; @override void initState() {…Когда использовать
await for, а когдаlistenна стриме?// Sequential: the next order is not delivered until this one is savedFuture<void> saveAll(Stream<Order> orders) async { await for (final order in orders) { await db.insert(order); if (order.isLast) break; // cancels the subscription }…Что напечатает этот код и в каком порядке?
Stream<int> numbers() async* { print('start'); yield 1; yield 2;} void main() { final stream = numbers(); print('created'); stream.listen(print); print('listened');}Что делает за вас
StreamBuilderи о чём говорятconnectionStateи snapshot?class _PricesState extends State<Prices> { late final Stream<Quote> _quotes = repo.watchQuotes(widget.symbol); @override Widget build(BuildContext context) { return StreamBuilder<Quote>(…Два виджета слушают один и тот же стрим, и прилетает «Bad state: Stream has already been listened to». Что происходит?
// wrong: two listeners on a single-subscription streamfinal controller = StreamController<int>();controller.stream.listen(print);controller.stream.listen(print); // Bad state: already listened to// right: broadcast when you own the controllerfinal broadcast = StreamController<int>.broadcast();…Как ошибки проходят через
Streamи что происходит с ошибкой, которую никто не обработал?// wrong: data handled, errors escape to the zone and the crash reporter_sub = repo.messages.listen((m) => setState(() => _messages.add(m))); // right: handle both channels_sub = repo.messages.listen( (m) => setState(() => _messages.add(m)),…Найдите баг: прогресс-бар доходит до 100%, но сообщение «Upload complete» так и не появляется.
Stream<double> upload(Uploader client, File file) { final progress = StreamController<double>(); client .send(file, onProgress: (sent, total) => progress.add(sent / total)) .catchError((Object e, StackTrace s) => progress.addError(e, s)); return progress.stream;} Future<void> finish(Uploader client, File file) async { await for (final p in upload(client, file)) { setProgress(p); } showDone();}Что такое
StreamTransformerи как написать свой оператор для стрима без rxdart?// Per event: trim lines and drop blank onesfinal trimNonEmpty = StreamTransformer<String, String>.fromHandlers( handleData: (line, sink) { final t = line.trim(); if (t.isNotEmpty) sink.add(t); },…Как объединить несколько стримов:
merge,zip,combineLatest?import 'package:rxdart/rxdart.dart'; // merge: every event from every source, in arrival orderfinal allMessages = Rx.merge<Message>([channelA.messages, channelB.messages]); // zip: the i-th question with the i-th answer…Что добавляет пакет rxdart поверх
dart:async: зачем нужныBehaviorSubject,debounceTimeиswitchMap?import 'package:rxdart/rxdart.dart'; class SearchViewModel { SearchViewModel(this._api); final SearchApi _api;…Репозиторий на каждом опросе присылает идентичный список, и страница каждый раз перестраивается; вы добавили
distinct(), но ничего не изменилось: почему?Вы пишете
await feed.quotes.first, чтобы прочитать текущую цену, и вызов не возвращается никогда: что вы предположили оStreamтакого, чего у него нет?Снекбар (
SnackBar) об успешном входе показывается три раза, потому что его вызывают внутриbuilderуStreamBuilder: какое правило нарушено?// wrong: runs on every rebuild, not on every eventbuilder: (context, snap) { if (snap.data is SignedIn) { ScaffoldMessenger.of(context).showSnackBar(welcome); } return const HomeBody();…Найдите баг: после переключения между чатами в открытом чате продолжают появляться сообщения из чата, который вы покинули.
class ChatController { ChatController(this._repo); final ChatRepository _repo; final messages = ValueNotifier<List<Message>>(const []); StreamSubscription<Message>? _sub; void open(String chatId) { messages.value = const []; _sub = _repo .watch(chatId) .listen((m) => messages.value = [...messages.value, m]); } void dispose() { _sub?.cancel(); messages.dispose(); }}Как стримы используются для управления состоянием во Flutter и почему состояние редко держат в голом
StreamController?Что такое backpressure и как с ним работают single-subscription и broadcast стримы?
// A producer that respects backpressure: suspends at yield while pausedStream<List<int>> readChunks(File file) async* { final raf = await file.open(); try { while (true) { final chunk = await raf.read(64 * 1024);…Лента котировок по WebSocket затихает на нестабильной мобильной сети, стрим не отдаёт ошибку, а UI бесконечно показывает старую цену: что добавить?
class PriceFeed { final _out = StreamController<Quote>.broadcast(); IOWebSocketChannel? _channel; StreamSubscription<Quote>? _sub; Timer? _retry; int _attempt = 0;…Тест добавляет событие в
StreamControllerи сразу проверяет слушателя, но тот ещё не отработал: решается ли это флагомsync: true?// wrong: the assertion runs before the listener doestest('stores the value', () { final controller = StreamController<int>(); var seen = 0; controller.stream.listen((v) => seen = v); controller.add(7);…