Implement StreamTransformer.fromBind

Closes #32021

Change-Id: I9b8c680ace8b3d7a7bb479ffe19fa0e01fc2cf2f
Reviewed-on: https://dart-review.googlesource.com/71923
Commit-Queue: Nate Bosch <nbosch@google.com>
Reviewed-by: Lasse R.H. Nielsen <lrn@google.com>

CHANGELOG.md

@@ -27,6 +27,10 @@
 *   Fix a bug where calling `stream.take(0).drain(value)` would not correctly
     forward the `value` through the returned `Future`.
 
+#### `dart:async`
+
+*   Add a `StreamTransformer.fromBind` constructor.
+
 ## 2.1.0-dev.3.0
 
 ### Core library changes

sdk/lib/async/stream.dart

@@ -2006,6 +2006,21 @@ abstract class StreamTransformer<S, T> {
       void handleError(Object error, StackTrace stackTrace, EventSink<T> sink),
       void handleDone(EventSink<T> sink)}) = _StreamHandlerTransformer<S, T>;
 
+  /**
+   * Creates a [StreamTransformer] based on a [bind] callback.
+   *
+   * The returned stream transformer uses the [bind] argument to implement the
+   * [StreamTransformer.bind] API and can be used when the transformation is
+   * available as a stream-to-stream function.
+   *
+   * ```dart
+   * final splitDecoded = StreamTransformer<List<int>, String>.fromBind(
+   *     (stream) => stream.transform(utf8.decoder).transform(LineSplitter()));
+   * ```
+   */
+  factory StreamTransformer.fromBind(Stream<T> Function(Stream<S>) bind) =
+      _StreamBindTransformer<S, T>;
+
   /**
    * Adapts [source] to be a `StreamTransfomer<TS, TT>`.
    *

sdk/lib/async/stream_transformers.dart

@@ -282,6 +282,16 @@ class _StreamHandlerTransformer<S, T> extends _StreamSinkTransformer<S, T> {
   }
 }
 
+/**
+ * A StreamTransformer that overrides [StreamTransformer.bind] with a callback.
+ */
+class _StreamBindTransformer<S, T> extends StreamTransformerBase<S, T> {
+  final Stream<T> Function(Stream<S>) _bind;
+  _StreamBindTransformer(this._bind);
+
+  Stream<T> bind(Stream<S> stream) => _bind(stream);
+}
+
 /// A closure mapping a stream and cancelOnError to a StreamSubscription.
 typedef StreamSubscription<T> _SubscriptionTransformer<S, T>(
     Stream<S> stream, bool cancelOnError);

tests/lib_2/async/stream_transformer_from_bind_test.dart

@@ -0,0 +1,25 @@
+// Copyright (c) 2018, 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_helper/async_helper.dart";
+import 'package:expect/expect.dart';
+
+import 'event_helper.dart';
+
+void main() {
+  asyncStart();
+  var transformer =
+      new StreamTransformer<int, String>.fromBind((s) => s.map((v) => '$v'));
+  var controller = new StreamController<int>(sync: true);
+  Events expected = new Events.fromIterable(['1', '2']);
+  Events input = new Events.fromIterable([1, 2]);
+  Events actual = new Events.capture(controller.stream.transform(transformer));
+  actual.onDone(() {
+    Expect.listEquals(expected.events, actual.events);
+    asyncEnd();
+  });
+  input.replay(controller);
+}