Enable and fix strict-raw-types checks (dart-lang/stream_transform#78) Most changes are in tests where we implicitly were using `<dynamic>` but only ever passing a single type. Add specific types and clean up the few cases where a different type from the norm was used. Extract utilities for constructing both types of stream to do the tests in a loop, since it's not using function literals anymore it can be a generic function. In `lib/` add a few `<void>` type arguments for things like "trigger" streams since we were never reading the value.
diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md index 9b8ab29..3f502d3 100644 --- a/pkgs/stream_transform/CHANGELOG.md +++ b/pkgs/stream_transform/CHANGELOG.md
@@ -1,3 +1,8 @@ +## 0.0.18 + +- Internal cleanup. Passed "trigger" streams or futures now allow `<void>` + generic type rather than an implicit `dynamic>` + ## 0.0.17 - Add concrete types to the `onError` callback in `tap`. @@ -47,7 +52,7 @@ - Updates to support Dart 2.0 core library changes (wave 2.2). See [issue 31847][sdk#31847] for details. - + [sdk#31847]: https://github.com/dart-lang/sdk/issues/31847 ## 0.0.9
diff --git a/pkgs/stream_transform/analysis_options.yaml b/pkgs/stream_transform/analysis_options.yaml index 82dd7e8..67baa3e 100644 --- a/pkgs/stream_transform/analysis_options.yaml +++ b/pkgs/stream_transform/analysis_options.yaml
@@ -2,6 +2,8 @@ analyzer: strong-mode: implicit-casts: false + language: + strict-raw-types: true errors: todo: ignore dead_code: error
diff --git a/pkgs/stream_transform/lib/src/async_map_buffer.dart b/pkgs/stream_transform/lib/src/async_map_buffer.dart index fd25a21..a4e346f 100644 --- a/pkgs/stream_transform/lib/src/async_map_buffer.dart +++ b/pkgs/stream_transform/lib/src/async_map_buffer.dart
@@ -28,7 +28,7 @@ /// pending conversions have finished. StreamTransformer<S, T> asyncMapBuffer<S, T>( Future<T> convert(List<S> collected)) { - var workFinished = StreamController() + var workFinished = StreamController<void>() // Let the first event through. ..add(null); return chainTransformers( @@ -39,8 +39,8 @@ /// rather than once per listener, and [then] is called after completing the /// work. StreamTransformer<S, T> _asyncMapThen<S, T>( - Future<T> convert(S event), void then(Object _)) { - Future pendingEvent; + Future<T> convert(S event), void Function(void) then) { + Future<void> pendingEvent; return fromHandlers(handleData: (event, sink) { pendingEvent = convert(event).then(sink.add).catchError(sink.addError).then(then);
diff --git a/pkgs/stream_transform/lib/src/buffer.dart b/pkgs/stream_transform/lib/src/buffer.dart index d09d2da..8641b15 100644 --- a/pkgs/stream_transform/lib/src/buffer.dart +++ b/pkgs/stream_transform/lib/src/buffer.dart
@@ -12,7 +12,8 @@ /// /// Errors from the source stream or the trigger are immediately forwarded to /// the output. -StreamTransformer<T, List<T>> buffer<T>(Stream trigger) => _Buffer<T>(trigger); +StreamTransformer<T, List<T>> buffer<T>(Stream<void> trigger) => + _Buffer<T>(trigger); /// A StreamTransformer which aggregates values and emits when it sees a value /// on [_trigger]. @@ -24,7 +25,7 @@ /// Errors from the source stream or the trigger are immediately forwarded to /// the output. class _Buffer<T> extends StreamTransformerBase<T, List<T>> { - final Stream _trigger; + final Stream<void> _trigger; _Buffer(this._trigger); @@ -38,8 +39,8 @@ var waitingForTrigger = true; var isTriggerDone = false; var isValueDone = false; - StreamSubscription valueSub; - StreamSubscription triggerSub; + StreamSubscription<T> valueSub; + StreamSubscription<void> triggerSub; emit() { controller.add(currentResults); @@ -107,7 +108,7 @@ }; } controller.onCancel = () { - var toCancel = <StreamSubscription>[]; + var toCancel = <StreamSubscription<void>>[]; if (!isValueDone) toCancel.add(valueSub); valueSub = null; if (_trigger.isBroadcast || !values.isBroadcast) {
diff --git a/pkgs/stream_transform/lib/src/combine_latest.dart b/pkgs/stream_transform/lib/src/combine_latest.dart index 1fe7d85..489f344 100644 --- a/pkgs/stream_transform/lib/src/combine_latest.dart +++ b/pkgs/stream_transform/lib/src/combine_latest.dart
@@ -51,8 +51,8 @@ ? _other.asBroadcastStream() : _other; - StreamSubscription sourceSubscription; - StreamSubscription otherSubscription; + StreamSubscription<S> sourceSubscription; + StreamSubscription<T> otherSubscription; var sourceDone = false; var otherDone = false;
diff --git a/pkgs/stream_transform/lib/src/combine_latest_all.dart b/pkgs/stream_transform/lib/src/combine_latest_all.dart index 16f2e80..045ccdc 100644 --- a/pkgs/stream_transform/lib/src/combine_latest_all.dart +++ b/pkgs/stream_transform/lib/src/combine_latest_all.dart
@@ -60,7 +60,7 @@ .toList(); } - List<StreamSubscription> subscriptions; + List<StreamSubscription<T>> subscriptions; controller.onListen = () { assert(subscriptions == null);
diff --git a/pkgs/stream_transform/lib/src/followed_by.dart b/pkgs/stream_transform/lib/src/followed_by.dart index 779831f..3c5c6b3 100644 --- a/pkgs/stream_transform/lib/src/followed_by.dart +++ b/pkgs/stream_transform/lib/src/followed_by.dart
@@ -33,7 +33,7 @@ ? _next.asBroadcastStream() : _next; - StreamSubscription subscription; + StreamSubscription<T> subscription; var currentStream = first; var firstDone = false; var secondDone = false;
diff --git a/pkgs/stream_transform/lib/src/merge.dart b/pkgs/stream_transform/lib/src/merge.dart index a5b4cf3..74a1e1a 100644 --- a/pkgs/stream_transform/lib/src/merge.dart +++ b/pkgs/stream_transform/lib/src/merge.dart
@@ -39,7 +39,7 @@ .toList(); } - List<StreamSubscription> subscriptions; + List<StreamSubscription<T>> subscriptions; controller.onListen = () { assert(subscriptions == null);
diff --git a/pkgs/stream_transform/lib/src/switch.dart b/pkgs/stream_transform/lib/src/switch.dart index 2f24704..e02917d 100644 --- a/pkgs/stream_transform/lib/src/switch.dart +++ b/pkgs/stream_transform/lib/src/switch.dart
@@ -69,7 +69,7 @@ }; } controller.onCancel = () { - var toCancel = <StreamSubscription>[]; + var toCancel = <StreamSubscription<void>>[]; if (!outerStreamDone) toCancel.add(outerSubscription); if (innerSubscription != null) { toCancel.add(innerSubscription);
diff --git a/pkgs/stream_transform/lib/src/take_until.dart b/pkgs/stream_transform/lib/src/take_until.dart index 62b1013..ad12ae8 100644 --- a/pkgs/stream_transform/lib/src/take_until.dart +++ b/pkgs/stream_transform/lib/src/take_until.dart
@@ -10,10 +10,11 @@ /// which are emitted before the trigger, but have further asynchronous delays /// in transformations following the takeUtil, will still go through. Cancelling /// a subscription immediately stops values. -StreamTransformer<T, T> takeUntil<T>(Future trigger) => _TakeUntil(trigger); +StreamTransformer<T, T> takeUntil<T>(Future<void> trigger) => + _TakeUntil(trigger); class _TakeUntil<T> extends StreamTransformerBase<T, T> { - final Future _trigger; + final Future<void> _trigger; _TakeUntil(this._trigger); @@ -23,7 +24,7 @@ ? StreamController<T>.broadcast(sync: true) : StreamController<T>(sync: true); - StreamSubscription subscription; + StreamSubscription<T> subscription; var isDone = false; _trigger.then((_) { if (isDone) return;
diff --git a/pkgs/stream_transform/pubspec.yaml b/pkgs/stream_transform/pubspec.yaml index a9be968..37a6cbf 100644 --- a/pkgs/stream_transform/pubspec.yaml +++ b/pkgs/stream_transform/pubspec.yaml
@@ -2,7 +2,7 @@ description: A collection of utilities to transform and manipulate streams. author: Dart Team <misc@dartlang.org> homepage: https://www.github.com/dart-lang/stream_transform -version: 0.0.17 +version: 0.0.18 environment: sdk: ">=2.2.0 <3.0.0"
diff --git a/pkgs/stream_transform/test/async_map_buffer_test.dart b/pkgs/stream_transform/test/async_map_buffer_test.dart index bf7b9d6..021755a 100644 --- a/pkgs/stream_transform/test/async_map_buffer_test.dart +++ b/pkgs/stream_transform/test/async_map_buffer_test.dart
@@ -8,25 +8,23 @@ import 'package:stream_transform/stream_transform.dart'; +import 'utils.dart'; + void main() { - var streamTypes = { - 'single subscription': () => StreamController(), - 'broadcast': () => StreamController.broadcast() - }; - StreamController values; - List emittedValues; + StreamController<int> values; + List<String> emittedValues; bool valuesCanceled; bool isDone; - List errors; - Stream transformed; - StreamSubscription subscription; + List<String> errors; + Stream<String> transformed; + StreamSubscription<String> subscription; - Completer finishWork; - List workArgument; + Completer<String> finishWork; + List<int> workArgument; /// Represents the async `convert` function and asserts that is is only called /// after the previous iteration has completed. - Future work(List values) { + Future<String> work(List<int> values) { expect(finishWork, isNull, reason: 'See $values befor previous work is complete'); workArgument = values; @@ -41,11 +39,11 @@ return finishWork.future; } - for (var streamType in streamTypes.keys) { + for (var streamType in streamTypes) { group('asyncMapBuffer for stream type: [$streamType]', () { setUp(() { valuesCanceled = false; - values = streamTypes[streamType]() + values = createController(streamType) ..onCancel = () { valuesCanceled = true; }; @@ -66,11 +64,9 @@ await Future(() {}); expect(emittedValues, isEmpty); expect(workArgument, [1]); - finishWork.complete(workArgument); + finishWork.complete('result'); await Future(() {}); - expect(emittedValues, [ - [1] - ]); + expect(emittedValues, ['result']); }); test('buffers values while work is ongoing', () async {
diff --git a/pkgs/stream_transform/test/audit_test.dart b/pkgs/stream_transform/test/audit_test.dart index 71925fd..168227a 100644 --- a/pkgs/stream_transform/test/audit_test.dart +++ b/pkgs/stream_transform/test/audit_test.dart
@@ -10,23 +10,19 @@ import 'utils.dart'; void main() { - var streamTypes = { - 'single subscription': () => StreamController(), - 'broadcast': () => StreamController.broadcast() - }; - for (var streamType in streamTypes.keys) { + for (var streamType in streamTypes) { group('Stream type [$streamType]', () { - StreamController values; - List emittedValues; + StreamController<int> values; + List<int> emittedValues; bool valuesCanceled; bool isDone; - List errors; - Stream transformed; - StreamSubscription subscription; + List<String> errors; + Stream<int> transformed; + StreamSubscription<int> subscription; - void setUpStreams(StreamTransformer transformer) { + void setUpStreams(StreamTransformer<int, int> transformer) { valuesCanceled = false; - values = streamTypes[streamType]() + values = createController(streamType) ..onCancel = () { valuesCanceled = true; };
diff --git a/pkgs/stream_transform/test/buffer_test.dart b/pkgs/stream_transform/test/buffer_test.dart index 2461085..3dcd505 100644 --- a/pkgs/stream_transform/test/buffer_test.dart +++ b/pkgs/stream_transform/test/buffer_test.dart
@@ -7,27 +7,25 @@ import 'package:stream_transform/stream_transform.dart'; +import 'utils.dart'; + void main() { - var streamTypes = { - 'single subscription': () => StreamController(), - 'broadcast': () => StreamController.broadcast() - }; - StreamController trigger; - StreamController values; - List emittedValues; + StreamController<void> trigger; + StreamController<int> values; + List<List<int>> emittedValues; bool valuesCanceled; bool triggerCanceled; bool triggerPaused; bool isDone; - List errors; - Stream transformed; - StreamSubscription subscription; + List<String> errors; + Stream<List<int>> transformed; + StreamSubscription<List<int>> subscription; void setUpForStreamTypes(String triggerType, String valuesType) { valuesCanceled = false; triggerCanceled = false; triggerPaused = false; - trigger = streamTypes[triggerType]() + trigger = createController(triggerType) ..onCancel = () { triggerCanceled = true; }; @@ -36,7 +34,7 @@ triggerPaused = true; }; } - values = streamTypes[valuesType]() + values = createController(valuesType) ..onCancel = () { valuesCanceled = true; }; @@ -50,8 +48,8 @@ }); } - for (var triggerType in streamTypes.keys) { - for (var valuesType in streamTypes.keys) { + for (var triggerType in streamTypes) { + for (var valuesType in streamTypes) { group('Trigger type: [$triggerType], Values type: [$valuesType]', () { setUp(() { setUpForStreamTypes(triggerType, valuesType); @@ -222,7 +220,7 @@ expect(triggerPaused, true); }); - for (var triggerType in streamTypes.keys) { + for (var triggerType in streamTypes) { test('cancel and relisten with [$triggerType] trigger', () async { setUpForStreamTypes(triggerType, 'broadcast'); values.add(1);
diff --git a/pkgs/stream_transform/test/concurrent_async_map_test.dart b/pkgs/stream_transform/test/concurrent_async_map_test.dart index d26ed03..dceaa87 100644 --- a/pkgs/stream_transform/test/concurrent_async_map_test.dart +++ b/pkgs/stream_transform/test/concurrent_async_map_test.dart
@@ -8,34 +8,32 @@ import 'package:stream_transform/stream_transform.dart'; +import 'utils.dart'; + void main() { - var streamTypes = { - 'single subscription': () => StreamController(), - 'broadcast': () => StreamController.broadcast() - }; - StreamController streamController; - List emittedValues; + StreamController<int> controller; + List<String> emittedValues; bool valuesCanceled; bool isDone; - List errors; - Stream transformed; - StreamSubscription subscription; + List<String> errors; + Stream<String> transformed; + StreamSubscription<String> subscription; - List<Completer> finishWork; + List<Completer<String>> finishWork; List<dynamic> values; - Future convert(dynamic value) { + Future<String> convert(int value) { values.add(value); - var completer = Completer(); + var completer = Completer<String>(); finishWork.add(completer); return completer.future; } - for (var streamType in streamTypes.keys) { + for (var streamType in streamTypes) { group('concurrentAsyncMap for stream type: [$streamType]', () { setUp(() { valuesCanceled = false; - streamController = streamTypes[streamType]() + controller = createController(streamType) ..onCancel = () { valuesCanceled = true; }; @@ -44,8 +42,7 @@ isDone = false; finishWork = []; values = []; - transformed = - streamController.stream.transform(concurrentAsyncMap(convert)); + transformed = controller.stream.transform(concurrentAsyncMap(convert)); subscription = transformed .listen(emittedValues.add, onError: errors.add, onDone: () { isDone = true; @@ -53,32 +50,32 @@ }); test('does not emit before convert finishes', () async { - streamController.add(1); + controller.add(1); await Future(() {}); expect(emittedValues, isEmpty); expect(values, [1]); - finishWork.first.complete(1); + finishWork.first.complete('result'); await Future(() {}); - expect(emittedValues, [1]); + expect(emittedValues, ['result']); }); test('allows calls to convert before the last one finished', () async { - streamController..add(1)..add(2)..add(3); + controller..add(1)..add(2)..add(3); await Future(() {}); expect(values, [1, 2, 3]); }); test('forwards errors directly without waiting for previous convert', () async { - streamController.add(1); + controller.add(1); await Future(() {}); - streamController.addError('error'); + controller.addError('error'); await Future(() {}); expect(errors, ['error']); }); test('forwards errors which occur during the convert', () async { - streamController.add(1); + controller.add(1); await Future(() {}); finishWork.first.completeError('error'); await Future(() {}); @@ -86,10 +83,10 @@ }); test('can continue handling events after an error', () async { - streamController.add(1); + controller.add(1); await Future(() {}); finishWork[0].completeError('error'); - streamController.add(2); + controller.add(2); await Future(() {}); expect(values, [1, 2]); finishWork[1].completeError('another'); @@ -105,7 +102,7 @@ test('closes when values end if no conversion is pending', () async { expect(isDone, false); - await streamController.close(); + await controller.close(); await Future(() {}); expect(isDone, true); }); @@ -114,7 +111,7 @@ test('multiple listeners all get values', () async { var otherValues = []; transformed.listen(otherValues.add); - streamController.add(1); + controller.add(1); await Future(() {}); finishWork.first.complete('result'); await Future(() {}); @@ -125,9 +122,9 @@ test('multiple listeners get done when values end', () async { var otherDone = false; transformed.listen(null, onDone: () => otherDone = true); - streamController.add(1); + controller.add(1); await Future(() {}); - await streamController.close(); + await controller.close(); expect(isDone, false); expect(otherDone, false); finishWork.first.complete(); @@ -137,15 +134,15 @@ }); test('can cancel and relisten', () async { - streamController.add(1); + controller.add(1); await Future(() {}); finishWork.first.complete('first'); await Future(() {}); await subscription.cancel(); - streamController.add(2); + controller.add(2); await Future(() {}); subscription = transformed.listen(emittedValues.add); - streamController.add(3); + controller.add(3); await Future(() {}); expect(values, [1, 3]); finishWork[1].complete('second');
diff --git a/pkgs/stream_transform/test/debounce_test.dart b/pkgs/stream_transform/test/debounce_test.dart index aa629db..89fede9 100644 --- a/pkgs/stream_transform/test/debounce_test.dart +++ b/pkgs/stream_transform/test/debounce_test.dart
@@ -10,37 +10,33 @@ import 'utils.dart'; void main() { - var streamTypes = { - 'single subscription': () => StreamController(), - 'broadcast': () => StreamController.broadcast() - }; - for (var streamType in streamTypes.keys) { + for (var streamType in streamTypes) { group('Stream type [$streamType]', () { - StreamController values; - List emittedValues; - bool valuesCanceled; - bool isDone; - List errors; - StreamSubscription subscription; - Stream transformed; - - void setUpStreams(StreamTransformer transformer) { - valuesCanceled = false; - values = streamTypes[streamType]() - ..onCancel = () { - valuesCanceled = true; - }; - emittedValues = []; - errors = []; - isDone = false; - transformed = values.stream.transform(transformer); - subscription = transformed - .listen(emittedValues.add, onError: errors.add, onDone: () { - isDone = true; - }); - } - group('debounce', () { + StreamController<int> values; + List<int> emittedValues; + bool valuesCanceled; + bool isDone; + List<String> errors; + StreamSubscription<int> subscription; + Stream<int> transformed; + + void setUpStreams(StreamTransformer<int, int> transformer) { + valuesCanceled = false; + values = createController(streamType) + ..onCancel = () { + valuesCanceled = true; + }; + emittedValues = []; + errors = []; + isDone = false; + transformed = values.stream.transform(transformer); + subscription = transformed + .listen(emittedValues.add, onError: errors.add, onDone: () { + isDone = true; + }); + } + setUp(() async { setUpStreams(debounce(const Duration(milliseconds: 5))); }); @@ -97,8 +93,18 @@ }); group('debounceBuffer', () { + StreamController<int> values; + List<List<int>> emittedValues; + List<String> errors; + Stream<List<int>> transformed; + setUp(() async { - setUpStreams(debounceBuffer(const Duration(milliseconds: 5))); + values = createController(streamType); + emittedValues = []; + errors = []; + transformed = values.stream + .transform(debounceBuffer(const Duration(milliseconds: 5))) + ..listen(emittedValues.add, onError: errors.add); }); test('Emits all values as a list', () async {
diff --git a/pkgs/stream_transform/test/followd_by_test.dart b/pkgs/stream_transform/test/followd_by_test.dart index 0e8853d..c5f48aa 100644 --- a/pkgs/stream_transform/test/followd_by_test.dart +++ b/pkgs/stream_transform/test/followd_by_test.dart
@@ -7,35 +7,33 @@ import 'package:stream_transform/stream_transform.dart'; -void main() { - var streamTypes = { - 'single subscription': () => StreamController(), - 'broadcast': () => StreamController.broadcast() - }; - for (var firstType in streamTypes.keys) { - for (var secondType in streamTypes.keys) { - group('followedBy [$firstType] with [$secondType]', () { - StreamController first; - StreamController second; +import 'utils.dart'; - List emittedValues; +void main() { + for (var firstType in streamTypes) { + for (var secondType in streamTypes) { + group('followedBy [$firstType] with [$secondType]', () { + StreamController<int> first; + StreamController<int> second; + + List<int> emittedValues; bool firstCanceled; bool secondCanceled; bool secondListened; bool isDone; - List errors; - Stream transformed; - StreamSubscription subscription; + List<String> errors; + Stream<int> transformed; + StreamSubscription<int> subscription; setUp(() async { firstCanceled = false; secondCanceled = false; secondListened = false; - first = streamTypes[firstType]() + first = createController(firstType) ..onCancel = () { firstCanceled = true; }; - second = streamTypes[secondType]() + second = createController(secondType) ..onCancel = () { secondCanceled = true; }
diff --git a/pkgs/stream_transform/test/from_handlers_test.dart b/pkgs/stream_transform/test/from_handlers_test.dart index 1953b29..50d59c5 100644 --- a/pkgs/stream_transform/test/from_handlers_test.dart +++ b/pkgs/stream_transform/test/from_handlers_test.dart
@@ -9,16 +9,16 @@ import 'package:stream_transform/src/from_handlers.dart'; void main() { - StreamController values; - List emittedValues; + StreamController<int> values; + List<int> emittedValues; bool valuesCanceled; bool isDone; - List errors; - Stream transformed; - StreamSubscription subscription; + List<String> errors; + Stream<int> transformed; + StreamSubscription<int> subscription; - void setUpForController( - StreamController controller, StreamTransformer transformer) { + void setUpForController(StreamController<int> controller, + StreamTransformer<int, int> transformer) { valuesCanceled = false; values = controller ..onCancel = () { @@ -68,10 +68,10 @@ }); group('broadcast stream with muliple listeners', () { - List emittedValues2; - List errors2; + List<int> emittedValues2; + List<String> errors2; bool isDone2; - StreamSubscription subscription2; + StreamSubscription<int> subscription2; setUp(() { setUpForController(StreamController.broadcast(), fromHandlers());
diff --git a/pkgs/stream_transform/test/scan_test.dart b/pkgs/stream_transform/test/scan_test.dart index f321539..662831c 100644 --- a/pkgs/stream_transform/test/scan_test.dart +++ b/pkgs/stream_transform/test/scan_test.dart
@@ -58,8 +58,10 @@ .transform(scan<int, Future<int>>(Future.value(0), sum)) .toList(); - expect( - result, [const TypeMatcher<Future>(), const TypeMatcher<Future>()]); + expect(result, [ + const TypeMatcher<Future<void>>(), + const TypeMatcher<Future<void>>() + ]); expect(await Future.wait(result), [1, 3]); });
diff --git a/pkgs/stream_transform/test/start_with_test.dart b/pkgs/stream_transform/test/start_with_test.dart index c02e2a3..8ce8075 100644 --- a/pkgs/stream_transform/test/start_with_test.dart +++ b/pkgs/stream_transform/test/start_with_test.dart
@@ -8,28 +8,27 @@ import 'package:stream_transform/stream_transform.dart'; -void main() { - var streamTypes = { - 'single subscription': () => StreamController(), - 'broadcast': () => StreamController.broadcast() - }; - StreamController values; - Stream transformed; - StreamSubscription subscription; +import 'utils.dart'; - List emittedValues; +void main() { + StreamController<int> values; + Stream<int> transformed; + StreamSubscription<int> subscription; + + List<int> emittedValues; bool isDone; - setupForStreamType(String streamType, StreamTransformer transformer) { + setupForStreamType( + String streamType, StreamTransformer<int, int> transformer) { emittedValues = []; isDone = false; - values = streamTypes[streamType](); + values = createController(streamType); transformed = values.stream.transform(transformer); subscription = transformed.listen(emittedValues.add, onDone: () => isDone = true); } - for (var streamType in streamTypes.keys) { + for (var streamType in streamTypes) { group('startWith then [$streamType]', () { setUp(() => setupForStreamType(streamType, startWith(1))); @@ -101,11 +100,11 @@ } }); - for (var startingStreamType in streamTypes.keys) { + for (var startingStreamType in streamTypes) { group('startWithStream [$startingStreamType] then [$streamType]', () { - StreamController starting; + StreamController<int> starting; setUp(() async { - starting = streamTypes[startingStreamType](); + starting = createController(startingStreamType); setupForStreamType(streamType, startWithStream(starting.stream)); });
diff --git a/pkgs/stream_transform/test/switch_test.dart b/pkgs/stream_transform/test/switch_test.dart index d73e7a8..e32026a 100644 --- a/pkgs/stream_transform/test/switch_test.dart +++ b/pkgs/stream_transform/test/switch_test.dart
@@ -8,42 +8,39 @@ import 'package:stream_transform/stream_transform.dart'; -void main() { - var streamTypes = { - 'single subscription': () => StreamController(), - 'broadcast': () => StreamController.broadcast() - }; - for (var outerType in streamTypes.keys) { - for (var innerType in streamTypes.keys) { - group('Outer type: [$outerType], Inner type: [$innerType]', () { - StreamController first; - StreamController second; - StreamController outer; +import 'utils.dart'; - List emittedValues; +void main() { + for (var outerType in streamTypes) { + for (var innerType in streamTypes) { + group('Outer type: [$outerType], Inner type: [$innerType]', () { + StreamController<int> first; + StreamController<int> second; + StreamController<Stream<int>> outer; + + List<int> emittedValues; bool firstCanceled; bool outerCanceled; bool isDone; - List errors; - StreamSubscription subscription; + List<String> errors; + StreamSubscription<int> subscription; setUp(() async { firstCanceled = false; outerCanceled = false; - outer = streamTypes[outerType]() + outer = createController(outerType) ..onCancel = () { outerCanceled = true; }; - first = streamTypes[innerType]() + first = createController(innerType) ..onCancel = () { firstCanceled = true; }; - second = streamTypes[innerType](); + second = createController(innerType); emittedValues = []; errors = []; isDone = false; subscription = outer.stream - .cast<Stream>() .transform(switchLatest()) .listen(emittedValues.add, onError: errors.add, onDone: () { isDone = true; @@ -129,7 +126,7 @@ group('switchMap', () { test('uses map function', () async { - var outer = StreamController<List>(); + var outer = StreamController<List<int>>(); var values = []; outer.stream
diff --git a/pkgs/stream_transform/test/take_until_test.dart b/pkgs/stream_transform/test/take_until_test.dart index e7614da..74c5167 100644 --- a/pkgs/stream_transform/test/take_until_test.dart +++ b/pkgs/stream_transform/test/take_until_test.dart
@@ -8,25 +8,23 @@ import 'package:stream_transform/stream_transform.dart'; +import 'utils.dart'; + void main() { - var streamTypes = { - 'single subscription': () => StreamController(), - 'broadcast': () => StreamController.broadcast() - }; - for (var streamType in streamTypes.keys) { + for (var streamType in streamTypes) { group('takeUntil on Stream type [$streamType]', () { - StreamController values; - List emittedValues; + StreamController<int> values; + List<int> emittedValues; bool valuesCanceled; bool isDone; - List errors; - Stream transformed; - StreamSubscription subscription; - Completer closeTrigger; + List<String> errors; + Stream<int> transformed; + StreamSubscription<int> subscription; + Completer<void> closeTrigger; setUp(() { valuesCanceled = false; - values = streamTypes[streamType]() + values = createController(streamType) ..onCancel = () { valuesCanceled = true; };
diff --git a/pkgs/stream_transform/test/throttle_test.dart b/pkgs/stream_transform/test/throttle_test.dart index 524f0c0..ae77b7c 100644 --- a/pkgs/stream_transform/test/throttle_test.dart +++ b/pkgs/stream_transform/test/throttle_test.dart
@@ -10,32 +10,25 @@ import 'utils.dart'; void main() { - var streamTypes = { - 'single subscription': () => StreamController(), - 'broadcast': () => StreamController.broadcast() - }; - for (var streamType in streamTypes.keys) { + for (var streamType in streamTypes) { group('Stream type [$streamType]', () { - StreamController values; - List emittedValues; + StreamController<int> values; + List<int> emittedValues; bool valuesCanceled; bool isDone; - List errors; - Stream transformed; - StreamSubscription subscription; + Stream<int> transformed; + StreamSubscription<int> subscription; - void setUpStreams(StreamTransformer transformer) { + void setUpStreams(StreamTransformer<int, int> transformer) { valuesCanceled = false; - values = streamTypes[streamType]() + values = createController(streamType) ..onCancel = () { valuesCanceled = true; }; emittedValues = []; - errors = []; isDone = false; transformed = values.stream.transform(transformer); - subscription = transformed - .listen(emittedValues.add, onError: errors.add, onDone: () { + subscription = transformed.listen(emittedValues.add, onDone: () { isDone = true; }); }
diff --git a/pkgs/stream_transform/test/utils.dart b/pkgs/stream_transform/test/utils.dart index 2bb7679..b6196d6 100644 --- a/pkgs/stream_transform/test/utils.dart +++ b/pkgs/stream_transform/test/utils.dart
@@ -6,6 +6,20 @@ /// Cycle the event loop to ensure timers are started, then wait for a delay /// longer than [milliseconds] to allow for the timer to fire. -Future waitForTimer(int milliseconds) => +Future<void> waitForTimer(int milliseconds) => Future(() {/* ensure Timer is started*/}) .then((_) => Future.delayed(Duration(milliseconds: milliseconds + 1))); + +StreamController<T> createController<T>(String streamType) { + switch (streamType) { + case 'single subscription': + return StreamController<T>(); + case 'broadcast': + return StreamController<T>.broadcast(); + default: + throw ArgumentError.value( + streamType, 'streamType', 'Must be one of $streamTypes'); + } +} + +const streamTypes = ['single subscription', 'broadcast'];