Rewrite transformers as extension methods (dart-lang/stream_transform#83) The usage is much nicer as extension methods. In most cases a small bit of code is copied for the replacement, though in some cases nearly the entire implementation is copied. This was done over extracting shared pieces so that the deprecated methods can be removed outright in the next major version change. Update and clean up some doc comments in the extension method version of these functions. Migrate tests to use the extension method versions. The old copies are now untested but they won't be modified before they are removed. Since there aren't useful needs within this package for `chainTransformers` or `map` they are deprecated without a replacement.

diff --git a/pkgs/stream_transform/.travis.yml b/pkgs/stream_transform/.travis.yml
index 5cdff4c..0bd0f9c 100644
--- a/pkgs/stream_transform/.travis.yml
+++ b/pkgs/stream_transform/.travis.yml

@@ -3,7 +3,7 @@
   only: [master]
 dart:
   - dev
-  - 2.2.0
+  # 2.2.0
 cache:
   directories:
     - $HOME/.pub-cache

diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md
index e631a97..52b1c51 100644
--- a/pkgs/stream_transform/CHANGELOG.md
+++ b/pkgs/stream_transform/CHANGELOG.md

@@ -1,3 +1,25 @@
+## 0.0.20-dev
+
+-   Add extension methods for most transformers. These should be used in place
+    of the current methods. All current implementations are deprecated and will
+    be removed in the next major version bump.
+    -   Migrating typical use: Instead of
+        `stream.transform(debounce(Duration(seconds: 1)))` use
+        `stream.debounce(Duration(seconds: 1))`.
+    -   To migrate a usage where a `StreamTransformer` instance is stored or
+        passed see "Getting a StreamTransformer instance" on the README.
+-   The `map` and `chainTransformers` utilities are no longer useful with the
+    new patterns so they are deprecated without a replacement. If you still have
+    a need for them they can be replicated with `StreamTransformer.fromBind`:
+
+    ```
+    // Replace `map(convert)`
+    StreamTransformer.fromBind((s) => s.map(convert));
+
+    // Replace `chainTransformers(first, second)`
+    StreamTransformer.fromBind((s) => s.transform(first).transform(second));
+    ```
+
 ## 0.0.19
 
 - Add `asyncMapSample` transform.

diff --git a/pkgs/stream_transform/README.md b/pkgs/stream_transform/README.md
index fc44207..313e6f1 100644
--- a/pkgs/stream_transform/README.md
+++ b/pkgs/stream_transform/README.md

@@ -1,84 +1,100 @@
-Utility methods to create `StreamTransfomer` instances to manipulate Streams.
+Extension methods on `Stream` adding common transform operators.
 
-# asyncMapBuffer
+# Operators
+
+## asyncMapBuffer
 
 Like `asyncMap` but events are buffered in a List until previous events have
 been processed rather than being called for each element individually.
 
-# asyncMapSample
+## asyncMapSample
 
 Like `asyncMap` but events are discarded, keeping only the latest, until
 previous events have been processed rather than being called for every element.
 
-# asyncWhere
+## asyncWhere
 
 Like `where` but allows an asynchronous predicate.
 
-# audit
+## audit
 
-Audit waits for a period of time after receiving a value and then only emits
-the most recent value.
+Waits for a period of time after receiving a value and then only emits the most
+recent value.
 
-# buffer
+## buffer
 
 Collects values from a source stream until a `trigger` stream fires and the
 collected values are emitted.
 
-# combineLatest
+## combineLatest
 
 Combine the most recent event from two streams through a callback and emit the
 result.
 
-# combineLatestAll
+## combineLatestAll
 
 Combines the latest events emitted from multiple source streams and yields a
 list of the values.
 
-# debounce, debounceBuffer
+## debounce, debounceBuffer
 
 Prevents a source stream from emitting too frequently by dropping or collecting
 values that occur within a given duration.
 
-# concurrentAsyncMap
+## concurrentAsyncMap
 
 Like `asyncMap` but the convert callback can be called with subsequent values
 before it has finished for previous values.
 
-# followedBy
+## followedBy
 
 Appends the values of a stream after another stream finishes.
 
-# merge, mergeAll
+## merge, mergeAll
 
 Interleaves events from multiple streams into a single stream.
 
-# scan
+## scan
 
 Scan is like fold, but instead of producing a single value it yields each
 intermediate accumulation.
 
-# startWith, startWithMany, startWithStream
+## startWith, startWithMany, startWithStream
 
 Prepend a value, an iterable, or a stream to the beginning of another stream.
 
-# switchMap, switchLatest
+## switchMap, switchLatest
 
 Flatten a Stream of Streams into a Stream which forwards values from the most
 recent Stream
 
-# takeUntil
+## takeUntil
 
 Let values through until a Future fires.
 
-# tap
+## tap
 
 Taps into a single-subscriber stream to react to values as they pass, without
 being a real subscriber.
 
-# throttle
+## throttle
 
 Blocks events for a duration after an event is successfully emitted.
 
-# whereType
+## whereType
 
 Like `Iterable.whereType` for a stream.
+
+# Getting a `StreamTransformer` instance
+
+It may be useful to pass an instance of `StreamTransformer` so that it can be
+used with `stream.transform` calls rather than reference the specific operator
+in place. Any operator on `Stream` that returns a `Stream` can be modeled as a
+`StreamTransformer` using the [`fromBind` constructor][fromBind].
+
+```dart
+final debounce = StreamTransformer.fromBind(
+    (s) => s.debounce(const Duration(milliseconds: 100)));
+```
+
+[fromBind]: https://api.dart.dev/stable/dart-async/StreamTransformer/StreamTransformer.fromBind.html

diff --git a/pkgs/stream_transform/lib/src/async_map_buffer.dart b/pkgs/stream_transform/lib/src/async_map_buffer.dart
index 70c5bdb..57defc2 100644
--- a/pkgs/stream_transform/lib/src/async_map_buffer.dart
+++ b/pkgs/stream_transform/lib/src/async_map_buffer.dart

@@ -9,6 +9,62 @@
 import 'chain_transformers.dart';
 import 'from_handlers.dart';
 
+extension AsyncMap<T> on Stream<T> {
+  /// Like [asyncMap] but events are buffered until previous events have been
+  /// processed by [convert].
+  ///
+  /// If the source stream is a broadcast stream the result will be as well. When
+  /// used with a broadcast stream behavior also differs from [Stream.asyncMap] in
+  /// that the [convert] function is only called once per event, rather than once
+  /// per listener per event.
+  ///
+  /// The first event from the source stream is always passed to [convert] as a
+  /// List with a single element. After that events are buffered until the
+  /// previous Future returned from [convert] has fired.
+  ///
+  /// Errors from the source stream are forwarded directly to the result stream.
+  /// Errors during the conversion are also forwarded to the result stream and
+  /// are considered completing work so the next values are let through.
+  ///
+  /// The result stream will not close until the source stream closes and all
+  /// pending conversions have finished.
+  Stream<S> asyncMapBuffer<S>(Future<S> Function(List<T>) convert) {
+    var workFinished = StreamController<void>()
+      // Let the first event through.
+      ..add(null);
+    return this
+        .buffer(workFinished.stream)
+        .transform(_asyncMapThen(convert, workFinished.add));
+  }
+
+  /// Like [asyncMap] but events are discarded while work is happening in
+  /// [convert].
+  ///
+  /// If the source stream is a broadcast stream the result will be as well. When
+  /// used with a broadcast stream behavior also differs from [Stream.asyncMap] in
+  /// that the [convert] function is only called once per event, rather than once
+  /// per listener per event.
+  ///
+  /// If no work is happening when an event is emitted it will be immediately
+  /// passed to [convert]. If there is ongoing work when an event is emitted it
+  /// will be held until the work is finished. New events emitted will replace a
+  /// pending event.
+  ///
+  /// Errors from the source stream are forwarded directly to the result stream.
+  /// Errors during the conversion are also forwarded to the result stream and are
+  /// considered completing work so the next values are let through.
+  ///
+  /// The result stream will not close until the source stream closes and all
+  /// pending conversions have finished.
+  Stream<S> asyncMapSample<S>(Future<S> Function(T) convert) {
+    var workFinished = StreamController<void>()
+      // Let the first event through.
+      ..add(null);
+    return transform(AggregateSample(workFinished.stream, _dropPrevious))
+        .transform(_asyncMapThen(convert, workFinished.add));
+  }
+}
+
 /// Like [Stream.asyncMap] but events are buffered until previous events have
 /// been processed by [convert].
 ///
@@ -27,6 +83,7 @@
 ///
 /// The result stream will not close until the source stream closes and all
 /// pending conversions have finished.
+@Deprecated('Use the extension instead')
 StreamTransformer<S, T> asyncMapBuffer<S, T>(
     Future<T> Function(List<S>) convert) {
   var workFinished = StreamController<void>()
@@ -55,6 +112,7 @@
 ///
 /// The result stream will not close until the source stream closes and all
 /// pending conversions have finished.
+@Deprecated('Use the extension instead')
 StreamTransformer<S, T> asyncMapSample<S, T>(Future<T> Function(S) convert) {
   var workFinished = StreamController<void>()
     // Let the first event through.

diff --git a/pkgs/stream_transform/lib/src/async_where.dart b/pkgs/stream_transform/lib/src/async_where.dart
index 734f65f..a611063 100644
--- a/pkgs/stream_transform/lib/src/async_where.dart
+++ b/pkgs/stream_transform/lib/src/async_where.dart

@@ -5,6 +5,43 @@
 
 import 'from_handlers.dart';
 
+extension AsyncWhere<T> on Stream<T> {
+  /// Like [where] but allows the [test] to return a [Future].
+  ///
+  /// Events on the result stream will be emitted in the order that [test]
+  /// completes which may not match the order of the original stream.
+  ///
+  /// If the source stream is a broadcast stream the result will be as well. When
+  /// used with a broadcast stream behavior also differs from [Stream.where] in
+  /// that the [test] function is only called once per event, rather than once
+  /// per listener per event.
+  ///
+  /// Errors from the source stream are forwarded directly to the result stream.
+  /// Errors from [test] are also forwarded to the result stream.
+  ///
+  /// The result stream will not close until the source stream closes and all
+  /// pending [test] calls have finished.
+  Stream<T> asyncWhere(FutureOr<bool> test(T element)) {
+    var valuesWaiting = 0;
+    var sourceDone = false;
+    return transform(fromHandlers(handleData: (element, sink) {
+      valuesWaiting++;
+      () async {
+        try {
+          if (await test(element)) sink.add(element);
+        } catch (e, st) {
+          sink.addError(e, st);
+        }
+        valuesWaiting--;
+        if (valuesWaiting <= 0 && sourceDone) sink.close();
+      }();
+    }, handleDone: (sink) {
+      sourceDone = true;
+      if (valuesWaiting <= 0) sink.close();
+    }));
+  }
+}
+
 /// Like [Stream.where] but allows the [test] to return a [Future].
 ///
 /// Events on the result stream will be emitted in the order that [test]
@@ -20,6 +57,7 @@
 ///
 /// The result stream will not close until the source stream closes and all
 /// pending [test] calls have finished.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> asyncWhere<T>(FutureOr<bool> test(T element)) {
   var valuesWaiting = 0;
   var sourceDone = false;

diff --git a/pkgs/stream_transform/lib/src/audit.dart b/pkgs/stream_transform/lib/src/audit.dart
index 7b4d767..b175e6d 100644
--- a/pkgs/stream_transform/lib/src/audit.dart
+++ b/pkgs/stream_transform/lib/src/audit.dart

@@ -5,6 +5,58 @@
 
 import 'from_handlers.dart';
 
+extension Audit<T> on Stream<T> {
+  /// Returns a Stream which only emits once per [duration], at the end of the
+  /// period.
+  ///
+  /// If the source stream is a broadcast stream, the result will be as well.
+  /// Errors are forwarded immediately.
+  ///
+  /// If there is no pending event when the source stream closes the output
+  /// stream will close immediately. If there is a pending event the output
+  /// stream will wait to emit it before closing.
+  ///
+  /// Differs from `throttle` in that it always emits the most recently received
+  /// event rather than the first in the period. The events that are emitted are
+  /// always delayed by some amount. If the event that started the period is the
+  /// one that is emitted it will be delayed by [duration]. If a later event
+  /// comes in within the period it's delay will be shorter by the difference in
+  /// arrival times.
+  ///
+  /// Differs from `debounce` in that a value will always be emitted after
+  /// [duration], the output will not be starved by values coming in repeatedly
+  /// within [duration].
+  ///
+  /// For example:
+  ///
+  ///     source.audit(Duration(seconds: 5));
+  ///
+  ///     source: a------b--c----d--|
+  ///     output: -----a------c--------d|
+  Stream<T> audit(Duration duration) {
+    Timer timer;
+    var shouldClose = false;
+    T recentData;
+
+    return transform(fromHandlers(handleData: (T data, EventSink<T> sink) {
+      recentData = data;
+      timer ??= Timer(duration, () {
+        sink.add(recentData);
+        timer = null;
+        if (shouldClose) {
+          sink.close();
+        }
+      });
+    }, handleDone: (EventSink<T> sink) {
+      if (timer != null) {
+        shouldClose = true;
+      } else {
+        sink.close();
+      }
+    }));
+  }
+}
+
 /// Creates a StreamTransformer which only emits once per [duration], at the
 /// end of the period.
 ///
@@ -16,6 +68,7 @@
 /// Differs from `debounce` in that a value will always be emitted after
 /// [duration], the output will not be starved by values coming in repeatedly
 /// within [duration].
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> audit<T>(Duration duration) {
   Timer timer;
   var shouldClose = false;

diff --git a/pkgs/stream_transform/lib/src/buffer.dart b/pkgs/stream_transform/lib/src/buffer.dart
index 3478a52..cf8b0c3 100644
--- a/pkgs/stream_transform/lib/src/buffer.dart
+++ b/pkgs/stream_transform/lib/src/buffer.dart

@@ -6,6 +6,21 @@
 
 import 'aggregate_sample.dart';
 
+extension Buffer<T> on Stream<T> {
+  /// Returns a Stream  which collects values and emits when it sees a value on
+  /// [trigger].
+  ///
+  /// If there are no pending values when [trigger] emits, the next value on the
+  /// source Stream will immediately flow through. Otherwise, the pending values
+  /// are released when [trigger] emits.
+  ///
+  /// If the source stream is a broadcast stream, the result will be as well.
+  /// Errors from the source stream or the trigger are immediately forwarded to
+  /// the output.
+  Stream<List<T>> buffer(Stream<void> trigger) =>
+      transform(AggregateSample<T, List<T>>(trigger, _collect));
+}
+
 /// Creates a [StreamTransformer] which collects values and emits when it sees a
 /// value on [trigger].
 ///
@@ -15,6 +30,7 @@
 ///
 /// Errors from the source stream or the trigger are immediately forwarded to
 /// the output.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, List<T>> buffer<T>(Stream<void> trigger) =>
     AggregateSample<T, List<T>>(trigger, _collect);
 

diff --git a/pkgs/stream_transform/lib/src/chain_transformers.dart b/pkgs/stream_transform/lib/src/chain_transformers.dart
index 549befc..1e812d9 100644
--- a/pkgs/stream_transform/lib/src/chain_transformers.dart
+++ b/pkgs/stream_transform/lib/src/chain_transformers.dart

@@ -17,6 +17,7 @@
 /// /// values.transform(utf8.decoder).transform(const LineSplitter())
 /// final splitDecoded = chainTransformers(utf8.decoder, const LineSplitter());
 /// ```
+@Deprecated('This utility should not be needed with extension methods')
 StreamTransformer<S, T> chainTransformers<S, I, T>(
         StreamTransformer<S, I> first, StreamTransformer<I, T> second) =>
     StreamTransformer.fromBind(

diff --git a/pkgs/stream_transform/lib/src/combine_latest.dart b/pkgs/stream_transform/lib/src/combine_latest.dart
index 489f344..7fb9ac9 100644
--- a/pkgs/stream_transform/lib/src/combine_latest.dart
+++ b/pkgs/stream_transform/lib/src/combine_latest.dart

@@ -4,6 +4,76 @@
 
 import 'dart:async';
 
+extension CombineLatest<T> on Stream<T> {
+  /// Returns a stream which combines the latest value from the source stream
+  /// with the latest value from [other] using [combine].
+  ///
+  /// No event will be emitted until both the source stream and [other] have
+  /// each emitted at least one event. If either the source stream or [other]
+  /// emit multiple events before the other emits the first event, all but the
+  /// last value will be discarded. Once both streams have emitted at least
+  /// once, the result stream will emit any time either input stream emits.
+  ///
+  /// The result stream will not close until both the source stream and [other]
+  /// have closed.
+  ///
+  /// For example:
+  ///
+  ///     source.combineLatest(other, (a, b) => a + b);
+  ///
+  ///     source: --1--2--------4--|
+  ///     other:  -------3--|
+  ///     result: -------5------7--|
+  ///
+  /// Errors thrown by [combine], along with any errors on the source stream or
+  /// [other], are forwarded to the result stream.
+  ///
+  /// If the source stream is a broadcast stream, the result stream will be as
+  /// well, regardless of [other]'s type. If a single subscription stream is
+  /// combined with a broadcast stream it may never be canceled.
+  Stream<S> combineLatest<T2, S>(
+          Stream<T2> other, FutureOr<S> Function(T, T2) combine) =>
+      transform(_CombineLatest(other, combine));
+
+  /// Combine the latest value emitted from the source stream with the latest
+  /// values emitted from [others].
+  ///
+  /// [combineLatestAll] subscribes to the source stream and [others] and when
+  /// any one of the streams emits, the result stream will emit a [List<T>] of
+  /// the latest values emitted from all streams.
+  ///
+  /// No event will be emitted until all source streams emit at least once. If a
+  /// source stream emits multiple values before another starts emitting, all
+  /// but the last value will be discarded. Once all source streams have emitted
+  /// at least once, the result stream will emit any time any source stream
+  /// emits.
+  ///
+  /// The result stream will not close until all source streams have closed. When
+  /// a source stream closes, the result stream will continue to emit the last
+  /// value from the closed stream when the other source streams emit until the
+  /// result stream has closed. If a source stream closes without emitting any
+  /// value, the result stream will close as well.
+  ///
+  /// For example:
+  ///
+  ///     final combined = first
+  ///         .combineLatestAll([second, third])
+  ///         .map((data) => data.join());
+  ///
+  ///     first:    a----b------------------c--------d---|
+  ///     second:   --1---------2-----------------|
+  ///     third:    -------&----------%---|
+  ///     combined: -------b1&--b2&---b2%---c2%------d2%-|
+  ///
+  /// Errors thrown by any source stream will be forwarded to the result stream.
+  ///
+  /// If the source stream is a broadcast stream, the result stream will be as
+  /// well, regardless of the types of [others]. If a single subscription stream
+  /// is combined with a broadcast source stream, it may never be canceled.
+  Stream<List<T>> combineLatestAll(Iterable<Stream<T>> others) =>
+      transform(_CombineLatestAll<T>(others));
+}
+
 /// Combine the latest value from the source stream with the latest value from
 /// [other] using [combine].
 ///
@@ -31,6 +101,7 @@
 /// If the source stream is a broadcast stream, the result stream will be as
 /// well, regardless of [other]'s type. If a single subscription stream is
 /// combined with a broadcast stream it may never be canceled.
+@Deprecated('Use the extension instead')
 StreamTransformer<S, R> combineLatest<S, T, R>(
         Stream<T> other, FutureOr<R> Function(S, T) combine) =>
     _CombineLatest(other, combine);
@@ -144,3 +215,110 @@
     return controller.stream;
   }
 }
+
+/// Combine the latest value emitted from the source stream with the latest
+/// values emitted from [others].
+///
+/// [combineLatestAll] subscribes to the source stream and [others] and when
+/// any one of the streams emits, the result stream will emit a [List<T>] of
+/// the latest values emitted from all streams.
+///
+/// The result stream will not emit until all source streams emit at least
+/// once. If a source stream emits multiple values before another starts
+/// emitting, all but the last value will be lost.
+///
+/// The result stream will not close until all source streams have closed. When
+/// a source stream closes, the result stream will continue to emit the last
+/// value from the closed stream when the other source streams emit until the
+/// result stream has closed. If a source stream closes without emitting any
+/// value, the result stream will close as well.
+///
+/// Errors thrown by any source stream will be forwarded to the result stream.
+///
+/// If the source stream is a broadcast stream, the result stream will be as
+/// well, regardless of the types of [others]. If a single subscription stream
+/// is combined with a broadcast source stream, it may never be canceled.
+///
+/// ## Example
+///
+/// (Suppose first, second, and third are Stream<String>)
+/// final combined = first
+///     .transform(combineLatestAll([second, third]))
+///     .map((data) => data.join());
+///
+/// first:    a----b------------------c--------d---|
+/// second:   --1---------2-----------------|
+/// third:    -------&----------%---|
+/// combined: -------b1&--b2&---b2%---c2%------d2%-|
+///
+@Deprecated('Use the extension instead')
+StreamTransformer<T, List<T>> combineLatestAll<T>(Iterable<Stream<T>> others) =>
+    _CombineLatestAll<T>(others);
+
+class _CombineLatestAll<T> extends StreamTransformerBase<T, List<T>> {
+  final Iterable<Stream<T>> _others;
+
+  _CombineLatestAll(this._others);
+
+  @override
+  Stream<List<T>> bind(Stream<T> source) {
+    final controller = source.isBroadcast
+        ? StreamController<List<T>>.broadcast(sync: true)
+        : StreamController<List<T>>(sync: true);
+
+    var allStreams = [source]..addAll(_others);
+    if (source.isBroadcast) {
+      allStreams = allStreams
+          .map((s) => s.isBroadcast ? s : s.asBroadcastStream())
+          .toList();
+    }
+
+    List<StreamSubscription<T>> subscriptions;
+
+    controller.onListen = () {
+      assert(subscriptions == null);
+
+      final latestData = List<T>(allStreams.length);
+      final hasEmitted = <int>{};
+      void handleData(int index, T data) {
+        latestData[index] = data;
+        hasEmitted.add(index);
+        if (hasEmitted.length == allStreams.length) {
+          controller.add(List.from(latestData));
+        }
+      }
+
+      var activeStreamCount = 0;
+      subscriptions = allStreams.map((stream) {
+        final index = activeStreamCount;
+        activeStreamCount++;
+        return stream.listen((data) => handleData(index, data),
+            onError: controller.addError, onDone: () {
+          if (--activeStreamCount <= 0 || !hasEmitted.contains(index)) {
+            controller.close();
+          }
+        });
+      }).toList();
+      if (!source.isBroadcast) {
+        controller
+          ..onPause = () {
+            for (var subscription in subscriptions) {
+              subscription.pause();
+            }
+          }
+          ..onResume = () {
+            for (var subscription in subscriptions) {
+              subscription.resume();
+            }
+          };
+      }
+      controller.onCancel = () {
+        final toCancel = subscriptions;
+        subscriptions = null;
+        if (activeStreamCount <= 0) return null;
+        return Future.wait(toCancel.map((s) => s.cancel()));
+      };
+    };
+    return controller.stream;
+  }
+}

diff --git a/pkgs/stream_transform/lib/src/combine_latest_all.dart b/pkgs/stream_transform/lib/src/combine_latest_all.dart
deleted file mode 100644
index 045ccdc..0000000
--- a/pkgs/stream_transform/lib/src/combine_latest_all.dart
+++ /dev/null

@@ -1,111 +0,0 @@
-// Copyright (c) 2019, 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';
-
-/// Combine the latest value emitted from the source stream with the latest
-/// values emitted from [others].
-///
-/// [combineLatestAll] subscribes to the source stream and [others] and when
-/// any one of the streams emits, the result stream will emit a [List<T>] of
-/// the latest values emitted from all streams.
-///
-/// The result stream will not emit until all source streams emit at least
-/// once. If a source stream emits multiple values before another starts
-/// emitting, all but the last value will be lost.
-///
-/// The result stream will not close until all source streams have closed. When
-/// a source stream closes, the result stream will continue to emit the last
-/// value from the closed stream when the other source streams emit until the
-/// result stream has closed. If a source stream closes without emitting any
-/// value, the result stream will close as well.
-///
-/// Errors thrown by any source stream will be forwarded to the result stream.
-///
-/// If the source stream is a broadcast stream, the result stream will be as
-/// well, regardless of the types of [others]. If a single subscription stream
-/// is combined with a broadcast source stream, it may never be canceled.
-///
-/// ## Example
-///
-/// (Suppose first, second, and third are Stream<String>)
-/// final combined = first
-///     .transform(combineLatestAll([second, third]))
-///     .map((data) => data.join());
-///
-/// first:    a----b------------------c--------d---|
-/// second:   --1---------2-----------------|
-/// third:    -------&----------%---|
-/// combined: -------b1&--b2&---b2%---c2%------d2%-|
-///
-StreamTransformer<T, List<T>> combineLatestAll<T>(Iterable<Stream<T>> others) =>
-    _CombineLatestAll<T>(others);
-
-class _CombineLatestAll<T> extends StreamTransformerBase<T, List<T>> {
-  final Iterable<Stream<T>> _others;
-
-  _CombineLatestAll(this._others);
-
-  @override
-  Stream<List<T>> bind(Stream<T> source) {
-    final controller = source.isBroadcast
-        ? StreamController<List<T>>.broadcast(sync: true)
-        : StreamController<List<T>>(sync: true);
-
-    var allStreams = [source]..addAll(_others);
-    if (source.isBroadcast) {
-      allStreams = allStreams
-          .map((s) => s.isBroadcast ? s : s.asBroadcastStream())
-          .toList();
-    }
-
-    List<StreamSubscription<T>> subscriptions;
-
-    controller.onListen = () {
-      assert(subscriptions == null);
-
-      final latestData = List<T>(allStreams.length);
-      final hasEmitted = <int>{};
-      void handleData(int index, T data) {
-        latestData[index] = data;
-        hasEmitted.add(index);
-        if (hasEmitted.length == allStreams.length) {
-          controller.add(List.from(latestData));
-        }
-      }
-
-      var activeStreamCount = 0;
-      subscriptions = allStreams.map((stream) {
-        final index = activeStreamCount;
-        activeStreamCount++;
-        return stream.listen((data) => handleData(index, data),
-            onError: controller.addError, onDone: () {
-          if (--activeStreamCount <= 0 || !hasEmitted.contains(index)) {
-            controller.close();
-          }
-        });
-      }).toList();
-      if (!source.isBroadcast) {
-        controller
-          ..onPause = () {
-            for (var subscription in subscriptions) {
-              subscription.pause();
-            }
-          }
-          ..onResume = () {
-            for (var subscription in subscriptions) {
-              subscription.resume();
-            }
-          };
-      }
-      controller.onCancel = () {
-        final toCancel = subscriptions;
-        subscriptions = null;
-        if (activeStreamCount <= 0) return null;
-        return Future.wait(toCancel.map((s) => s.cancel()));
-      };
-    };
-    return controller.stream;
-  }
-}

diff --git a/pkgs/stream_transform/lib/src/concurrent_async_map.dart b/pkgs/stream_transform/lib/src/concurrent_async_map.dart
index 791c84c..c0fae2f 100644
--- a/pkgs/stream_transform/lib/src/concurrent_async_map.dart
+++ b/pkgs/stream_transform/lib/src/concurrent_async_map.dart

@@ -6,6 +6,45 @@
 
 import 'from_handlers.dart';
 
+extension ConcurrentAsyncMap<T> on Stream<T> {
+  /// Like [asyncMap] but the [convert] callback may be called for an element
+  /// before processing for the previous element is finished.
+  ///
+  /// Events on the result stream will be emitted in the order that [convert]
+  /// completed which may not match the order of the original stream.
+  ///
+  /// If the source stream is a broadcast stream the result will be as well.
+  /// When used with a broadcast stream behavior also differs from [asyncMap] in
+  /// that the [convert] function is only called once per event, rather than
+  /// once per listener per event. The [convert] callback won't be called for
+  /// events while a broadcast stream has no listener.
+  ///
+  /// Errors from [convert] or the source stream are forwarded directly to the
+  /// result stream.
+  ///
+  /// The result stream will not close until the source stream closes and all
+  /// pending conversions have finished.
+  Stream<S> concurrentAsyncMap<S>(FutureOr<S> convert(T event)) {
+    var valuesWaiting = 0;
+    var sourceDone = false;
+    return transform(fromHandlers(handleData: (element, sink) {
+      valuesWaiting++;
+      () async {
+        try {
+          sink.add(await convert(element));
+        } catch (e, st) {
+          sink.addError(e, st);
+        }
+        valuesWaiting--;
+        if (valuesWaiting <= 0 && sourceDone) sink.close();
+      }();
+    }, handleDone: (sink) {
+      sourceDone = true;
+      if (valuesWaiting <= 0) sink.close();
+    }));
+  }
+}
+
 /// Like [Stream.asyncMap] but the [convert] callback may be called for an
 /// element before processing for the previous element is finished.
 ///
@@ -23,6 +62,7 @@
 ///
 /// The result stream will not close until the source stream closes and all
 /// pending conversions have finished.
+@Deprecated('Use the extension instead')
 StreamTransformer<S, T> concurrentAsyncMap<S, T>(FutureOr<T> convert(S event)) {
   var valuesWaiting = 0;
   var sourceDone = false;

diff --git a/pkgs/stream_transform/lib/src/debounce.dart b/pkgs/stream_transform/lib/src/debounce.dart
index ff3f21f..0987ec6 100644
--- a/pkgs/stream_transform/lib/src/debounce.dart
+++ b/pkgs/stream_transform/lib/src/debounce.dart

@@ -5,12 +5,53 @@
 
 import 'from_handlers.dart';
 
+extension Debounce<T> on Stream<T> {
+  /// Returns a Stream which only emits when the source stream does not emit for
+  /// [duration].
+  ///
+  /// Values will always be delayed by at least [duration], and values which
+  /// come within this time will replace the old values, only the most
+  /// recent value will be emitted.
+  ///
+  /// If the source stream is a broadcast stream, the result will be as well.
+  /// Errors are forwarded immediately.
+  ///
+  /// If there is an event waiting during the debounce period when the source
+  /// stream closes the returned stream will wait to emit it following the
+  /// debounce period before closing. If there is no pending debounced event
+  /// when the source stream closes the returned stream will close immediately.
+  ///
+  /// To collect values emitted during the debounce period see [debounceBuffer].
+  Stream<T> debounce(Duration duration) =>
+      transform(_debounceAggregate(duration, _dropPrevious));
+
+  /// Returns a Stream which collects values until the source stream does not
+  /// emit for [duration] then emits the collected values.
+  ///
+  /// Values will always be delayed by at least [duration], and values which
+  /// come within this time will be aggregated into the same list.
+  ///
+  /// If the source stream is a broadcast stream, the result will be as well.
+  /// Errors are forwarded immediately.
+  ///
+  /// If there are events waiting during the debounce period when the source
+  /// stream closes the returned stream will wait to emit them following the
+  /// debounce period before closing. If there are no pending debounced events
+  /// when the source stream closes the returned stream will close immediately.
+  ///
+  /// To keep only the most recent event during the debounce perios see
+  /// [debounce].
+  Stream<List<T>> debounceBuffer(Duration duration) =>
+      transform(_debounceAggregate(duration, _collectToList));
+}
+
 /// Creates a StreamTransformer which only emits when the source stream does not
 /// emit for [duration].
 ///
 /// Source values will always be delayed by at least [duration], and values
 /// which come within this time will replace the old values, only the most
 /// recent value will be emitted.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> debounce<T>(Duration duration) =>
     _debounceAggregate(duration, _dropPrevious);
 
@@ -19,6 +60,7 @@
 ///
 /// This differs from [debounce] in that values are aggregated instead of
 /// skipped.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, List<T>> debounceBuffer<T>(Duration duration) =>
     _debounceAggregate(duration, _collectToList);
 

diff --git a/pkgs/stream_transform/lib/src/followed_by.dart b/pkgs/stream_transform/lib/src/followed_by.dart
index 3c5c6b3..41c7718 100644
--- a/pkgs/stream_transform/lib/src/followed_by.dart
+++ b/pkgs/stream_transform/lib/src/followed_by.dart

@@ -4,6 +4,25 @@
 
 import 'dart:async';
 
+extension FollowedBy<T> on Stream<T> {
+  /// Returns a stream which emits values and errors from [next] after the
+  /// original stream is complete.
+  ///
+  /// If the source stream never finishes, the [next] stream will never be
+  /// listened to.
+  ///
+  /// If the source stream is a broadcast stream, the result will be as well.
+  /// If a single-subscription follows a broadcast stream it may be listened
+  /// to and never canceled since there may be broadcast listeners added later.
+  ///
+  /// If a broadcast stream follows any other stream it will miss any events or
+  /// errors which occur before the first stream is done. If a broadcast stream
+  /// follows a single-subscription stream, pausing the stream while it is
+  /// listening to the second stream will cause events to be dropped rather than
+  /// buffered.
+  Stream<T> followedBy(Stream<T> next) => transform(_FollowedBy(next));
+}
+
 /// Starts emitting values from [next] after the original stream is complete.
 ///
 /// If the initial stream never finishes, the [next] stream will never be
@@ -16,6 +35,7 @@
 /// occur before the first stream is done. If a broadcast stream follows a
 /// single-subscription stream, pausing the stream while it is listening to the
 /// second stream will cause events to be dropped rather than buffered.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> followedBy<T>(Stream<T> next) => _FollowedBy<T>(next);
 
 class _FollowedBy<T> extends StreamTransformerBase<T, T> {

diff --git a/pkgs/stream_transform/lib/src/map.dart b/pkgs/stream_transform/lib/src/map.dart
index 09ea0d6..e1f7137 100644
--- a/pkgs/stream_transform/lib/src/map.dart
+++ b/pkgs/stream_transform/lib/src/map.dart

@@ -16,5 +16,6 @@
 /// final sinkMapper = new StreamSinkTransformer.fromStreamTransformer(
 ///     map((v) => '$v'));
 /// ```
+@Deprecated('This utility should not be needed with extension methods')
 StreamTransformer<S, T> map<S, T>(T convert(S event)) =>
     StreamTransformer.fromBind((stream) => stream.map(convert));

diff --git a/pkgs/stream_transform/lib/src/merge.dart b/pkgs/stream_transform/lib/src/merge.dart
index 74a1e1a..878a8d1 100644
--- a/pkgs/stream_transform/lib/src/merge.dart
+++ b/pkgs/stream_transform/lib/src/merge.dart

@@ -4,11 +4,63 @@
 
 import 'dart:async';
 
+extension Merge<T> on Stream<T> {
+  /// Returns a stream which emits values and errors from the source stream and
+  /// [other] in any order as they arrive.
+  ///
+  /// The result stream will not close until both the source stream and [other]
+  /// have closed.
+  ///
+  /// For example:
+  ///
+  ///    final result = source.merge(other);
+  ///
+  ///    source:  1--2-----3--|
+  ///    other:   ------4-------5--|
+  ///    result:  1--2--4--3----5--|
+  ///
+  /// If the source stream is a broadcast stream, the result stream will be as
+  /// well, regardless of [other]'s type. If a single subscription stream is
+  /// merged into a broadcast stream it may never be canceled since there may be
+  /// broadcast listeners added later.
+  ///
+  /// If a broadcast stream is merged into a single-subscription stream any
+  /// events emitted by [other] before the result stream has a subscriber will
+  /// be discarded.
+  Stream<T> merge(Stream<T> other) => transform(_Merge([other]));
+
+  /// Returns a stream which emits values and errors from the source stream and
+  /// any stream in [others] in any order as they arrive.
+  ///
+  /// The result stream will not close until the source stream and all streams
+  /// in [others] have closed.
+  ///
+  /// For example:
+  ///
+  ///    final result = first.mergeAll([second, third]);
+  ///
+  ///    first:   1--2--------3--|
+  ///    second:  ---------4-------5--|
+  ///    third:   ------6---------------7--|
+  ///    result:  1--2--6--4--3----5----7--|
+  ///
+  /// If the source stream is a broadcast stream, the result stream will be as
+  /// well, regardless the types of streams in [others]. If a single
+  /// subscription stream is merged into a broadcast stream it may never be
+  /// canceled since there may be broadcast listeners added later.
+  ///
+  /// If a broadcast stream is merged into a single-subscription stream any
+  /// events emitted by that stream before the result stream has a subscriber
+  /// will be discarded.
+  Stream<T> mergeAll(Iterable<Stream<T>> others) => transform(_Merge(others));
+}
+
 /// Emits values from the source stream and [other] in any order as they arrive.
 ///
 /// If the source stream is a broadcast stream, the result stream will be as
 /// well, regardless of [other]'s type. If a single subscription stream is
 /// merged into a broadcast stream it may never be canceled.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> merge<T>(Stream<T> other) => _Merge<T>([other]);
 
 /// Emits values from the source stream and all streams in [others] in any order
@@ -18,6 +70,7 @@
 /// well, regardless of the types of streams in [others]. If single
 /// subscription streams are merged into a broadcast stream they may never be
 /// canceled.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> mergeAll<T>(Iterable<Stream<T>> others) =>
     _Merge<T>(others);
 

diff --git a/pkgs/stream_transform/lib/src/scan.dart b/pkgs/stream_transform/lib/src/scan.dart
index b2f76ef..be9899e 100644
--- a/pkgs/stream_transform/lib/src/scan.dart
+++ b/pkgs/stream_transform/lib/src/scan.dart

@@ -4,6 +4,28 @@
 
 import 'dart:async';
 
+extension Scan<T> on Stream<T> {
+  /// Like [fold], but instead of producing a single value it yields each
+  /// intermediate accumulation.
+  ///
+  /// If [combine] returns a Future it will not be called again for subsequent
+  /// events from the source until it completes, therefor the combine callback
+  /// is always called for elements in order, and the result stream always
+  /// maintains the same order as the original.
+  Stream<S> scan<S>(
+      S initialValue, FutureOr<S> combine(S previousValue, T element)) {
+    var accumulated = initialValue;
+    return asyncMap((value) {
+      var result = combine(accumulated, value);
+      if (result is Future<S>) {
+        return result.then((r) => accumulated = r);
+      } else {
+        return accumulated = result as S;
+      }
+    });
+  }
+}
+
 /// Scan is like fold, but instead of producing a single value it yields
 /// each intermediate accumulation.
 ///
@@ -11,6 +33,7 @@
 /// events from the source until it completes, therefor the combine callback is
 /// always called for elements in order, and the result stream always maintains
 /// the same order as the original.
+@Deprecated('Use the extension instead')
 StreamTransformer<S, T> scan<S, T>(
         T initialValue, FutureOr<T> combine(T previousValue, S element)) =>
     StreamTransformer.fromBind((source) {

diff --git a/pkgs/stream_transform/lib/src/start_with.dart b/pkgs/stream_transform/lib/src/start_with.dart
index fbaf46a..a9ecaef 100644
--- a/pkgs/stream_transform/lib/src/start_with.dart
+++ b/pkgs/stream_transform/lib/src/start_with.dart

@@ -6,9 +6,41 @@
 
 import 'followed_by.dart';
 
+extension StartWith<T> on Stream<T> {
+  /// Returns a stream which emits [initial] before any values from the original
+  /// stream.
+  ///
+  /// If the original stream is a broadcast stream the result will be as well.
+  Stream<T> startWith(T initial) =>
+      startWithStream(Future.value(initial).asStream());
+
+  /// Returns a stream which 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.
+  Stream<T> startWithMany(Iterable<T> initial) =>
+      startWithStream(Stream.fromIterable(initial));
+
+  /// Returns a stream which 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.
+  Stream<T> startWithStream(Stream<T> initial) {
+    if (isBroadcast && !initial.isBroadcast) {
+      initial = initial.asBroadcastStream();
+    }
+    return initial.followedBy(this);
+  }
+}
+
 /// Emits [initial] before any values from the original stream.
 ///
 /// If the original stream is a broadcast stream the result will be as well.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> startWith<T>(T initial) =>
     startWithStream<T>(Future.value(initial).asStream());
 
@@ -17,6 +49,7 @@
 /// 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.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> startWithMany<T>(Iterable<T> initial) =>
     startWithStream<T>(Stream.fromIterable(initial));
 
@@ -25,6 +58,7 @@
 /// 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.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> startWithStream<T>(Stream<T> initial) =>
     StreamTransformer.fromBind((values) {
       if (values.isBroadcast && !initial.isBroadcast) {

diff --git a/pkgs/stream_transform/lib/src/switch.dart b/pkgs/stream_transform/lib/src/switch.dart
index e02917d..e4f4787 100644
--- a/pkgs/stream_transform/lib/src/switch.dart
+++ b/pkgs/stream_transform/lib/src/switch.dart

@@ -1,10 +1,33 @@
 // 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 'chain_transformers.dart';
-import 'map.dart';
+extension Switch<T> on Stream<T> {
+  /// Maps events to a Stream and emits values from the most recently created
+  /// Stream.
+  ///
+  /// When the source emits a value it will be converted to a [Stream] using
+  /// [convert] and the output will switch to emitting events from that result.
+  ///
+  /// If the source stream is a broadcast stream, the result stream will be as
+  /// well, regardless of the types of the streams produced by [convert].
+  Stream<S> switchMap<S>(Stream<S> convert(T event)) {
+    return map(convert).switchLatest();
+  }
+}
+
+extension SwitchLatest<T> on Stream<Stream<T>> {
+  /// Emits values from the most recently emitted Stream.
+  ///
+  /// When the source emits a stream the output will switch to emitting events
+  /// from that stream.
+  ///
+  /// If the source stream is a broadcast stream, the result stream will be as
+  /// well, regardless of the types of streams emitted.
+  Stream<T> switchLatest() => transform(_SwitchTransformer<T>());
+}
 
 /// Maps events to a Stream and emits values from the most recently created
 /// Stream.
@@ -14,8 +37,10 @@
 ///
 /// If the source stream is a broadcast stream, the result stream will be as
 /// well, regardless of the types of the streams produced by [map].
+@Deprecated('Use the extension instead')
 StreamTransformer<S, T> switchMap<S, T>(Stream<T> convert(S event)) =>
-    chainTransformers(map(convert), switchLatest());
+    StreamTransformer.fromBind(
+        (source) => source.map(convert).transform(switchLatest()));
 
 /// Emits values from the most recently emitted Stream.
 ///
@@ -24,6 +49,7 @@
 ///
 /// If the source stream is a broadcast stream, the result stream will be as
 /// well, regardless of the types of streams emitted.
+@Deprecated('Use the extension instead')
 StreamTransformer<Stream<T>, T> switchLatest<T>() => _SwitchTransformer<T>();
 
 class _SwitchTransformer<T> extends StreamTransformerBase<Stream<T>, T> {

diff --git a/pkgs/stream_transform/lib/src/take_until.dart b/pkgs/stream_transform/lib/src/take_until.dart
index ad12ae8..d415c24 100644
--- a/pkgs/stream_transform/lib/src/take_until.dart
+++ b/pkgs/stream_transform/lib/src/take_until.dart

@@ -4,12 +4,24 @@
 
 import 'dart:async';
 
+extension TakeUntil<T> on Stream<T> {
+  /// Returns a stram which emits values from the source stream until [trigger]
+  /// fires.
+  ///
+  /// Completing [trigger] differs from canceling a subscription in that values
+  /// 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.
+  Stream<T> takeUntil(Future<void> trigger) => transform(_TakeUntil(trigger));
+}
+
 /// Emits values from the stream until [trigger] fires.
 ///
 /// Completing [trigger] differs from canceling a subscription in that values
 /// 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.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> takeUntil<T>(Future<void> trigger) =>
     _TakeUntil(trigger);
 

diff --git a/pkgs/stream_transform/lib/src/tap.dart b/pkgs/stream_transform/lib/src/tap.dart
index 1b799fa..61ee198 100644
--- a/pkgs/stream_transform/lib/src/tap.dart
+++ b/pkgs/stream_transform/lib/src/tap.dart

@@ -5,6 +5,45 @@
 
 import 'from_handlers.dart';
 
+extension Tap<T> on Stream<T> {
+  /// Taps into this stream to allow additional handling on a single-subscriber
+  /// stream without first wrapping as a broadcast stream.
+  ///
+  /// The [onValue] callback will be called with every value from the source
+  /// stream before it is forwarded to listeners on the resulting stream. May be
+  /// null if only [onError] or [onDone] callbacks are needed.
+  ///
+  /// The [onError] callback will be called with every error from the source
+  /// stream before it is forwarded to listeners on the resulting stream.
+  ///
+  /// The [onDone] callback will be called after the source stream closes and
+  /// before the resulting stream is closed.
+  ///
+  /// Errors from any of the callbacks are caught and ignored.
+  ///
+  /// The callbacks may not be called until the tapped stream has a listener,
+  /// and may not be called after the listener has canceled the subscription.
+  Stream<T> tap(void Function(T) onValue,
+          {void Function(Object, StackTrace) onError,
+          void Function() onDone}) =>
+      transform(fromHandlers(handleData: (value, sink) {
+        try {
+          onValue?.call(value);
+        } catch (_) {/*Ignore*/}
+        sink.add(value);
+      }, handleError: (error, stackTrace, sink) {
+        try {
+          onError?.call(error, stackTrace);
+        } catch (_) {/*Ignore*/}
+        sink.addError(error, stackTrace);
+      }, handleDone: (sink) {
+        try {
+          onDone?.call();
+        } catch (_) {/*Ignore*/}
+        sink.close();
+      }));
+}
+
 /// Taps into a Stream to allow additional handling on a single-subscriber
 /// stream without first wrapping as a broadcast stream.
 ///
@@ -22,6 +61,7 @@
 ///
 /// The callbacks may not be called until the tapped stream has a listener, and
 /// may not be called after the listener has canceled the subscription.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> tap<T>(void Function(T) onValue,
         {void Function(Object, StackTrace) onError, void Function() onDone}) =>
     fromHandlers(handleData: (value, sink) {

diff --git a/pkgs/stream_transform/lib/src/throttle.dart b/pkgs/stream_transform/lib/src/throttle.dart
index 475ad92..bea2980 100644
--- a/pkgs/stream_transform/lib/src/throttle.dart
+++ b/pkgs/stream_transform/lib/src/throttle.dart

@@ -5,8 +5,29 @@
 
 import 'from_handlers.dart';
 
+extension Throttle<T> on Stream<T> {
+  /// Returns a stream which only emits once per [duration], at the beginning of
+  /// the period.
+  ///
+  /// Events emitted by the source stream within [duration] following an emitted
+  /// event will be discarded. Errors are always forwarded immediately.
+  Stream<T> throttle(Duration duration) {
+    Timer timer;
+
+    return transform(fromHandlers(handleData: (data, sink) {
+      if (timer == null) {
+        sink.add(data);
+        timer = Timer(duration, () {
+          timer = null;
+        });
+      }
+    }));
+  }
+}
+
 /// Creates a StreamTransformer which only emits once per [duration], at the
 /// beginning of the period.
+@Deprecated('Use the extension instead')
 StreamTransformer<T, T> throttle<T>(Duration duration) {
   Timer timer;
 

diff --git a/pkgs/stream_transform/lib/src/where_type.dart b/pkgs/stream_transform/lib/src/where_type.dart
index 9164f91..eb79bb5 100644
--- a/pkgs/stream_transform/lib/src/where_type.dart
+++ b/pkgs/stream_transform/lib/src/where_type.dart

@@ -4,6 +4,19 @@
 
 import 'dart:async';
 
+extension WhereType<T> on Stream<T> {
+  /// Returns a stream which emits only the events which have type [S].
+  ///
+  /// If the source stream is a broadcast stream the result will be as well.
+  ///
+  /// Errors from the source stream are forwarded directly to the result stream.
+  ///
+  /// [S] should be a subtype of the stream's generic type, otherwise nothing of
+  /// type [S] could possibly be emitted, however there is no static or runtime
+  /// checking that this is the case.
+  Stream<S> whereType<S>() => transform(_WhereType<S>());
+}
+
 /// Emits only the events which have type [R].
 ///
 /// If the source stream is a broadcast stream the result will be as well.
@@ -19,6 +32,7 @@
 /// [R] should be a subtype of the stream's generic type, otherwise nothing of
 /// type [R] could possibly be emitted, however there is no static or runtime
 /// checking that this is the case.
+@Deprecated('Use the extension instead')
 StreamTransformer<Null, R> whereType<R>() => _WhereType<R>();
 
 class _WhereType<R> extends StreamTransformerBase<Null, R> {

diff --git a/pkgs/stream_transform/lib/stream_transform.dart b/pkgs/stream_transform/lib/stream_transform.dart
index b8b15f4..557fc82 100644
--- a/pkgs/stream_transform/lib/stream_transform.dart
+++ b/pkgs/stream_transform/lib/stream_transform.dart

@@ -8,7 +8,6 @@
 export 'src/buffer.dart';
 export 'src/chain_transformers.dart';
 export 'src/combine_latest.dart';
-export 'src/combine_latest_all.dart';
 export 'src/concat.dart';
 export 'src/concurrent_async_map.dart';
 export 'src/debounce.dart';

diff --git a/pkgs/stream_transform/pubspec.yaml b/pkgs/stream_transform/pubspec.yaml
index 38ad8d5..7542887 100644
--- a/pkgs/stream_transform/pubspec.yaml
+++ b/pkgs/stream_transform/pubspec.yaml

@@ -2,10 +2,10 @@
 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.19
+version: 0.0.20-dev
 
 environment:
-  sdk: ">=2.2.0 <3.0.0"
+  sdk: ">=2.6.0-dev.5.0 <3.0.0"
 
 dev_dependencies:
   pedantic: ^1.5.0

diff --git a/pkgs/stream_transform/test/async_map_buffer_test.dart b/pkgs/stream_transform/test/async_map_buffer_test.dart
index 021755a..9b46e1f 100644
--- a/pkgs/stream_transform/test/async_map_buffer_test.dart
+++ b/pkgs/stream_transform/test/async_map_buffer_test.dart

@@ -52,7 +52,7 @@
         isDone = false;
         finishWork = null;
         workArgument = null;
-        transformed = values.stream.transform(asyncMapBuffer(work));
+        transformed = values.stream.asyncMapBuffer(work);
         subscription = transformed
             .listen(emittedValues.add, onError: errors.add, onDone: () {
           isDone = true;

diff --git a/pkgs/stream_transform/test/async_map_sample_test.dart b/pkgs/stream_transform/test/async_map_sample_test.dart
index 1a6440e..119e0c8 100644
--- a/pkgs/stream_transform/test/async_map_sample_test.dart
+++ b/pkgs/stream_transform/test/async_map_sample_test.dart

@@ -52,7 +52,7 @@
         isDone = false;
         finishWork = null;
         workArgument = null;
-        transformed = values.stream.transform(asyncMapSample(work));
+        transformed = values.stream.asyncMapSample(work);
         subscription = transformed
             .listen(emittedValues.add, onError: errors.add, onDone: () {
           isDone = true;

diff --git a/pkgs/stream_transform/test/async_where_test.dart b/pkgs/stream_transform/test/async_where_test.dart
index 9d0c9f2..a90210c 100644
--- a/pkgs/stream_transform/test/async_where_test.dart
+++ b/pkgs/stream_transform/test/async_where_test.dart

@@ -10,34 +10,34 @@
 void main() {
   test('forwards only events that pass the predicate', () async {
     var values = Stream.fromIterable([1, 2, 3, 4]);
-    var filtered = values.transform(asyncWhere((e) async => e > 2));
+    var filtered = values.asyncWhere((e) async => e > 2);
     expect(await filtered.toList(), [3, 4]);
   });
 
   test('allows predicates that go through event loop', () async {
     var values = Stream.fromIterable([1, 2, 3, 4]);
-    var filtered = values.transform(asyncWhere((e) async {
+    var filtered = values.asyncWhere((e) async {
       await Future(() {});
       return e > 2;
-    }));
+    });
     expect(await filtered.toList(), [3, 4]);
   });
 
   test('allows synchronous predicate', () async {
     var values = Stream.fromIterable([1, 2, 3, 4]);
-    var filtered = values.transform(asyncWhere((e) => e > 2));
+    var filtered = values.asyncWhere((e) => e > 2);
     expect(await filtered.toList(), [3, 4]);
   });
 
   test('can result in empty stream', () async {
     var values = Stream.fromIterable([1, 2, 3, 4]);
-    var filtered = values.transform(asyncWhere((e) => e > 4));
+    var filtered = values.asyncWhere((e) => e > 4);
     expect(await filtered.isEmpty, true);
   });
 
   test('forwards values to multiple listeners', () async {
     var values = StreamController<int>.broadcast();
-    var filtered = values.stream.transform(asyncWhere((e) async => e > 2));
+    var filtered = values.stream.asyncWhere((e) async => e > 2);
     var firstValues = [];
     var secondValues = [];
     filtered..listen(firstValues.add)..listen(secondValues.add);
@@ -50,7 +50,7 @@
   test('closes streams with multiple listeners', () async {
     var values = StreamController.broadcast();
     var predicate = Completer<bool>();
-    var filtered = values.stream.transform(asyncWhere((_) => predicate.future));
+    var filtered = values.stream.asyncWhere((_) => predicate.future);
     var firstDone = false;
     var secondDone = false;
     filtered
@@ -71,11 +71,11 @@
     var errors = [];
     var emitted = [];
     var values = Stream.fromIterable([1, 2, 3, 4]);
-    var filtered = values.transform(asyncWhere((e) async {
+    var filtered = values.asyncWhere((e) async {
       await Future(() {});
       if (e.isEven) throw Exception('$e');
       return true;
-    }));
+    });
     var done = Completer();
     filtered.listen(emitted.add, onError: errors.add, onDone: done.complete);
     await done.future;

diff --git a/pkgs/stream_transform/test/audit_test.dart b/pkgs/stream_transform/test/audit_test.dart
index 168227a..58cb95b 100644
--- a/pkgs/stream_transform/test/audit_test.dart
+++ b/pkgs/stream_transform/test/audit_test.dart

@@ -20,25 +20,21 @@
       Stream<int> transformed;
       StreamSubscription<int> subscription;
 
-      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;
-        });
-      }
-
       group('audit', () {
         setUp(() async {
-          setUpStreams(audit(const Duration(milliseconds: 5)));
+          valuesCanceled = false;
+          values = createController(streamType)
+            ..onCancel = () {
+              valuesCanceled = true;
+            };
+          emittedValues = [];
+          errors = [];
+          isDone = false;
+          transformed = values.stream.audit(const Duration(milliseconds: 5));
+          subscription = transformed
+              .listen(emittedValues.add, onError: errors.add, onDone: () {
+            isDone = true;
+          });
         });
 
         test('cancels values', () async {

diff --git a/pkgs/stream_transform/test/buffer_test.dart b/pkgs/stream_transform/test/buffer_test.dart
index 3dcd505..5800586 100644
--- a/pkgs/stream_transform/test/buffer_test.dart
+++ b/pkgs/stream_transform/test/buffer_test.dart

@@ -41,7 +41,7 @@
     emittedValues = [];
     errors = [];
     isDone = false;
-    transformed = values.stream.transform(buffer(trigger.stream));
+    transformed = values.stream.buffer(trigger.stream);
     subscription =
         transformed.listen(emittedValues.add, onError: errors.add, onDone: () {
       isDone = true;

diff --git a/pkgs/stream_transform/test/combine_latest_all_test.dart b/pkgs/stream_transform/test/combine_latest_all_test.dart
index cdfedb0..2943449 100644
--- a/pkgs/stream_transform/test/combine_latest_all_test.dart
+++ b/pkgs/stream_transform/test/combine_latest_all_test.dart

@@ -16,9 +16,8 @@
       final first = StreamController<String>();
       final second = StreamController<String>();
       final third = StreamController<String>();
-      final combined = first.stream
-          .transform(combineLatestAll([second.stream, third.stream]))
-          .map((data) => data.join());
+      final combined = first.stream.combineLatestAll(
+          [second.stream, third.stream]).map((data) => data.join());
 
       // first:    a----b------------------c--------d---|
       // second:   --1---------2-----------------|
@@ -54,8 +53,7 @@
     test('ends if a Stream closes without ever emitting a value', () async {
       final first = StreamController<String>();
       final second = StreamController<String>();
-      final combined =
-          first.stream.transform(combineLatestAll([second.stream]));
+      final combined = first.stream.combineLatestAll([second.stream]);
 
       // first:    -a------b-------|
       // second:   -----|
@@ -74,8 +72,7 @@
       final first = StreamController<String>();
       final second = StreamController<String>();
       final combined = first.stream
-          .transform(combineLatestAll([second.stream]))
-          .map((data) => data.join());
+          .combineLatestAll([second.stream]).map((data) => data.join());
 
       // first:    -a---------|
       // second:   ----1---#
@@ -95,9 +92,8 @@
       final second = StreamController<String>();
 
       var done = false;
-      first.stream
-          .transform(combineLatestAll([second.stream]))
-          .listen(null, onDone: () => done = true);
+      first.stream.combineLatestAll([second.stream]).listen(null,
+          onDone: () => done = true);
 
       // first:    -a---|
       // second:   --------1--|
@@ -123,8 +119,7 @@
         final first = StreamController<String>.broadcast();
         final second = StreamController<String>.broadcast();
         final combined = first.stream
-            .transform(combineLatestAll([second.stream]))
-            .map((data) => data.join());
+            .combineLatestAll([second.stream]).map((data) => data.join());
 
         // first:    a------b----------------c------d----e---|
         // second:   --1---------2---3---4------5-|

diff --git a/pkgs/stream_transform/test/combine_latest_test.dart b/pkgs/stream_transform/test/combine_latest_test.dart
index f8156bc..2b35e2c 100644
--- a/pkgs/stream_transform/test/combine_latest_test.dart
+++ b/pkgs/stream_transform/test/combine_latest_test.dart

@@ -17,9 +17,8 @@
       int sum(int a, int b) => a + b;
 
       var results = <int>[];
-      unawaited(source.stream
-          .transform(combineLatest(other.stream, sum))
-          .forEach(results.add));
+      unawaited(
+          source.stream.combineLatest(other.stream, sum).forEach(results.add));
 
       source.add(1);
       await Future(() {});
@@ -49,7 +48,7 @@
 
       var results = <String>[];
       unawaited(source.stream
-          .transform(combineLatest(other.stream, times))
+          .combineLatest(other.stream, times)
           .forEach(results.add));
 
       source..add('a')..add('b');
@@ -76,7 +75,7 @@
 
       var done = false;
       source.stream
-          .transform(combineLatest(other.stream, sum))
+          .combineLatest(other.stream, sum)
           .listen(null, onDone: () => done = true);
 
       source.add(1);
@@ -99,7 +98,7 @@
 
       var done = false;
       source
-          .transform(combineLatest(other.stream, sum))
+          .combineLatest(other.stream, sum)
           .listen(null, onDone: () => done = true);
 
       await Future(() {});
@@ -115,7 +114,7 @@
 
       var done = false;
       source.stream
-          .transform(combineLatest(other, sum))
+          .combineLatest(other, sum)
           .listen(null, onDone: () => done = true);
 
       await Future(() {});
@@ -130,7 +129,7 @@
 
       var errors = [];
       source.stream
-          .transform(combineLatest(other.stream, sum))
+          .combineLatest(other.stream, sum)
           .listen(null, onError: errors.add);
 
       source.addError(_NumberedException(1));
@@ -151,8 +150,7 @@
         int combine(int a, int b) => a + b;
 
         var emittedValues = <int>[];
-        var transformed =
-            source.stream.transform(combineLatest(other.stream, combine));
+        var transformed = source.stream.combineLatest(other.stream, combine);
 
         var subscription = transformed.listen(emittedValues.add);
 

diff --git a/pkgs/stream_transform/test/concurrent_async_map_test.dart b/pkgs/stream_transform/test/concurrent_async_map_test.dart
index dceaa87..6da2101 100644
--- a/pkgs/stream_transform/test/concurrent_async_map_test.dart
+++ b/pkgs/stream_transform/test/concurrent_async_map_test.dart

@@ -42,7 +42,7 @@
         isDone = false;
         finishWork = [];
         values = [];
-        transformed = controller.stream.transform(concurrentAsyncMap(convert));
+        transformed = controller.stream.concurrentAsyncMap(convert);
         subscription = transformed
             .listen(emittedValues.add, onError: errors.add, onDone: () {
           isDone = true;

diff --git a/pkgs/stream_transform/test/debounce_test.dart b/pkgs/stream_transform/test/debounce_test.dart
index 89fede9..cb027c5 100644
--- a/pkgs/stream_transform/test/debounce_test.dart
+++ b/pkgs/stream_transform/test/debounce_test.dart

@@ -21,7 +21,7 @@
         StreamSubscription<int> subscription;
         Stream<int> transformed;
 
-        void setUpStreams(StreamTransformer<int, int> transformer) {
+        setUp(() async {
           valuesCanceled = false;
           values = createController(streamType)
             ..onCancel = () {
@@ -30,15 +30,11 @@
           emittedValues = [];
           errors = [];
           isDone = false;
-          transformed = values.stream.transform(transformer);
+          transformed = values.stream.debounce(const Duration(milliseconds: 5));
           subscription = transformed
               .listen(emittedValues.add, onError: errors.add, onDone: () {
             isDone = true;
           });
-        }
-
-        setUp(() async {
-          setUpStreams(debounce(const Duration(milliseconds: 5)));
         });
 
         test('cancels values', () async {
@@ -103,7 +99,7 @@
           emittedValues = [];
           errors = [];
           transformed = values.stream
-              .transform(debounceBuffer(const Duration(milliseconds: 5)))
+              .debounceBuffer(const Duration(milliseconds: 5))
                 ..listen(emittedValues.add, onError: errors.add);
         });
 

diff --git a/pkgs/stream_transform/test/followd_by_test.dart b/pkgs/stream_transform/test/followd_by_test.dart
index c5f48aa..c8864b5 100644
--- a/pkgs/stream_transform/test/followd_by_test.dart
+++ b/pkgs/stream_transform/test/followd_by_test.dart

@@ -43,7 +43,7 @@
           emittedValues = [];
           errors = [];
           isDone = false;
-          transformed = first.stream.transform(followedBy(second.stream));
+          transformed = first.stream.followedBy(second.stream);
           subscription = transformed
               .listen(emittedValues.add, onError: errors.add, onDone: () {
             isDone = true;

diff --git a/pkgs/stream_transform/test/merge_test.dart b/pkgs/stream_transform/test/merge_test.dart
index 4d3050a..24cd76b 100644
--- a/pkgs/stream_transform/test/merge_test.dart
+++ b/pkgs/stream_transform/test/merge_test.dart

@@ -13,7 +13,7 @@
     test('includes all values', () async {
       var first = Stream.fromIterable([1, 2, 3]);
       var second = Stream.fromIterable([4, 5, 6]);
-      var allValues = await first.transform(merge(second)).toList();
+      var allValues = await first.merge(second).toList();
       expect(allValues, containsAllInOrder([1, 2, 3]));
       expect(allValues, containsAllInOrder([4, 5, 6]));
       expect(allValues, hasLength(6));
@@ -30,8 +30,7 @@
         ..onCancel = () {
           secondCanceled = true;
         };
-      var subscription =
-          first.stream.transform(merge(second.stream)).listen((_) {});
+      var subscription = first.stream.merge(second.stream).listen((_) {});
       await subscription.cancel();
       expect(firstCanceled, true);
       expect(secondCanceled, true);
@@ -41,7 +40,7 @@
       var first = StreamController();
       var second = StreamController();
       var isDone = false;
-      first.stream.transform(merge(second.stream)).listen((_) {}, onDone: () {
+      first.stream.merge(second.stream).listen((_) {}, onDone: () {
         isDone = true;
       });
       await first.close();
@@ -54,7 +53,7 @@
       var first = StreamController.broadcast();
       var second = StreamController();
       var emittedValues = [];
-      var transformed = first.stream.transform(merge(second.stream));
+      var transformed = first.stream.merge(second.stream);
       var subscription = transformed.listen(emittedValues.add);
       first.add(1);
       second.add(2);
@@ -77,7 +76,7 @@
       var first = Stream.fromIterable([1, 2, 3]);
       var second = Stream.fromIterable([4, 5, 6]);
       var third = Stream.fromIterable([7, 8, 9]);
-      var allValues = await first.transform(mergeAll([second, third])).toList();
+      var allValues = await first.mergeAll([second, third]).toList();
       expect(allValues, containsAllInOrder([1, 2, 3]));
       expect(allValues, containsAllInOrder([4, 5, 6]));
       expect(allValues, containsAllInOrder([7, 8, 9]));
@@ -101,8 +100,8 @@
           secondSingleCanceled = true;
         };
 
-      var merged = first.stream
-          .transform(mergeAll([secondBroadcast.stream, secondSingle.stream]));
+      var merged =
+          first.stream.mergeAll([secondBroadcast.stream, secondSingle.stream]);
 
       var firstListenerValues = [];
       var secondListenerValues = [];

diff --git a/pkgs/stream_transform/test/scan_test.dart b/pkgs/stream_transform/test/scan_test.dart
index 662831c..4e58680 100644
--- a/pkgs/stream_transform/test/scan_test.dart
+++ b/pkgs/stream_transform/test/scan_test.dart

@@ -14,7 +14,7 @@
     test('produces intermediate values', () async {
       var source = Stream.fromIterable([1, 2, 3, 4]);
       int sum(int x, int y) => x + y;
-      var result = await source.transform(scan(0, sum)).toList();
+      var result = await source.scan(0, sum).toList();
 
       expect(result, [1, 3, 6, 10]);
     });
@@ -22,7 +22,7 @@
     test('can create a broadcast stream', () {
       var source = StreamController.broadcast();
 
-      var transformed = source.stream.transform(scan(null, null));
+      var transformed = source.stream.scan(null, null);
 
       expect(transformed.isBroadcast, true);
     });
@@ -34,7 +34,7 @@
 
       var errors = [];
 
-      source.stream.transform(scan(0, sum)).listen(null, onError: errors.add);
+      source.stream.scan(0, sum).listen(null, onError: errors.add);
 
       source.addError(StateError('fail'));
       await Future(() {});
@@ -46,7 +46,7 @@
       test('returns a Stream of non-futures', () async {
         var source = Stream.fromIterable([1, 2, 3, 4]);
         Future<int> sum(int x, int y) async => x + y;
-        var result = await source.transform(scan(0, sum)).toList();
+        var result = await source.scan(0, sum).toList();
 
         expect(result, [1, 3, 6, 10]);
       });
@@ -54,9 +54,8 @@
       test('can return a Stream of futures when specified', () async {
         var source = Stream.fromIterable([1, 2]);
         Future<int> sum(Future<int> x, int y) async => (await x) + y;
-        var result = await source
-            .transform(scan<int, Future<int>>(Future.value(0), sum))
-            .toList();
+        var result =
+            await source.scan<Future<int>>(Future.value(0), sum).toList();
 
         expect(result, [
           const TypeMatcher<Future<void>>(),
@@ -78,8 +77,7 @@
 
         var results = <int>[];
 
-        unawaited(
-            source.stream.transform(scan(0, combine)).forEach(results.add));
+        unawaited(source.stream.scan(0, combine).forEach(results.add));
 
         source..add(1)..add(2);
         await Future(() {});
@@ -99,9 +97,7 @@
 
         var errors = [];
 
-        source.stream
-            .transform(scan(0, combine))
-            .listen(null, onError: errors.add);
+        source.stream.scan(0, combine).listen(null, onError: errors.add);
 
         source.add(1);
         await Future(() {});

diff --git a/pkgs/stream_transform/test/start_with_test.dart b/pkgs/stream_transform/test/start_with_test.dart
index 8ce8075..3208903 100644
--- a/pkgs/stream_transform/test/start_with_test.dart
+++ b/pkgs/stream_transform/test/start_with_test.dart

@@ -19,18 +19,18 @@
   bool isDone;
 
   setupForStreamType(
-      String streamType, StreamTransformer<int, int> transformer) {
+      String streamType, Stream<int> Function(Stream<int>) transform) {
     emittedValues = [];
     isDone = false;
     values = createController(streamType);
-    transformed = values.stream.transform(transformer);
+    transformed = transform(values.stream);
     subscription =
         transformed.listen(emittedValues.add, onDone: () => isDone = true);
   }
 
   for (var streamType in streamTypes) {
     group('startWith then [$streamType]', () {
-      setUp(() => setupForStreamType(streamType, startWith(1)));
+      setUp(() => setupForStreamType(streamType, (s) => s.startWith(1)));
 
       test('outputs all values', () async {
         values..add(2)..add(3);
@@ -65,7 +65,7 @@
 
     group('startWithMany then [$streamType]', () {
       setUp(() async {
-        setupForStreamType(streamType, startWithMany([1, 2]));
+        setupForStreamType(streamType, (s) => s.startWithMany([1, 2]));
         // Ensure all initial values go through
         await Future(() {});
       });
@@ -105,7 +105,8 @@
         StreamController<int> starting;
         setUp(() async {
           starting = createController(startingStreamType);
-          setupForStreamType(streamType, startWithStream(starting.stream));
+          setupForStreamType(
+              streamType, (s) => s.startWithStream(starting.stream));
         });
 
         test('outputs all values', () async {

diff --git a/pkgs/stream_transform/test/switch_test.dart b/pkgs/stream_transform/test/switch_test.dart
index e32026a..ba26503 100644
--- a/pkgs/stream_transform/test/switch_test.dart
+++ b/pkgs/stream_transform/test/switch_test.dart

@@ -41,7 +41,7 @@
           errors = [];
           isDone = false;
           subscription = outer.stream
-              .transform(switchLatest())
+              .switchLatest()
               .listen(emittedValues.add, onError: errors.add, onDone: () {
             isDone = true;
           });
@@ -129,9 +129,7 @@
       var outer = StreamController<List<int>>();
 
       var values = [];
-      outer.stream
-          .transform(switchMap((l) => Stream.fromIterable(l)))
-          .listen(values.add);
+      outer.stream.switchMap((l) => Stream.fromIterable(l)).listen(values.add);
 
       outer.add([1, 2, 3]);
       await Future(() {});
@@ -143,7 +141,7 @@
     test('can create a broadcast stream', () async {
       var outer = StreamController.broadcast();
 
-      var transformed = outer.stream.transform(switchMap(null));
+      var transformed = outer.stream.switchMap(null);
 
       expect(transformed.isBroadcast, true);
     });

diff --git a/pkgs/stream_transform/test/take_until_test.dart b/pkgs/stream_transform/test/take_until_test.dart
index 74c5167..c25d342 100644
--- a/pkgs/stream_transform/test/take_until_test.dart
+++ b/pkgs/stream_transform/test/take_until_test.dart

@@ -32,7 +32,7 @@
         errors = [];
         isDone = false;
         closeTrigger = Completer();
-        transformed = values.stream.transform(takeUntil(closeTrigger.future));
+        transformed = values.stream.takeUntil(closeTrigger.future);
         subscription = transformed
             .listen(emittedValues.add, onError: errors.add, onDone: () {
           isDone = true;

diff --git a/pkgs/stream_transform/test/tap_test.dart b/pkgs/stream_transform/test/tap_test.dart
index 816c80f..79d47b4 100644
--- a/pkgs/stream_transform/test/tap_test.dart
+++ b/pkgs/stream_transform/test/tap_test.dart

@@ -12,24 +12,22 @@
   test('calls function for values', () async {
     var valuesSeen = [];
     var stream = Stream.fromIterable([1, 2, 3]);
-    await stream.transform(tap(valuesSeen.add)).last;
+    await stream.tap(valuesSeen.add).last;
     expect(valuesSeen, [1, 2, 3]);
   });
 
   test('forwards values', () async {
     var stream = Stream.fromIterable([1, 2, 3]);
-    var values = await stream.transform(tap((_) {})).toList();
+    var values = await stream.tap((_) {}).toList();
     expect(values, [1, 2, 3]);
   });
 
   test('calls function for errors', () async {
     dynamic error;
     var source = StreamController();
-    source.stream
-        .transform(tap((_) {}, onError: (e, st) {
-          error = e;
-        }))
-        .listen((_) {}, onError: (_) {});
+    source.stream.tap((_) {}, onError: (e, st) {
+      error = e;
+    }).listen((_) {}, onError: (_) {});
     source.addError('error');
     await Future(() {});
     expect(error, 'error');
@@ -38,8 +36,7 @@
   test('forwards errors', () async {
     dynamic error;
     var source = StreamController();
-    source.stream.transform(tap((_) {}, onError: (e, st) {})).listen((_) {},
-        onError: (e) {
+    source.stream.tap((_) {}, onError: (e, st) {}).listen((_) {}, onError: (e) {
       error = e;
     });
     source.addError('error');
@@ -50,11 +47,9 @@
   test('calls function on done', () async {
     var doneCalled = false;
     var source = StreamController();
-    source.stream
-        .transform(tap((_) {}, onDone: () {
-          doneCalled = true;
-        }))
-        .listen((_) {});
+    source.stream.tap((_) {}, onDone: () {
+      doneCalled = true;
+    }).listen((_) {});
     await source.close();
     expect(doneCalled, true);
   });
@@ -63,9 +58,9 @@
       () async {
     var dataCallCount = 0;
     var source = StreamController.broadcast();
-    source.stream.transform(tap((_) {
+    source.stream.tap((_) {
       dataCallCount++;
-    }))
+    })
       ..listen((_) {})
       ..listen((_) {});
     source.add(1);
@@ -78,9 +73,9 @@
       () async {
     var errorCallCount = 0;
     var source = StreamController.broadcast();
-    source.stream.transform(tap((_) {}, onError: (_, __) {
+    source.stream.tap((_) {}, onError: (_, __) {
       errorCallCount++;
-    }))
+    })
       ..listen((_) {}, onError: (_, __) {})
       ..listen((_) {}, onError: (_, __) {});
     source.addError('error');
@@ -92,9 +87,9 @@
       () async {
     var doneCallCount = 0;
     var source = StreamController.broadcast();
-    source.stream.transform(tap((_) {}, onDone: () {
+    source.stream.tap((_) {}, onDone: () {
       doneCallCount++;
-    }))
+    })
       ..listen((_) {})
       ..listen((_) {});
     await source.close();
@@ -105,7 +100,7 @@
     var source = StreamController.broadcast();
     var emittedValues1 = [];
     var emittedValues2 = [];
-    source.stream.transform(tap((_) {}))
+    source.stream.tap((_) {})
       ..listen(emittedValues1.add)
       ..listen(emittedValues2.add);
     source.add(1);
@@ -116,6 +111,6 @@
 
   test('allows null callback', () async {
     var stream = Stream.fromIterable([1, 2, 3]);
-    await stream.transform(tap(null)).last;
+    await stream.tap(null).last;
   });
 }

diff --git a/pkgs/stream_transform/test/throttle_test.dart b/pkgs/stream_transform/test/throttle_test.dart
index ae77b7c..e51ef84 100644
--- a/pkgs/stream_transform/test/throttle_test.dart
+++ b/pkgs/stream_transform/test/throttle_test.dart

@@ -19,23 +19,19 @@
       Stream<int> transformed;
       StreamSubscription<int> subscription;
 
-      void setUpStreams(StreamTransformer<int, int> transformer) {
-        valuesCanceled = false;
-        values = createController(streamType)
-          ..onCancel = () {
-            valuesCanceled = true;
-          };
-        emittedValues = [];
-        isDone = false;
-        transformed = values.stream.transform(transformer);
-        subscription = transformed.listen(emittedValues.add, onDone: () {
-          isDone = true;
-        });
-      }
-
       group('throttle', () {
         setUp(() async {
-          setUpStreams(throttle(const Duration(milliseconds: 5)));
+          valuesCanceled = false;
+          values = createController(streamType)
+            ..onCancel = () {
+              valuesCanceled = true;
+            };
+          emittedValues = [];
+          isDone = false;
+          transformed = values.stream.throttle(const Duration(milliseconds: 5));
+          subscription = transformed.listen(emittedValues.add, onDone: () {
+            isDone = true;
+          });
         });
 
         test('cancels values', () async {

diff --git a/pkgs/stream_transform/test/where_type_test.dart b/pkgs/stream_transform/test/where_type_test.dart
index 75062c2..2ea4782 100644
--- a/pkgs/stream_transform/test/where_type_test.dart
+++ b/pkgs/stream_transform/test/where_type_test.dart

@@ -11,19 +11,19 @@
 void main() {
   test('forwards only events that match the type', () async {
     var values = Stream.fromIterable([1, 'a', 2, 'b']);
-    var filtered = values.transform(whereType<String>());
+    var filtered = values.whereType<String>();
     expect(await filtered.toList(), ['a', 'b']);
   });
 
   test('can result in empty stream', () async {
     var values = Stream.fromIterable([1, 2, 3, 4]);
-    var filtered = values.transform(whereType<String>());
+    var filtered = values.whereType<String>();
     expect(await filtered.isEmpty, true);
   });
 
   test('forwards values to multiple listeners', () async {
     var values = StreamController.broadcast();
-    var filtered = values.stream.transform(whereType<String>());
+    var filtered = values.stream.whereType<String>();
     var firstValues = [];
     var secondValues = [];
     filtered..listen(firstValues.add)..listen(secondValues.add);
@@ -35,7 +35,7 @@
 
   test('closes streams with multiple listeners', () async {
     var values = StreamController.broadcast();
-    var filtered = values.stream.transform(whereType<String>());
+    var filtered = values.stream.whereType<String>();
     var firstDone = false;
     var secondDone = false;
     filtered