Add startWith, startWithMany, and startWithStream (dart-lang/stream_transform#36) Closes dart-lang/stream_transform#19, closes dart-lang/stream_transform#20
diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md index 20f2e9b..93645f7 100644 --- a/pkgs/stream_transform/CHANGELOG.md +++ b/pkgs/stream_transform/CHANGELOG.md
@@ -2,6 +2,7 @@ - Bug Fix: Streams produces with `scan` and `switchMap` now correctly report `isBroadcast`. +- Add `startWith`, `startWithMany`, and `startWithStream`. ## 0.0.6
diff --git a/pkgs/stream_transform/README.md b/pkgs/stream_transform/README.md index bf69198..2b88382 100644 --- a/pkgs/stream_transform/README.md +++ b/pkgs/stream_transform/README.md
@@ -33,6 +33,10 @@ Scan is like fold, but instead of producing a single value it yields each intermediate accumulation. +# startWith, startWithMany, startWithStream + +Prepend a value, an iterable, or a stream to the beginning of another stream. + # switchMap, switchLatest Flatten a Stream of Streams into a Stream which forwards values from the most
diff --git a/pkgs/stream_transform/lib/src/start_with.dart b/pkgs/stream_transform/lib/src/start_with.dart new file mode 100644 index 0000000..a98eb6d --- /dev/null +++ b/pkgs/stream_transform/lib/src/start_with.dart
@@ -0,0 +1,35 @@ +// Copyright (c) 2017, 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 'bind.dart'; +import 'concat.dart'; + +/// Emits [initial] before any values from the original stream. +/// +/// If the original stream is a broadcast stream the result will be as well. +StreamTransformer<T, T> startWith<T>(T initial) => + startWithStream<T>(new Future.value(initial).asStream()); + +/// Emits all values in [initial] before any values from the original stream. +/// +/// If the original stream is a broadcast stream the result will be as well. If +/// the original stream is a broadcast stream it will miss any events which +/// occur before the initial values are all emitted. +StreamTransformer<T, T> startWithMany<T>(Iterable<T> initial) => + startWithStream<T>(new Stream.fromIterable(initial)); + +/// Emits all values in [initial] before any values from the original stream. +/// +/// If the original stream is a broadcast stream the result will be as well. If +/// the original stream is a broadcast stream it will miss any events which +/// occur before [initial] closes. +StreamTransformer<T, T> startWithStream<T>(Stream<T> initial) => + fromBind((values) { + if (values.isBroadcast && !initial.isBroadcast) { + initial = initial.asBroadcastStream(); + } + return initial.transform(concat(values)); + });
diff --git a/pkgs/stream_transform/lib/stream_transform.dart b/pkgs/stream_transform/lib/stream_transform.dart index 8b4d601..4983639 100644 --- a/pkgs/stream_transform/lib/stream_transform.dart +++ b/pkgs/stream_transform/lib/stream_transform.dart
@@ -9,6 +9,7 @@ export 'src/debounce.dart'; export 'src/merge.dart'; export 'src/scan.dart'; +export 'src/start_with.dart'; export 'src/switch.dart'; export 'src/tap.dart'; export 'src/throttle.dart';
diff --git a/pkgs/stream_transform/test/start_with_test.dart b/pkgs/stream_transform/test/start_with_test.dart new file mode 100644 index 0000000..97f0529 --- /dev/null +++ b/pkgs/stream_transform/test/start_with_test.dart
@@ -0,0 +1,156 @@ +// Copyright (c) 2017, 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:test/test.dart'; + +import 'package:stream_transform/stream_transform.dart'; + +void main() { + var streamTypes = { + 'single subscription': () => new StreamController(), + 'broadcast': () => new StreamController.broadcast() + }; + StreamController values; + Stream transformed; + StreamSubscription subscription; + + List emittedValues; + bool isDone; + + setupForStreamType(String streamType, StreamTransformer transformer) { + emittedValues = []; + isDone = false; + values = streamTypes[streamType](); + transformed = values.stream.transform(transformer); + subscription = + transformed.listen(emittedValues.add, onDone: () => isDone = true); + } + + for (var streamType in streamTypes.keys) { + group('startWith then [$streamType]', () { + setUp(() => setupForStreamType(streamType, startWith(1))); + + test('outputs all values', () async { + values..add(2)..add(3); + await new Future(() {}); + expect(emittedValues, [1, 2, 3]); + }); + + test('outputs initial when followed by empty stream', () async { + await values.close(); + expect(emittedValues, [1]); + }); + + test('closes with values', () async { + expect(isDone, false); + await values.close(); + expect(isDone, true); + }); + + if (streamType == 'broadcast') { + test('can cancel and relisten', () async { + values.add(2); + await new Future(() {}); + await subscription.cancel(); + subscription = transformed.listen(emittedValues.add); + values.add(3); + await new Future(() {}); + await new Future(() {}); + expect(emittedValues, [1, 2, 3]); + }); + } + }); + + group('startWithMany then [$streamType]', () { + setUp(() async { + setupForStreamType(streamType, startWithMany([1, 2])); + // Ensure all initial values go through + await new Future(() {}); + }); + + test('outputs all values', () async { + values..add(3)..add(4); + await new Future(() {}); + expect(emittedValues, [1, 2, 3, 4]); + }); + + test('outputs initial when followed by empty stream', () async { + await values.close(); + expect(emittedValues, [1, 2]); + }); + + test('closes with values', () async { + expect(isDone, false); + await values.close(); + expect(isDone, true); + }); + + if (streamType == 'broadcast') { + test('can cancel and relisten', () async { + values.add(3); + await new Future(() {}); + await subscription.cancel(); + subscription = transformed.listen(emittedValues.add); + values.add(4); + await new Future(() {}); + expect(emittedValues, [1, 2, 3, 4]); + }); + } + }); + + for (var startingStreamType in streamTypes.keys) { + group('startWithStream [$startingStreamType] then [$streamType]', () { + StreamController starting; + setUp(() async { + starting = streamTypes[startingStreamType](); + setupForStreamType(streamType, startWithStream(starting.stream)); + }); + + test('outputs all values', () async { + starting..add(1)..add(2); + await starting.close(); + values..add(3)..add(4); + await new Future(() {}); + expect(emittedValues, [1, 2, 3, 4]); + }); + + test('closes with values', () async { + expect(isDone, false); + await starting.close(); + expect(isDone, false); + await values.close(); + expect(isDone, true); + }); + + if (streamType == 'broadcast') { + test('can cancel and relisten during starting', () async { + starting.add(1); + await new Future(() {}); + await subscription.cancel(); + subscription = transformed.listen(emittedValues.add); + starting.add(2); + await starting.close(); + values..add(3)..add(4); + await new Future(() {}); + expect(emittedValues, [1, 2, 3, 4]); + }); + + test('can cancel and relisten during values', () async { + starting..add(1)..add(2); + await starting.close(); + values.add(3); + await new Future(() {}); + await subscription.cancel(); + subscription = transformed.listen(emittedValues.add); + values.add(4); + await new Future(() {}); + expect(emittedValues, [1, 2, 3, 4]); + }); + } + }); + } + } +}