blob: eadae19a0cd637e805407b7ede279f44adb1eca7 [file] [log] [blame]
// Copyright (c) 2015, the Dart project authors. Please see the AUTHORS file
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
import 'dart:async';
import 'package:async/async.dart';
import 'package:test/test.dart';
void main() {
group('single-subscription', () {
late StreamGroup<String> streamGroup;
setUp(() {
streamGroup = StreamGroup<String>();
});
test('buffers events from multiple sources', () async {
var controller1 = StreamController<String>();
streamGroup.add(controller1.stream);
controller1.add('first');
controller1.close();
var controller2 = StreamController<String>();
streamGroup.add(controller2.stream);
controller2.add('second');
controller2.close();
await flushMicrotasks();
expect(streamGroup.close(), completes);
expect(streamGroup.stream.toList(),
completion(unorderedEquals(['first', 'second'])));
});
test('buffers errors from multiple sources', () async {
var controller1 = StreamController<String>();
streamGroup.add(controller1.stream);
controller1.addError('first');
controller1.close();
var controller2 = StreamController<String>();
streamGroup.add(controller2.stream);
controller2.addError('second');
controller2.close();
await flushMicrotasks();
expect(streamGroup.close(), completes);
var transformed = streamGroup.stream.transform(
StreamTransformer<String, String>.fromHandlers(
handleError: (error, _, sink) => sink.add('error: $error')));
expect(transformed.toList(),
completion(equals(['error: first', 'error: second'])));
});
test('buffers events and errors together', () async {
var controller = StreamController<String>();
streamGroup.add(controller.stream);
controller.add('first');
controller.addError('second');
controller.add('third');
controller.addError('fourth');
controller.addError('fifth');
controller.add('sixth');
controller.close();
await flushMicrotasks();
expect(streamGroup.close(), completes);
var transformed = streamGroup.stream.transform(
StreamTransformer<String, String>.fromHandlers(
handleData: (data, sink) => sink.add('data: $data'),
handleError: (error, _, sink) => sink.add('error: $error')));
expect(
transformed.toList(),
completion(equals([
'data: first',
'error: second',
'data: third',
'error: fourth',
'error: fifth',
'data: sixth'
])));
});
test("emits events once there's a listener", () {
var controller = StreamController<String>();
streamGroup.add(controller.stream);
expect(
streamGroup.stream.toList(), completion(equals(['first', 'second'])));
controller.add('first');
controller.add('second');
controller.close();
expect(streamGroup.close(), completes);
});
test("doesn't buffer events from a broadcast stream", () async {
var controller = StreamController<String>.broadcast();
streamGroup.add(controller.stream);
controller.add('first');
controller.add('second');
controller.close();
await flushMicrotasks();
expect(streamGroup.close(), completes);
expect(streamGroup.stream.toList(), completion(isEmpty));
});
test('when paused, buffers events from a broadcast stream', () async {
var controller = StreamController<String>.broadcast();
streamGroup.add(controller.stream);
var events = [];
var subscription = streamGroup.stream.listen(events.add);
subscription.pause();
controller.add('first');
controller.add('second');
controller.close();
await flushMicrotasks();
subscription.resume();
expect(streamGroup.close(), completes);
await flushMicrotasks();
expect(events, equals(['first', 'second']));
});
test("emits events from a broadcast stream once there's a listener", () {
var controller = StreamController<String>.broadcast();
streamGroup.add(controller.stream);
expect(
streamGroup.stream.toList(), completion(equals(['first', 'second'])));
controller.add('first');
controller.add('second');
controller.close();
expect(streamGroup.close(), completes);
});
test('forwards cancel errors', () async {
var subscription = streamGroup.stream.listen(null);
var controller = StreamController<String>(onCancel: () => throw 'error');
streamGroup.add(controller.stream);
await flushMicrotasks();
expect(subscription.cancel(), throwsA('error'));
});
test('forwards a cancel future', () async {
var subscription = streamGroup.stream.listen(null);
var completer = Completer();
var controller =
StreamController<String>(onCancel: () => completer.future);
streamGroup.add(controller.stream);
await flushMicrotasks();
var fired = false;
subscription.cancel().then((_) => fired = true);
await flushMicrotasks();
expect(fired, isFalse);
completer.complete();
await flushMicrotasks();
expect(fired, isTrue);
});
test(
'add() while active pauses the stream if the group is paused, then '
'resumes once the group resumes', () async {
var subscription = streamGroup.stream.listen(null);
await flushMicrotasks();
var paused = false;
var controller = StreamController<String>(
onPause: () => paused = true, onResume: () => paused = false);
subscription.pause();
await flushMicrotasks();
streamGroup.add(controller.stream);
await flushMicrotasks();
expect(paused, isTrue);
subscription.resume();
await flushMicrotasks();
expect(paused, isFalse);
});
group('add() while canceled', () {
setUp(() async {
streamGroup.stream.listen(null).cancel();
await flushMicrotasks();
});
test('immediately listens to and cancels the stream', () async {
var listened = false;
var canceled = false;
var controller = StreamController<String>(onListen: () {
listened = true;
}, onCancel: expectAsync0(() {
expect(listened, isTrue);
canceled = true;
}));
streamGroup.add(controller.stream);
await flushMicrotasks();
expect(listened, isTrue);
expect(canceled, isTrue);
});
test('forwards cancel errors', () {
var controller =
StreamController<String>(onCancel: () => throw 'error');
expect(streamGroup.add(controller.stream), throwsA('error'));
});
test('forwards a cancel future', () async {
var completer = Completer();
var controller =
StreamController<String>(onCancel: () => completer.future);
var fired = false;
streamGroup.add(controller.stream)!.then((_) => fired = true);
await flushMicrotasks();
expect(fired, isFalse);
completer.complete();
await flushMicrotasks();
expect(fired, isTrue);
});
});
});
group('broadcast', () {
late StreamGroup<String> streamGroup;
setUp(() {
streamGroup = StreamGroup<String>.broadcast();
});
test('buffers events from multiple sources', () async {
var controller1 = StreamController<String>();
streamGroup.add(controller1.stream);
controller1.add('first');
controller1.close();
var controller2 = StreamController<String>();
streamGroup.add(controller2.stream);
controller2.add('second');
controller2.close();
await flushMicrotasks();
expect(streamGroup.close(), completes);
expect(
streamGroup.stream.toList(), completion(equals(['first', 'second'])));
});
test("emits events from multiple sources once there's a listener", () {
var controller1 = StreamController<String>();
streamGroup.add(controller1.stream);
var controller2 = StreamController<String>();
streamGroup.add(controller2.stream);
expect(
streamGroup.stream.toList(), completion(equals(['first', 'second'])));
controller1.add('first');
controller2.add('second');
controller1.close();
controller2.close();
expect(streamGroup.close(), completes);
});
test("doesn't buffer events once a listener has been added and removed",
() async {
var controller = StreamController<String>();
streamGroup.add(controller.stream);
streamGroup.stream.listen(null).cancel();
await flushMicrotasks();
controller.add('first');
controller.addError('second');
controller.close();
await flushMicrotasks();
expect(streamGroup.close(), completes);
expect(streamGroup.stream.toList(), completion(isEmpty));
});
test("doesn't buffer events from a broadcast stream", () async {
var controller = StreamController<String>.broadcast();
streamGroup.add(controller.stream);
controller.add('first');
controller.addError('second');
controller.close();
await flushMicrotasks();
expect(streamGroup.close(), completes);
expect(streamGroup.stream.toList(), completion(isEmpty));
});
test("emits events from a broadcast stream once there's a listener", () {
var controller = StreamController<String>.broadcast();
streamGroup.add(controller.stream);
expect(
streamGroup.stream.toList(), completion(equals(['first', 'second'])));
controller.add('first');
controller.add('second');
controller.close();
expect(streamGroup.close(), completes);
});
test('cancels and re-listens broadcast streams', () async {
var subscription = streamGroup.stream.listen(null);
var controller = StreamController<String>.broadcast();
streamGroup.add(controller.stream);
await flushMicrotasks();
expect(controller.hasListener, isTrue);
subscription.cancel();
await flushMicrotasks();
expect(controller.hasListener, isFalse);
streamGroup.stream.listen(null);
await flushMicrotasks();
expect(controller.hasListener, isTrue);
});
test('never cancels single-subscription streams', () async {
var subscription = streamGroup.stream.listen(null);
var controller =
StreamController<String>(onCancel: expectAsync0(() {}, count: 0));
streamGroup.add(controller.stream);
await flushMicrotasks();
subscription.cancel();
await flushMicrotasks();
streamGroup.stream.listen(null);
await flushMicrotasks();
});
test('drops events from a single-subscription stream while dormant',
() async {
var events = [];
var subscription = streamGroup.stream.listen(events.add);
var controller = StreamController<String>();
streamGroup.add(controller.stream);
await flushMicrotasks();
controller.add('first');
await flushMicrotasks();
expect(events, equals(['first']));
subscription.cancel();
controller.add('second');
await flushMicrotasks();
expect(events, equals(['first']));
streamGroup.stream.listen(events.add);
controller.add('third');
await flushMicrotasks();
expect(events, equals(['first', 'third']));
});
test('a single-subscription stream can be removed while dormant', () async {
var controller = StreamController<String>();
streamGroup.add(controller.stream);
await flushMicrotasks();
streamGroup.stream.listen(null).cancel();
await flushMicrotasks();
streamGroup.remove(controller.stream);
expect(controller.hasListener, isFalse);
await flushMicrotasks();
expect(streamGroup.stream.toList(), completion(isEmpty));
controller.add('first');
expect(streamGroup.close(), completes);
});
});
group('regardless of type', () {
group('single-subscription', () {
regardlessOfType(() => StreamGroup<String>());
});
group('broadcast', () {
regardlessOfType(() => StreamGroup<String>.broadcast());
});
});
test('merge() emits events from all components streams', () async {
var controller1 = StreamController<String>();
var controller2 = StreamController<String>();
var merged = StreamGroup.merge([controller1.stream, controller2.stream]);
controller1.add('first');
controller1.close();
controller2.add('second');
controller2.close();
expect(await merged.toList(), ['first', 'second']);
});
test('mergeBroadcast() emits events from all components streams', () async {
var controller1 = StreamController<String>();
var controller2 = StreamController<String>();
var merged =
StreamGroup.mergeBroadcast([controller1.stream, controller2.stream]);
controller1.add('first');
controller1.close();
controller2.add('second');
controller2.close();
expect(merged.isBroadcast, isTrue);
expect(await merged.toList(), ['first', 'second']);
});
}
void regardlessOfType(StreamGroup<String> Function() newStreamGroup) {
late StreamGroup<String> streamGroup;
setUp(() {
streamGroup = newStreamGroup();
});
group('add()', () {
group('while dormant', () {
test("doesn't listen to the stream until the group is listened to",
() async {
var controller = StreamController<String>();
expect(streamGroup.add(controller.stream), isNull);
await flushMicrotasks();
expect(controller.hasListener, isFalse);
streamGroup.stream.listen(null);
await flushMicrotasks();
expect(controller.hasListener, isTrue);
});
test('is a no-op if the stream is already in the group', () {
var controller = StreamController<String>();
streamGroup.add(controller.stream);
streamGroup.add(controller.stream);
streamGroup.add(controller.stream);
// If the stream was actually listened to multiple times, this would
// throw a StateError.
streamGroup.stream.listen(null);
});
});
group('while active', () {
setUp(() async {
streamGroup.stream.listen(null);
await flushMicrotasks();
});
test('listens to the stream immediately', () async {
var controller = StreamController<String>();
expect(streamGroup.add(controller.stream), isNull);
await flushMicrotasks();
expect(controller.hasListener, isTrue);
});
test('is a no-op if the stream is already in the group', () async {
var controller = StreamController<String>();
// If the stream were actually listened to more than once, future
// calls to [add] would throw [StateError]s.
streamGroup.add(controller.stream);
streamGroup.add(controller.stream);
streamGroup.add(controller.stream);
});
});
});
group('remove()', () {
group('while dormant', () {
test("stops emitting events for a stream that's removed", () async {
var controller = StreamController<String>();
streamGroup.add(controller.stream);
expect(streamGroup.stream.toList(), completion(equals(['first'])));
controller.add('first');
await flushMicrotasks();
controller.add('second');
expect(streamGroup.remove(controller.stream), completion(null));
expect(streamGroup.close(), completes);
});
test('is a no-op for an unknown stream', () {
var controller = StreamController<String>();
expect(streamGroup.remove(controller.stream), isNull);
});
test('and closed closes the group when the last stream is removed',
() async {
var controller1 = StreamController<String>();
var controller2 = StreamController<String>();
streamGroup.add(controller1.stream);
streamGroup.add(controller2.stream);
await flushMicrotasks();
streamGroup.close();
streamGroup.remove(controller1.stream);
await flushMicrotasks();
streamGroup.remove(controller2.stream);
await flushMicrotasks();
expect(streamGroup.stream.toList(), completion(isEmpty));
});
});
group('while listening', () {
test("doesn't emit events from a removed stream", () {
var controller = StreamController<String>();
streamGroup.add(controller.stream);
// The subscription to [controller.stream] is canceled synchronously, so
// the first event is dropped even though it was added before the
// removal. This is documented in [StreamGroup.remove].
expect(streamGroup.stream.toList(), completion(isEmpty));
controller.add('first');
expect(streamGroup.remove(controller.stream), completion(null));
controller.add('second');
expect(streamGroup.close(), completes);
});
test("cancels the stream's subscription", () async {
var controller = StreamController<String>();
streamGroup.add(controller.stream);
streamGroup.stream.listen(null);
await flushMicrotasks();
expect(controller.hasListener, isTrue);
streamGroup.remove(controller.stream);
await flushMicrotasks();
expect(controller.hasListener, isFalse);
});
test('forwards cancel errors', () async {
var controller =
StreamController<String>(onCancel: () => throw 'error');
streamGroup.add(controller.stream);
streamGroup.stream.listen(null);
await flushMicrotasks();
expect(streamGroup.remove(controller.stream), throwsA('error'));
});
test('forwards cancel futures', () async {
var completer = Completer();
var controller =
StreamController<String>(onCancel: () => completer.future);
streamGroup.stream.listen(null);
await flushMicrotasks();
streamGroup.add(controller.stream);
await flushMicrotasks();
var fired = false;
streamGroup.remove(controller.stream)!.then((_) => fired = true);
await flushMicrotasks();
expect(fired, isFalse);
completer.complete();
await flushMicrotasks();
expect(fired, isTrue);
});
test('is a no-op for an unknown stream', () async {
var controller = StreamController<String>();
streamGroup.stream.listen(null);
await flushMicrotasks();
expect(streamGroup.remove(controller.stream), isNull);
});
test('and closed closes the group when the last stream is removed',
() async {
var done = false;
streamGroup.stream.listen(null, onDone: () => done = true);
await flushMicrotasks();
var controller1 = StreamController<String>();
var controller2 = StreamController<String>();
streamGroup.add(controller1.stream);
streamGroup.add(controller2.stream);
await flushMicrotasks();
streamGroup.close();
streamGroup.remove(controller1.stream);
await flushMicrotasks();
expect(done, isFalse);
streamGroup.remove(controller2.stream);
await flushMicrotasks();
expect(done, isTrue);
});
});
});
group('close()', () {
group('while dormant', () {
test('if there are no streams, closes the group', () {
expect(streamGroup.close(), completes);
expect(streamGroup.stream.toList(), completion(isEmpty));
});
test(
'if there are streams, closes the group once those streams close '
"and there's a listener", () async {
var controller1 = StreamController<String>();
var controller2 = StreamController<String>();
streamGroup.add(controller1.stream);
streamGroup.add(controller2.stream);
await flushMicrotasks();
streamGroup.close();
controller1.close();
controller2.close();
expect(streamGroup.stream.toList(), completion(isEmpty));
});
});
group('while active', () {
test('if there are no streams, closes the group', () {
expect(streamGroup.stream.toList(), completion(isEmpty));
expect(streamGroup.close(), completes);
});
test('if there are streams, closes the group once those streams close',
() async {
var done = false;
streamGroup.stream.listen(null, onDone: () => done = true);
await flushMicrotasks();
var controller1 = StreamController<String>();
var controller2 = StreamController<String>();
streamGroup.add(controller1.stream);
streamGroup.add(controller2.stream);
await flushMicrotasks();
streamGroup.close();
await flushMicrotasks();
expect(done, isFalse);
controller1.close();
await flushMicrotasks();
expect(done, isFalse);
controller2.close();
await flushMicrotasks();
expect(done, isTrue);
});
});
test('returns a Future that completes once all events are dispatched',
() async {
var events = [];
streamGroup.stream.listen(events.add);
var controller = StreamController<String>();
streamGroup.add(controller.stream);
await flushMicrotasks();
// Add a bunch of events. Each of these will get dispatched in a
// separate microtask, so we can test that [close] only completes once
// all of them have dispatched.
controller.add('one');
controller.add('two');
controller.add('three');
controller.add('four');
controller.add('five');
controller.add('six');
controller.close();
await streamGroup.close();
expect(events, equals(['one', 'two', 'three', 'four', 'five', 'six']));
});
});
}
/// Wait for all microtasks to complete.
Future flushMicrotasks() => Future.delayed(Duration.zero);