Partition

lib/src/partition.dart

@@ -0,0 +1,45 @@
+part of stream_transformers;
+
+/// Splits the events of a stream into 2 partitions.
+/// The predicate acts as a switch to determine in which partition the values go.
+/// If the predicate yields true, the value goes into the first partition.
+/// If the predicate yields false, the value goes into the second partition.
+/// The returned stream is a Map containing both partitions.
+/// The keys of the Map, representing the true and false partitions, can be passed via named arguments.
+/// 
+/// **Example:**
+///
+///     var source = new Stream.fromIterable([1, 2, 3, 4, 5]);
+///     var stream = source.transform(new Partition((int value) => value % 2 == 0, 'evens', 'odds'));
+///     stream.listen(print);
+///
+///     // {evens: null, odds: 1}
+///     // {evens: 2, odds: 1}
+///     // {evens: 2, odds: 3}
+///     // {evens: 4, odds: 3}
+///     // {evens: 4, odds: 5}
+class Partition<T> implements StreamTransformer<T, T> {
+  final Function _predicate;
+  final String _nameWhenTrue, _nameWhenFalse;
+
+  Partition(bool predicate(T value), {String nameWhenTrue: 'predicateTrue', String nameWhenFalse: 'predicateFalse'}) :
+    _predicate = predicate,
+    _nameWhenTrue = nameWhenTrue,
+    _nameWhenFalse = nameWhenFalse;
+
+  Stream<T> bind(Stream<T> stream) {
+
+    return _bindStream(like: stream, onListen: (EventSink<T> sink) {
+      T predicateTrueValue, predicateFalseValue;
+
+      return stream
+        .transform(new FlatMapLatest((T value) {
+          if (_predicate(value)) predicateTrueValue = value;
+          else predicateFalseValue = value;
+
+          return new Stream.fromIterable(<Map<String, T>>[<String, T>{_nameWhenTrue: predicateTrueValue, _nameWhenFalse: predicateFalseValue}]);
+        }))
+        .listen(sink.add, onError: sink.addError, onDone: sink.close);
+    });
+  }
+}

lib/stream_transformers.dart

@@ -21,6 +21,7 @@ part 'src/flat_map.dart';
 part 'src/flat_map_latest.dart';
 part 'src/merge.dart';
 part 'src/merge_all.dart';
+part 'src/partition.dart';
 part 'src/sample_on.dart';
 part 'src/sample_periodically.dart';
 part 'src/select_first.dart';

test/all_tests.dart

@@ -10,6 +10,7 @@ import 'flat_map_test.dart' as flat_map;
 import 'flat_map_latest_test.dart' as flat_map_latest;
 import 'merge_all_test.dart' as merge_all;
 import 'merge_test.dart' as merge;
+import 'partition_test.dart' as partition;
 import 'sample_on_test.dart' as sample_on;
 import 'sample_periodically_test.dart' as sample_periodically;
 import 'scan_test.dart' as scan;
@@ -31,6 +32,7 @@ void main() {
   flat_map_latest.main();
   merge.main();
   merge_all.main();
+  partition.main();
   sample_on.main();
   sample_periodically.main();
   scan.main();

test/partition_test.dart

@@ -0,0 +1,60 @@
+library partition_test;
+
+import 'dart:async';
+import 'package:guinness/guinness.dart';
+import 'package:stream_transformers/stream_transformers.dart';
+import 'util.dart';
+
+void main() => describe("Partition", () {
+  describe("with single subscription stream", () {
+    testWithStreamController(() => new StreamController());
+  });
+
+  describe("with broadcast stream", () {
+    testWithStreamController(() => new StreamController.broadcast());
+  });
+});
+
+void testWithStreamController(StreamController provider()) {
+  StreamController controller;
+
+  beforeEach(() {
+    controller = provider();
+  });
+
+  afterEach(() {
+    controller.close();
+  });
+
+  it("partitions get filled with correct values", () {
+    return testStream(controller.stream.transform(new Partition((int value) => value % 2 == 0, nameWhenTrue: 'evens', nameWhenFalse: 'odds')), // even, odd
+        behavior: () {
+          controller.add(1);
+          controller.add(2);
+          controller.add(3);
+          controller.add(4);
+          controller.add(5);
+        },
+        expectation: (values) => expect(values).toEqual([
+          {'evens': null, 'odds': 1}, // controller.add(1) => evens [] odds [1]
+          {'evens': 2, 'odds': 1},    // controller.add(1) => evens [2] odds [1]
+          {'evens': 2, 'odds': 3},    // controller.add(1) => evens [2] odds [1, 3]
+          {'evens': 4, 'odds': 3},    // controller.add(1) => evens [2, 4] odds [1, 3]
+          {'evens': 4, 'odds': 5}     // controller.add(1) => evens [2, 4] odds [1, 3, 5]
+        ]));
+  });
+
+  it("forwards errors from source and toggle stream", () {
+    return testErrorsAreForwarded(
+        controller.stream.transform(new Partition((int value) => value % 2 == 0)),
+        behavior: () {
+          controller.addError(1);
+        },
+        expectation: (errors) => expect(errors).toEqual([1]));
+  });
+
+  it("returns a stream of the same type", () {
+    var stream = controller.stream.transform(new Partition((int value) => value % 2 == 0));
+    expect(stream.isBroadcast).toBe(controller.stream.isBroadcast);
+  });
+}