diff --git a/.github/ISSUE_TEMPLATE/stream_transform.md b/.github/ISSUE_TEMPLATE/stream_transform.md
new file mode 100644
index 0000000..475bd83
--- /dev/null
+++ b/.github/ISSUE_TEMPLATE/stream_transform.md
@@ -0,0 +1,5 @@
+---
+name: "package:stream_transform"
+about: "Create a bug or file a feature request against package:stream_transform."
+labels: "package:stream_transform"
+---
\ No newline at end of file
diff --git a/.github/labeler.yml b/.github/labeler.yml
index c502a11..bfef316 100644
--- a/.github/labeler.yml
+++ b/.github/labeler.yml
@@ -108,6 +108,10 @@
   - changed-files:
     - any-glob-to-any-file: 'pkgs/sse/**'
 
+'package:stream_transform':
+  - changed-files:
+    - any-glob-to-any-file: 'pkgs/stream_transform/**'
+
 'package:term_glyph':
   - changed-files:
     - any-glob-to-any-file: 'pkgs/term_glyph/**'
diff --git a/.github/workflows/clock.yaml b/.github/workflows/clock.yaml
index aef0895..a09a601 100644
--- a/.github/workflows/clock.yaml
+++ b/.github/workflows/clock.yaml
@@ -5,12 +5,12 @@
   push:
     branches: [ main ]
     paths:
-      - '.github/workflows/clock.yml'
+      - '.github/workflows/clock.yaml'
       - 'pkgs/clock/**'
   pull_request:
     branches: [ main ]
     paths:
-      - '.github/workflows/clock.yml'
+      - '.github/workflows/clock.yaml'
       - 'pkgs/clock/**'
   schedule:
     - cron: "0 0 * * 0"
diff --git a/.github/workflows/stream_transform.yaml b/.github/workflows/stream_transform.yaml
new file mode 100644
index 0000000..a36a776
--- /dev/null
+++ b/.github/workflows/stream_transform.yaml
@@ -0,0 +1,73 @@
+name: package:stream_transform
+
+on:
+  # Run on PRs and pushes to the default branch.
+  push:
+    branches: [ main ]
+    paths:
+      - '.github/workflows/stream_transform.yaml'
+      - 'pkgs/stream_transform/**'
+  pull_request:
+    branches: [ main ]
+    paths:
+      - '.github/workflows/stream_transform.yaml'
+      - 'pkgs/stream_transform/**'
+  schedule:
+    - cron: "0 0 * * 0"
+
+env:
+  PUB_ENVIRONMENT: bot.github
+
+
+defaults:
+  run:
+    working-directory: pkgs/stream_transform/
+
+jobs:
+  # Check code formatting and static analysis on a single OS (linux)
+  # against Dart dev.
+  analyze:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        sdk: [dev]
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
+      - uses: dart-lang/setup-dart@e630b99d28a3b71860378cafdc2a067c71107f94
+        with:
+          sdk: ${{ matrix.sdk }}
+      - id: install
+        name: Install dependencies
+        run: dart pub get
+      - name: Check formatting
+        run: dart format --output=none --set-exit-if-changed .
+        if: always() && steps.install.outcome == 'success'
+      - name: Analyze code
+        run: dart analyze --fatal-infos
+        if: always() && steps.install.outcome == 'success'
+
+  # Run tests on a matrix consisting of two dimensions:
+  # 1. OS: ubuntu-latest, (macos-latest, windows-latest)
+  # 2. release channel: dev
+  test:
+    needs: analyze
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        # Add macos-latest and/or windows-latest if relevant for this package.
+        os: [ubuntu-latest]
+        # Bump SDK for Legacy tests when changing min SDK.
+        sdk: [3.1, dev]
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
+      - uses: dart-lang/setup-dart@e630b99d28a3b71860378cafdc2a067c71107f94
+        with:
+          sdk: ${{ matrix.sdk }}
+      - id: install
+        name: Install dependencies
+        run: dart pub get
+      - name: Run tests
+        run: dart test -p chrome,vm --test-randomize-ordering-seed=random
+        if: always() && steps.install.outcome == 'success'
diff --git a/README.md b/README.md
index 7b742de..d1a1d04 100644
--- a/README.md
+++ b/README.md
@@ -40,6 +40,7 @@
 | [source_maps](pkgs/source_maps/) | A library to programmatically manipulate source map files. | [![package issues](https://img.shields.io/badge/package:source_maps-4774bc)](https://github.com/dart-lang/tools/issues?q=is%3Aissue+is%3Aopen+label%3Apackage%3Asource_maps) | [![pub package](https://img.shields.io/pub/v/source_maps.svg)](https://pub.dev/packages/source_maps) |
 | [source_span](pkgs/source_span/) | Provides a standard representation for source code locations and spans. | [![package issues](https://img.shields.io/badge/package:source_span-4774bc)](https://github.com/dart-lang/tools/issues?q=is%3Aissue+is%3Aopen+label%3Apackage%3Asource_span) | [![pub package](https://img.shields.io/pub/v/source_span.svg)](https://pub.dev/packages/source_span) |
 | [sse](pkgs/sse/) | Provides client and server functionality for setting up bi-directional communication through Server Sent Events (SSE) and corresponding POST requests. | [![package issues](https://img.shields.io/badge/package:sse-4774bc)](https://github.com/dart-lang/tools/issues?q=is%3Aissue+is%3Aopen+label%3Apackage%3Asse) | [![pub package](https://img.shields.io/pub/v/sse.svg)](https://pub.dev/packages/sse) |
+| [stream_transform](pkgs/stream_transform/) | A collection of utilities to transform and manipulate streams. | [![package issues](https://img.shields.io/badge/package:stream_transform-4774bc)](https://github.com/dart-lang/tools/issues?q=is%3Aissue+is%3Aopen+label%3Apackage%3Astream_transform) | [![pub package](https://img.shields.io/pub/v/stream_transform.svg)](https://pub.dev/packages/stream_transform) |
 | [term_glyph](pkgs/term_glyph/) | Useful Unicode glyphs and ASCII substitutes. | [![package issues](https://img.shields.io/badge/package:term_glyph-4774bc)](https://github.com/dart-lang/tools/issues?q=is%3Aissue+is%3Aopen+label%3Apackage%3Aterm_glyph) | [![pub package](https://img.shields.io/pub/v/term_glyph.svg)](https://pub.dev/packages/term_glyph) |
 | [test_reflective_loader](pkgs/test_reflective_loader/) | Support for discovering tests and test suites using reflection. | [![package issues](https://img.shields.io/badge/package:test_reflective_loader-4774bc)](https://github.com/dart-lang/tools/issues?q=is%3Aissue+is%3Aopen+label%3Apackage%3Atest_reflective_loader) | [![pub package](https://img.shields.io/pub/v/test_reflective_loader.svg)](https://pub.dev/packages/test_reflective_loader) |
 | [timing](pkgs/timing/) | A simple package for tracking the performance of synchronous and asynchronous actions. | [![package issues](https://img.shields.io/badge/package:timing-4774bc)](https://github.com/dart-lang/tools/issues?q=is%3Aissue+is%3Aopen+label%3Apackage%3Atiming) | [![pub package](https://img.shields.io/pub/v/timing.svg)](https://pub.dev/packages/timing) |
diff --git a/pkgs/stream_transform/.gitignore b/pkgs/stream_transform/.gitignore
new file mode 100644
index 0000000..bfffcc6
--- /dev/null
+++ b/pkgs/stream_transform/.gitignore
@@ -0,0 +1,6 @@
+.pub/
+.dart_tool/
+build/
+packages
+pubspec.lock
+.packages
diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md
new file mode 100644
index 0000000..a71b2fb
--- /dev/null
+++ b/pkgs/stream_transform/CHANGELOG.md
@@ -0,0 +1,185 @@
+## 2.1.1
+
+- Require Dart 3.1 or greater
+- Forward errors from the `trigger` future through to the result stream in
+  `takeUntil`. Previously an error would have not closed the stream, and instead
+  raised as an unhandled async error.
+- Move to `dart-lang/tools` monorepo.
+
+## 2.1.0
+
+- Add `whereNotNull`.
+
+## 2.0.1
+
+- Require Dart 2.14 or greater.
+- Wait for the future returned from `StreamSubscription.cancel()` before
+  listening to the subsequent stream in `switchLatest` and `switchMap`.
+
+## 2.0.0
+
+- Migrate to null safety.
+- Improve tests of `switchMap` and improve documentation with links and
+  clarification.
+- Add `trailing` argument to `throttle`.
+
+## 1.2.0
+
+-  Add support for emitting the "leading" event in `debounce`.
+
+## 1.1.1
+
+-   Fix a bug in `asyncMapSample`, `buffer`, `combineLatest`,
+    `combineLatestAll`, `merge`, and `mergeAll` which would cause an exception
+    when cancelling a subscription after using the transformer if the original
+    stream(s) returned `null` from cancelling their subscriptions.
+
+## 1.1.0
+
+-   Add `concurrentAsyncExpand` to interleave events emitted by multiple sub
+    streams created by a callback.
+
+## 1.0.0
+
+-   Remove the top level methods and retain the extensions only.
+
+## 0.0.20
+
+-   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.
+
+## 0.0.18
+
+- Internal cleanup. Passed "trigger" streams or futures now allow `<void>`
+  generic type rather than an implicit `dynamic>`
+
+## 0.0.17
+
+- Add concrete types to the `onError` callback in `tap`.
+
+## 0.0.16+1
+
+- Remove usage of Set literal which is not available before Dart 2.2.0
+
+## 0.0.16
+
+- Allow a `combine` callback to return a `FutureOr<T>` in `scan`. There are no
+  behavior changes for synchronous callbacks. **Potential breaking change** In
+  the unlikely situation where `scan` was used to produce a `Stream<Future>`
+  inference may now fail and require explicit generic type arguments.
+- Add `combineLatest`.
+- Add `combineLatestAll`.
+
+## 0.0.15
+
+- Add `whereType`.
+
+## 0.0.14+1
+
+- Allow using non-dev Dart 2 SDK.
+
+## 0.0.14
+
+- `asyncWhere` will now forward exceptions thrown by the callback through the
+  result Stream.
+- Added `concurrentAsyncMap`.
+
+## 0.0.13
+
+- `mergeAll` now accepts an `Iterable<Stream>` instead of only `List<Stream>`.
+
+## 0.0.12
+
+- Add `chainTransformers` and `map` for use cases where `StreamTransformer`
+  instances are stored as variables or passed to methods other than `transform`.
+
+## 0.0.11
+
+- Renamed `concat` as `followedBy` to match the naming of `Iterable.followedBy`.
+  `concat` is now deprecated.
+
+## 0.0.10
+
+- Updates to support Dart 2.0 core library changes (wave
+  2.2). See [issue 31847][sdk#31847] for details.
+
+  [sdk#31847]: https://github.com/dart-lang/sdk/issues/31847
+
+## 0.0.9
+
+- Add `asyncMapBuffer`.
+
+## 0.0.8
+
+- Add `takeUntil`.
+
+## 0.0.7
+
+- Bug Fix: Streams produced with `scan` and `switchMap` now correctly report
+  `isBroadcast`.
+- Add `startWith`, `startWithMany`, and `startWithStream`.
+
+## 0.0.6
+
+- Bug Fix: Some transformers did not correctly add data to all listeners on
+  broadcast streams. Fixed for `throttle`, `debounce`, `asyncWhere` and `audit`.
+- Bug Fix: Only call the `tap` data callback once per event rather than once per
+  listener.
+- Bug Fix: Allow canceling and re-listening to broadcast streams after a
+  `merge` transform.
+- Bug Fix: Broadcast streams which are buffered using a single-subscription
+  trigger can be canceled and re-listened.
+- Bug Fix: Buffer outputs one more value if there is a pending trigger before
+  the trigger closes.
+- Bug Fix: Single-subscription streams concatted after broadcast streams are
+  handled correctly.
+- Use sync `StreamControllers` for forwarding where possible.
+
+## 0.0.5
+
+- Bug Fix: Allow compiling switchLatest with Dart2Js.
+- Add `asyncWhere`: Like `where` but allows an asynchronous predicate.
+
+## 0.0.4
+- Add `scan`: fold which returns intermediate values
+- Add `throttle`: block events for a duration after emitting a value
+- Add `audit`: emits the last event received after a duration
+
+## 0.0.3
+
+- Add `tap`: React to values as they pass without being a subscriber on a stream
+- Add `switchMap` and `switchLatest`: Flatten a Stream of Streams into a Stream
+  which forwards values from the most recent Stream
+
+## 0.0.2
+
+- Add `concat`: Appends streams in series
+- Add `merge` and `mergeAll`: Interleaves streams
+
+## 0.0.1
+
+- Initial release with the following utilities:
+  - `buffer`: Collects events in a `List` until a `trigger` stream fires.
+  - `debounce`, `debounceBuffer`: Collect or drop events which occur closer in
+    time than a given duration.
diff --git a/pkgs/stream_transform/LICENSE b/pkgs/stream_transform/LICENSE
new file mode 100644
index 0000000..03af64a
--- /dev/null
+++ b/pkgs/stream_transform/LICENSE
@@ -0,0 +1,27 @@
+Copyright 2017, the Dart project authors. 
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+    * Redistributions of source code must retain the above copyright
+      notice, this list of conditions and the following disclaimer.
+    * Redistributions in binary form must reproduce the above
+      copyright notice, this list of conditions and the following
+      disclaimer in the documentation and/or other materials provided
+      with the distribution.
+    * Neither the name of Google LLC nor the names of its
+      contributors may be used to endorse or promote products derived
+      from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git a/pkgs/stream_transform/README.md b/pkgs/stream_transform/README.md
new file mode 100644
index 0000000..e7049bd
--- /dev/null
+++ b/pkgs/stream_transform/README.md
@@ -0,0 +1,141 @@
+[![Build Status](https://github.com/dart-lang/tools/actions/workflows/stream_transform.yaml/badge.svg)](https://github.com/dart-lang/tools/actions/workflows/stream_transform.yaml)
+[![pub package](https://img.shields.io/pub/v/stream_transform.svg)](https://pub.dev/packages/stream_transform)
+[![package publisher](https://img.shields.io/pub/publisher/stream_transform.svg)](https://pub.dev/packages/stream_transform/publisher)
+
+Extension methods on `Stream` adding common transform operators.
+
+## Operators
+
+### asyncMapBuffer, asyncMapSample, concurrentAsyncMap
+
+Alternatives to `asyncMap`. `asyncMapBuffer` prevents the callback from
+overlapping execution and collects events while it is executing.
+`asyncMapSample` prevents overlapping execution and discards events while it is
+executing. `concurrentAsyncMap` allows overlap and removes ordering guarantees
+for higher throughput.
+
+Like `asyncMap` but events are buffered in a List until previous events have
+been processed rather than being called for each element individually.
+
+### asyncWhere
+
+Like `where` but allows an asynchronous predicate.
+
+### audit
+
+Waits for a period of time after receiving a value and then only emits the most
+recent value.
+
+### buffer
+
+Collects values from a source stream until a `trigger` stream fires and the
+collected values are emitted.
+
+### combineLatest, combineLatestAll
+
+Combine the most recent event from multiple streams through a callback or into a
+list.
+
+### debounce, debounceBuffer
+
+Prevents a source stream from emitting too frequently by dropping or collecting
+values that occur within a given duration.
+
+### followedBy
+
+Appends the values of a stream after another stream finishes.
+
+### merge, mergeAll, concurrentAsyncExpand
+
+Interleaves events from multiple streams into a single stream.
+
+### scan
+
+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
+recent Stream
+
+### takeUntil
+
+Let values through until a Future fires.
+
+### tap
+
+Taps into a single-subscriber stream to react to values as they pass, without
+being a real subscriber.
+
+### throttle
+
+Blocks events for a duration after an event is successfully emitted.
+
+### whereType
+
+Like `Iterable.whereType` for a stream.
+
+## Comparison to Rx Operators
+
+The semantics and naming in this package have some overlap, and some conflict,
+with the [ReactiveX](https://reactivex.io/) suite of libraries. Some of the
+conflict is intentional - Dart `Stream` predates `Observable` and coherence with
+the Dart ecosystem semantics and naming is a strictly higher priority than
+consistency with ReactiveX.
+
+Rx Operator Category      | variation                                              | `stream_transform`
+------------------------- | ------------------------------------------------------ | ------------------
+[`sample`][rx_sample]     | `sample/throttleLast(Duration)`                        | `sample(Stream.periodic(Duration), longPoll: false)`
+&#x200B;                  | `throttleFirst(Duration)`                              | [`throttle`][throttle]
+&#x200B;                  | `sample(Observable)`                                   | `sample(trigger, longPoll: false)`
+[`debounce`][rx_debounce] | `debounce/throttleWithTimeout(Duration)`               | [`debounce`][debounce]
+&#x200B;                  | `debounce(Observable)`                                 | No equivalent
+[`buffer`][rx_buffer]     | `buffer(boundary)`, `bufferWithTime`,`bufferWithCount` | No equivalent
+&#x200B;                  | `buffer(boundaryClosingSelector)`                      | `buffer(trigger, longPoll: false)`
+RxJs extensions           | [`audit(callback)`][rxjs_audit]                        | No equivalent
+&#x200B;                  | [`auditTime(Duration)`][rxjs_auditTime]                | [`audit`][audit]
+&#x200B;                  | [`exhaustMap`][rxjs_exhaustMap]                        | No equivalent
+&#x200B;                  | [`throttleTime(trailing: true)`][rxjs_throttleTime]    | `throttle(trailing: true)`
+&#x200B;                  | `throttleTime(leading: false, trailing: true)`         | No equivalent
+No equivalent?            |                                                        | [`asyncMapBuffer`][asyncMapBuffer]
+&#x200B;                  |                                                        | [`asyncMapSample`][asyncMapSample]
+&#x200B;                  |                                                        | [`buffer`][buffer]
+&#x200B;                  |                                                        | [`sample`][sample]
+&#x200B;                  |                                                        | [`debounceBuffer`][debounceBuffer]
+&#x200B;                  |                                                        | `debounce(leading: true, trailing: false)`
+&#x200B;                  |                                                        | `debounce(leading: true, trailing: true)`
+
+[rx_sample]:https://reactivex.io/documentation/operators/sample.html
+[rx_debounce]:https://reactivex.io/documentation/operators/debounce.html
+[rx_buffer]:https://reactivex.io/documentation/operators/buffer.html
+[rxjs_audit]:https://rxjs.dev/api/operators/audit
+[rxjs_auditTime]:https://rxjs.dev/api/operators/auditTime
+[rxjs_throttleTime]:https://rxjs.dev/api/operators/throttleTime
+[rxjs_exhaustMap]:https://rxjs.dev/api/operators/exhaustMap
+[asyncMapBuffer]:https://pub.dev/documentation/stream_transform/latest/stream_transform/AsyncMap/asyncMapBuffer.html
+[asyncMapSample]:https://pub.dev/documentation/stream_transform/latest/stream_transform/AsyncMap/asyncMapSample.html
+[audit]:https://pub.dev/documentation/stream_transform/latest/stream_transform/RateLimit/audit.html
+[buffer]:https://pub.dev/documentation/stream_transform/latest/stream_transform/RateLimit/buffer.html
+[sample]:https://pub.dev/documentation/stream_transform/latest/stream_transform/RateLimit/sample.html
+[debounceBuffer]:https://pub.dev/documentation/stream_transform/latest/stream_transform/RateLimit/debounceBuffer.html
+[debounce]:https://pub.dev/documentation/stream_transform/latest/stream_transform/RateLimit/debounce.html
+[throttle]:https://pub.dev/documentation/stream_transform/latest/stream_transform/RateLimit/throttle.html
+
+## 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/analysis_options.yaml b/pkgs/stream_transform/analysis_options.yaml
new file mode 100644
index 0000000..05f1af1
--- /dev/null
+++ b/pkgs/stream_transform/analysis_options.yaml
@@ -0,0 +1,16 @@
+include: package:dart_flutter_team_lints/analysis_options.yaml
+
+analyzer:
+  language:
+    strict-casts: true
+    strict-raw-types: true
+
+linter:
+  rules:
+    - avoid_bool_literals_in_conditional_expressions
+    - avoid_classes_with_only_static_members
+    - avoid_returning_this
+    - avoid_unused_constructor_parameters
+    - cascade_invocations
+    - join_return_with_assignment
+    - no_adjacent_strings_in_list
diff --git a/pkgs/stream_transform/example/index.html b/pkgs/stream_transform/example/index.html
new file mode 100644
index 0000000..aecdc09
--- /dev/null
+++ b/pkgs/stream_transform/example/index.html
@@ -0,0 +1,11 @@
+<html>
+  <head>
+    <script defer src="main.dart.js" type="application/javascript"></script>
+  </head>
+  <body>
+    <input id="first_input"><br>
+    <input id="second_input"><br>
+    <p id="output">
+    </p>
+  </body>
+</html>
diff --git a/pkgs/stream_transform/example/main.dart b/pkgs/stream_transform/example/main.dart
new file mode 100644
index 0000000..70b3e7f
--- /dev/null
+++ b/pkgs/stream_transform/example/main.dart
@@ -0,0 +1,26 @@
+// 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:html';
+
+import 'package:stream_transform/stream_transform.dart';
+
+void main() {
+  var firstInput = document.querySelector('#first_input') as InputElement;
+  var secondInput = document.querySelector('#second_input') as InputElement;
+  var output = document.querySelector('#output')!;
+
+  _inputValues(firstInput)
+      .combineLatest(_inputValues(secondInput),
+          (first, second) => 'First: $first, Second: $second')
+      .tap((v) {
+    print('Saw: $v');
+  }).forEach((v) {
+    output.text = v;
+  });
+}
+
+Stream<String?> _inputValues(InputElement element) => element.onKeyUp
+    .debounce(const Duration(milliseconds: 100))
+    .map((_) => element.value);
diff --git a/pkgs/stream_transform/lib/src/aggregate_sample.dart b/pkgs/stream_transform/lib/src/aggregate_sample.dart
new file mode 100644
index 0000000..f2ff8ed
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/aggregate_sample.dart
@@ -0,0 +1,146 @@
+// 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';
+
+import 'common_callbacks.dart';
+
+extension AggregateSample<T> on Stream<T> {
+  /// Computes a value based on sequences of events, then emits that value when
+  /// [trigger] emits an event.
+  ///
+  /// Every time this stream emits an event, an intermediate value is created
+  /// by combining the new event with the previous intermediate value, or with
+  /// `null` if there is no previous value, using the [aggregate] function.
+  ///
+  /// When [trigger] emits value, the returned stream emits the current
+  /// intermediate value and clears it.
+  ///
+  /// If [longPoll] is `false`, if there is no intermediate value when [trigger]
+  /// emits an event, the [onEmpty] function is called with a [Sink] which can
+  /// add events to the returned stream.
+  ///
+  /// If [longPoll] is `true`, and there is no intermediate value when [trigger]
+  /// emits one or more events, then the *next* event from this stream is
+  /// immediately put through [aggregate] and emitted on the returned stream.
+  /// Subsequent events on [trigger] while there have been no events on this
+  /// stream are ignored.
+  /// In that case, [onEmpty] is never used.
+  ///
+  /// The result stream will close as soon as there is a guarantee it will not
+  /// emit any more events. There will not be any more events emitted if:
+  /// - [trigger] is closed and there is no waiting long poll.
+  /// - Or, the source stream is closed and there are no buffered events.
+  ///
+  /// 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<S> aggregateSample<S>(
+      {required Stream<void> trigger,
+      required S Function(T, S?) aggregate,
+      required bool longPoll,
+      required void Function(Sink<S>) onEmpty}) {
+    var controller = isBroadcast
+        ? StreamController<S>.broadcast(sync: true)
+        : StreamController<S>(sync: true);
+
+    S? currentResults;
+    var hasCurrentResults = false;
+    var activeLongPoll = false;
+    var isTriggerDone = false;
+    var isValueDone = false;
+    StreamSubscription<T>? valueSub;
+    StreamSubscription<void>? triggerSub;
+
+    void emit(S results) {
+      currentResults = null;
+      hasCurrentResults = false;
+      controller.add(results);
+    }
+
+    void onValue(T value) {
+      currentResults = aggregate(value, currentResults);
+      hasCurrentResults = true;
+      if (!longPoll) return;
+
+      if (activeLongPoll) {
+        activeLongPoll = false;
+        emit(currentResults as S);
+      }
+
+      if (isTriggerDone) {
+        valueSub!.cancel();
+        controller.close();
+      }
+    }
+
+    void onValuesDone() {
+      isValueDone = true;
+      if (!hasCurrentResults) {
+        triggerSub?.cancel();
+        controller.close();
+      }
+    }
+
+    void onTrigger(_) {
+      if (hasCurrentResults) {
+        emit(currentResults as S);
+      } else if (longPoll) {
+        activeLongPoll = true;
+      } else {
+        onEmpty(controller);
+      }
+
+      if (isValueDone) {
+        triggerSub!.cancel();
+        controller.close();
+      }
+    }
+
+    void onTriggerDone() {
+      isTriggerDone = true;
+      if (!activeLongPoll) {
+        valueSub?.cancel();
+        controller.close();
+      }
+    }
+
+    controller.onListen = () {
+      assert(valueSub == null);
+      valueSub =
+          listen(onValue, onError: controller.addError, onDone: onValuesDone);
+      final priorTriggerSub = triggerSub;
+      if (priorTriggerSub != null) {
+        if (priorTriggerSub.isPaused) priorTriggerSub.resume();
+      } else {
+        triggerSub = trigger.listen(onTrigger,
+            onError: controller.addError, onDone: onTriggerDone);
+      }
+      if (!isBroadcast) {
+        controller
+          ..onPause = () {
+            valueSub?.pause();
+            triggerSub?.pause();
+          }
+          ..onResume = () {
+            valueSub?.resume();
+            triggerSub?.resume();
+          };
+      }
+      controller.onCancel = () {
+        var cancels = <Future<void>>[if (!isValueDone) valueSub!.cancel()];
+        valueSub = null;
+        if (trigger.isBroadcast || !isBroadcast) {
+          if (!isTriggerDone) cancels.add(triggerSub!.cancel());
+          triggerSub = null;
+        } else {
+          triggerSub!.pause();
+        }
+        if (cancels.isEmpty) return null;
+        return cancels.wait.then(ignoreArgument);
+      };
+    };
+    return controller.stream;
+  }
+}
diff --git a/pkgs/stream_transform/lib/src/async_expand.dart b/pkgs/stream_transform/lib/src/async_expand.dart
new file mode 100644
index 0000000..28d2f40
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/async_expand.dart
@@ -0,0 +1,89 @@
+// Copyright (c) 2022, 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 'common_callbacks.dart';
+import 'switch.dart';
+
+/// Alternatives to [asyncExpand].
+///
+/// The built in [asyncExpand] will not overlap the inner streams and every
+/// event will be sent to the callback individually.
+///
+/// - [concurrentAsyncExpand] allow overlap and merges inner streams without
+///   ordering guarantees.
+extension AsyncExpand<T> on Stream<T> {
+  /// Like [asyncExpand] but the [convert] callback may be called for an element
+  /// before the [Stream] emitted by the previous element has closed.
+  ///
+  /// Events on the result stream will be emitted in the order they are emitted
+  /// by the sub streams, which may not match the order of this stream.
+  ///
+  /// Errors from [convert], the source stream, or any of the sub streams are
+  /// forwarded to the result stream.
+  ///
+  /// The result stream will not close until the source stream closes and all
+  /// sub streams have closed.
+  ///
+  /// If the source stream is a broadcast stream, the result will be as well,
+  /// regardless of the types of streams created by [convert]. In this case,
+  /// some care should be taken:
+  /// -  If [convert] returns a single subscription stream it may be listened to
+  /// and never canceled.
+  /// -  For any period of time where there are no listeners on the result
+  /// stream, any sub streams from previously emitted events will be ignored,
+  /// regardless of whether they emit further events after a listener is added
+  /// back.
+  ///
+  /// See also:
+  /// - [switchMap], which cancels subscriptions to the previous sub stream
+  /// instead of concurrently emitting events from all sub streams.
+  Stream<S> concurrentAsyncExpand<S>(Stream<S> Function(T) convert) {
+    final controller = isBroadcast
+        ? StreamController<S>.broadcast(sync: true)
+        : StreamController<S>(sync: true);
+
+    controller.onListen = () {
+      final subscriptions = <StreamSubscription<dynamic>>[];
+      final outerSubscription = map(convert).listen((inner) {
+        if (isBroadcast && !inner.isBroadcast) {
+          inner = inner.asBroadcastStream();
+        }
+        final subscription =
+            inner.listen(controller.add, onError: controller.addError);
+        subscription.onDone(() {
+          subscriptions.remove(subscription);
+          if (subscriptions.isEmpty) controller.close();
+        });
+        subscriptions.add(subscription);
+      }, onError: controller.addError);
+      outerSubscription.onDone(() {
+        subscriptions.remove(outerSubscription);
+        if (subscriptions.isEmpty) controller.close();
+      });
+      subscriptions.add(outerSubscription);
+      if (!isBroadcast) {
+        controller
+          ..onPause = () {
+            for (final subscription in subscriptions) {
+              subscription.pause();
+            }
+          }
+          ..onResume = () {
+            for (final subscription in subscriptions) {
+              subscription.resume();
+            }
+          };
+      }
+      controller.onCancel = () {
+        if (subscriptions.isEmpty) return null;
+        return [for (var s in subscriptions) s.cancel()]
+            .wait
+            .then(ignoreArgument);
+      };
+    };
+    return controller.stream;
+  }
+}
diff --git a/pkgs/stream_transform/lib/src/async_map.dart b/pkgs/stream_transform/lib/src/async_map.dart
new file mode 100644
index 0000000..094df9c
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/async_map.dart
@@ -0,0 +1,136 @@
+// 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 'aggregate_sample.dart';
+import 'common_callbacks.dart';
+import 'from_handlers.dart';
+import 'rate_limit.dart';
+
+/// Alternatives to [asyncMap].
+///
+/// The built in [asyncMap] will not overlap execution of the passed callback,
+/// and every event will be sent to the callback individually.
+///
+/// - [asyncMapBuffer] prevents the callback from overlapping execution and
+///   collects events while it is executing to process in batches.
+/// - [asyncMapSample] prevents overlapping execution and discards events while
+///   it is executing.
+/// - [concurrentAsyncMap] allows overlap and removes ordering guarantees.
+extension AsyncMap<T> on Stream<T> {
+  /// Like [asyncMap] but events are buffered until previous events have been
+  /// processed by [convert].
+  ///
+  /// If this 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 first event from this 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 completed.
+  ///
+  /// Errors from this 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 this 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 buffer(workFinished.stream)._asyncMapThen(convert, workFinished.add);
+  }
+
+  /// Like [asyncMap] but events are discarded while work is happening in
+  /// [convert].
+  ///
+  /// If this 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.
+  ///
+  /// 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 this 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 this 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 aggregateSample(
+            trigger: workFinished.stream,
+            aggregate: _dropPrevious,
+            longPoll: true,
+            onEmpty: ignoreArgument)
+        ._asyncMapThen(convert, workFinished.add);
+  }
+
+  /// 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 this stream.
+  ///
+  /// If this 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 this stream are forwarded directly to the
+  /// result stream.
+  ///
+  /// The result stream will not close until this stream closes and all pending
+  /// conversions have finished.
+  Stream<S> concurrentAsyncMap<S>(FutureOr<S> Function(T) convert) {
+    var valuesWaiting = 0;
+    var sourceDone = false;
+    return transformByHandlers(onData: (element, sink) {
+      valuesWaiting++;
+      () async {
+        try {
+          sink.add(await convert(element));
+        } catch (e, st) {
+          sink.addError(e, st);
+        }
+        valuesWaiting--;
+        if (valuesWaiting <= 0 && sourceDone) sink.close();
+      }();
+    }, onDone: (sink) {
+      sourceDone = true;
+      if (valuesWaiting <= 0) sink.close();
+    });
+  }
+
+  /// Like [Stream.asyncMap] but the [convert] is only called once per event,
+  /// rather than once per listener, and [then] is called after completing the
+  /// work.
+  Stream<S> _asyncMapThen<S>(
+      Future<S> Function(T) convert, void Function(void) then) {
+    Future<void>? pendingEvent;
+    return transformByHandlers(onData: (event, sink) {
+      pendingEvent =
+          convert(event).then(sink.add).catchError(sink.addError).then(then);
+    }, onDone: (sink) {
+      if (pendingEvent != null) {
+        pendingEvent!.then((_) => sink.close());
+      } else {
+        sink.close();
+      }
+    });
+  }
+}
+
+T _dropPrevious<T>(T event, _) => event;
diff --git a/pkgs/stream_transform/lib/src/combine_latest.dart b/pkgs/stream_transform/lib/src/combine_latest.dart
new file mode 100644
index 0000000..f02a19e
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/combine_latest.dart
@@ -0,0 +1,240 @@
+// 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';
+
+import 'common_callbacks.dart';
+
+/// Utilities to combine events from multiple streams through a callback or into
+/// a list.
+extension CombineLatest<T> on Stream<T> {
+  /// Combines the latest values from this stream with the latest values 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) {
+    final controller = isBroadcast
+        ? StreamController<S>.broadcast(sync: true)
+        : StreamController<S>(sync: true);
+
+    other =
+        (isBroadcast && !other.isBroadcast) ? other.asBroadcastStream() : other;
+
+    StreamSubscription<T>? sourceSubscription;
+    StreamSubscription<T2>? otherSubscription;
+
+    var sourceDone = false;
+    var otherDone = false;
+
+    late T latestSource;
+    late T2 latestOther;
+
+    var sourceStarted = false;
+    var otherStarted = false;
+
+    void emitCombined() {
+      if (!sourceStarted || !otherStarted) return;
+      FutureOr<S> result;
+      try {
+        result = combine(latestSource, latestOther);
+      } catch (e, s) {
+        controller.addError(e, s);
+        return;
+      }
+      if (result is Future<S>) {
+        sourceSubscription!.pause();
+        otherSubscription!.pause();
+        result
+            .then(controller.add, onError: controller.addError)
+            .whenComplete(() {
+          sourceSubscription!.resume();
+          otherSubscription!.resume();
+        });
+      } else {
+        controller.add(result);
+      }
+    }
+
+    controller.onListen = () {
+      assert(sourceSubscription == null);
+      sourceSubscription = listen(
+          (s) {
+            sourceStarted = true;
+            latestSource = s;
+            emitCombined();
+          },
+          onError: controller.addError,
+          onDone: () {
+            sourceDone = true;
+            if (otherDone) {
+              controller.close();
+            } else if (!sourceStarted) {
+              // Nothing can ever be emitted
+              otherSubscription!.cancel();
+              controller.close();
+            }
+          });
+      otherSubscription = other.listen(
+          (o) {
+            otherStarted = true;
+            latestOther = o;
+            emitCombined();
+          },
+          onError: controller.addError,
+          onDone: () {
+            otherDone = true;
+            if (sourceDone) {
+              controller.close();
+            } else if (!otherStarted) {
+              // Nothing can ever be emitted
+              sourceSubscription!.cancel();
+              controller.close();
+            }
+          });
+      if (!isBroadcast) {
+        controller
+          ..onPause = () {
+            sourceSubscription!.pause();
+            otherSubscription!.pause();
+          }
+          ..onResume = () {
+            sourceSubscription!.resume();
+            otherSubscription!.resume();
+          };
+      }
+      controller.onCancel = () {
+        var cancels = [
+          sourceSubscription!.cancel(),
+          otherSubscription!.cancel()
+        ];
+        sourceSubscription = null;
+        otherSubscription = null;
+        return cancels.wait.then(ignoreArgument);
+      };
+    };
+    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.
+  ///
+  /// 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) {
+    final controller = isBroadcast
+        ? StreamController<List<T>>.broadcast(sync: true)
+        : StreamController<List<T>>(sync: true);
+
+    final allStreams = [
+      this,
+      for (final other in others)
+        !isBroadcast || other.isBroadcast ? other : other.asBroadcastStream(),
+    ];
+
+    controller.onListen = () {
+      final subscriptions = <StreamSubscription<T>>[];
+
+      final latestData = List<T?>.filled(allStreams.length, null);
+      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 streamId = 0;
+      for (final stream in allStreams) {
+        final index = streamId;
+
+        final subscription = stream.listen((data) => handleData(index, data),
+            onError: controller.addError);
+        subscription.onDone(() {
+          assert(subscriptions.contains(subscription));
+          subscriptions.remove(subscription);
+          if (subscriptions.isEmpty || !hasEmitted.contains(index)) {
+            controller.close();
+          }
+        });
+        subscriptions.add(subscription);
+
+        streamId++;
+      }
+      if (!isBroadcast) {
+        controller
+          ..onPause = () {
+            for (final subscription in subscriptions) {
+              subscription.pause();
+            }
+          }
+          ..onResume = () {
+            for (final subscription in subscriptions) {
+              subscription.resume();
+            }
+          };
+      }
+      controller.onCancel = () {
+        if (subscriptions.isEmpty) return null;
+        return [for (var s in subscriptions) s.cancel()]
+            .wait
+            .then(ignoreArgument);
+      };
+    };
+    return controller.stream;
+  }
+}
diff --git a/pkgs/stream_transform/lib/src/common_callbacks.dart b/pkgs/stream_transform/lib/src/common_callbacks.dart
new file mode 100644
index 0000000..c239220
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/common_callbacks.dart
@@ -0,0 +1,5 @@
+// Copyright (c) 2024, 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.
+
+void ignoreArgument(_) {}
diff --git a/pkgs/stream_transform/lib/src/concatenate.dart b/pkgs/stream_transform/lib/src/concatenate.dart
new file mode 100644
index 0000000..0330dd7
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/concatenate.dart
@@ -0,0 +1,112 @@
+// 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';
+
+/// Utilities to append or prepend to a stream.
+extension Concatenate<T> on Stream<T> {
+  /// Emits all values and errors from [next] following all values and errors
+  /// from this stream.
+  ///
+  /// If this stream never finishes, the [next] stream will never get a
+  /// listener.
+  ///
+  /// If this 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 this 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) {
+    var controller = isBroadcast
+        ? StreamController<T>.broadcast(sync: true)
+        : StreamController<T>(sync: true);
+
+    next = isBroadcast && !next.isBroadcast ? next.asBroadcastStream() : next;
+
+    StreamSubscription<T>? subscription;
+    var currentStream = this;
+    var thisDone = false;
+    var secondDone = false;
+
+    late void Function() currentDoneHandler;
+
+    void listen() {
+      subscription = currentStream.listen(controller.add,
+          onError: controller.addError, onDone: () => currentDoneHandler());
+    }
+
+    void onSecondDone() {
+      secondDone = true;
+      controller.close();
+    }
+
+    void onThisDone() {
+      thisDone = true;
+      currentStream = next;
+      currentDoneHandler = onSecondDone;
+      listen();
+    }
+
+    currentDoneHandler = onThisDone;
+
+    controller.onListen = () {
+      assert(subscription == null);
+      listen();
+      if (!isBroadcast) {
+        controller
+          ..onPause = () {
+            if (!thisDone || !next.isBroadcast) return subscription!.pause();
+            subscription!.cancel();
+            subscription = null;
+          }
+          ..onResume = () {
+            if (!thisDone || !next.isBroadcast) return subscription!.resume();
+            listen();
+          };
+      }
+      controller.onCancel = () {
+        if (secondDone) return null;
+        var toCancel = subscription!;
+        subscription = null;
+        return toCancel.cancel();
+      };
+    };
+    return controller.stream;
+  }
+
+  /// Emits [initial] before any values or errors from the this stream.
+  ///
+  /// If this stream is a broadcast stream the result will be as well.
+  /// If this stream is a broadcast stream, the returned stream will only
+  /// contain events of this stream that are emitted after the [initial] value
+  /// has been emitted on the returned stream.
+  Stream<T> startWith(T initial) =>
+      startWithStream(Future.value(initial).asStream());
+
+  /// Emits all values in [initial] before any values or errors from this
+  /// stream.
+  ///
+  /// If this stream is a broadcast stream the result will be as well.
+  /// If this 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));
+
+  /// Emits all values and errors in [initial] before any values or errors from
+  /// this stream.
+  ///
+  /// If this stream is a broadcast stream the result will be as well.
+  /// If this 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);
+  }
+}
diff --git a/pkgs/stream_transform/lib/src/from_handlers.dart b/pkgs/stream_transform/lib/src/from_handlers.dart
new file mode 100644
index 0000000..1146a13
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/from_handlers.dart
@@ -0,0 +1,58 @@
+// 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';
+
+extension TransformByHandlers<S> on Stream<S> {
+  /// Transform a stream by callbacks.
+  ///
+  /// This is similar to `transform(StreamTransformer.fromHandler(...))` except
+  /// that the handlers are called once per event rather than called for the
+  /// same event for each listener on a broadcast stream.
+  Stream<T> transformByHandlers<T>(
+      {required void Function(S, EventSink<T>) onData,
+      void Function(Object, StackTrace, EventSink<T>)? onError,
+      void Function(EventSink<T>)? onDone}) {
+    final handleError = onError ?? _defaultHandleError;
+    final handleDone = onDone ?? _defaultHandleDone;
+
+    var controller = isBroadcast
+        ? StreamController<T>.broadcast(sync: true)
+        : StreamController<T>(sync: true);
+
+    StreamSubscription<S>? subscription;
+    controller.onListen = () {
+      assert(subscription == null);
+      var valuesDone = false;
+      subscription = listen((value) => onData(value, controller),
+          onError: (Object error, StackTrace stackTrace) {
+        handleError(error, stackTrace, controller);
+      }, onDone: () {
+        valuesDone = true;
+        handleDone(controller);
+      });
+      if (!isBroadcast) {
+        controller
+          ..onPause = subscription!.pause
+          ..onResume = subscription!.resume;
+      }
+      controller.onCancel = () {
+        var toCancel = subscription;
+        subscription = null;
+        if (!valuesDone) return toCancel!.cancel();
+        return null;
+      };
+    };
+    return controller.stream;
+  }
+
+  static void _defaultHandleError<T>(
+      Object error, StackTrace stackTrace, EventSink<T> sink) {
+    sink.addError(error, stackTrace);
+  }
+
+  static void _defaultHandleDone<T>(EventSink<T> sink) {
+    sink.close();
+  }
+}
diff --git a/pkgs/stream_transform/lib/src/merge.dart b/pkgs/stream_transform/lib/src/merge.dart
new file mode 100644
index 0000000..3bfe06c
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/merge.dart
@@ -0,0 +1,102 @@
+// 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 'common_callbacks.dart';
+
+/// Utilities to interleave events from multiple streams.
+extension Merge<T> on Stream<T> {
+  /// Merges values and errors from this stream and [other] in any order as they
+  /// arrive.
+  ///
+  /// The result stream will not close until both this 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 this 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) => mergeAll([other]);
+
+  /// Merges values and errors from this stream and any stream in [others] in
+  /// any order as they arrive.
+  ///
+  /// The result stream will not close until this 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 this 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) {
+    final controller = isBroadcast
+        ? StreamController<T>.broadcast(sync: true)
+        : StreamController<T>(sync: true);
+
+    final allStreams = [
+      this,
+      for (final other in others)
+        !isBroadcast || other.isBroadcast ? other : other.asBroadcastStream(),
+    ];
+
+    controller.onListen = () {
+      final subscriptions = <StreamSubscription<T>>[];
+      for (final stream in allStreams) {
+        final subscription =
+            stream.listen(controller.add, onError: controller.addError);
+        subscription.onDone(() {
+          subscriptions.remove(subscription);
+          if (subscriptions.isEmpty) controller.close();
+        });
+        subscriptions.add(subscription);
+      }
+      if (!isBroadcast) {
+        controller
+          ..onPause = () {
+            for (final subscription in subscriptions) {
+              subscription.pause();
+            }
+          }
+          ..onResume = () {
+            for (final subscription in subscriptions) {
+              subscription.resume();
+            }
+          };
+      }
+      controller.onCancel = () {
+        if (subscriptions.isEmpty) return null;
+        return [for (var s in subscriptions) s.cancel()]
+            .wait
+            .then(ignoreArgument);
+      };
+    };
+    return controller.stream;
+  }
+}
diff --git a/pkgs/stream_transform/lib/src/rate_limit.dart b/pkgs/stream_transform/lib/src/rate_limit.dart
new file mode 100644
index 0000000..299c230
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/rate_limit.dart
@@ -0,0 +1,356 @@
+// 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';
+
+import 'aggregate_sample.dart';
+import 'common_callbacks.dart';
+import 'from_handlers.dart';
+
+/// Utilities to rate limit events.
+///
+/// - [debounce] - emit the the _first_ or _last_ event of a series of closely
+///   spaced events.
+/// - [debounceBuffer] - emit _all_ events at the _end_ of a series of closely
+///   spaced events.
+/// - [throttle] - emit the _first_ event at the _beginning_ of the period.
+/// - [audit] - emit the _last_ event at the _end_ of the period.
+/// - [buffer] - emit _all_ events on a _trigger_.
+extension RateLimit<T> on Stream<T> {
+  /// Suppresses events with less inter-event spacing than [duration].
+  ///
+  /// Events which are emitted with less than [duration] elapsed between them
+  /// are considered to be part of the same "series". If [leading] is `true`,
+  /// the first event of this series is emitted immediately. If [trailing] is
+  /// `true` the last event of this series is emitted with a delay of at least
+  /// [duration]. By default only trailing events are emitted, both arguments
+  /// must be specified with `leading: true, trailing: false` to emit only
+  /// leading events.
+  ///
+  /// If this stream is a broadcast stream, the result will be as well.
+  /// Errors are forwarded immediately.
+  ///
+  /// If there is a trailing 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 this stream closes the returned stream will close immediately.
+  ///
+  /// For example:
+  ///
+  ///     source.debounce(Duration(seconds: 1));
+  ///
+  ///     source: 1-2-3---4---5-6-|
+  ///     result: ------3---4-----6|
+  ///
+  ///     source.debounce(Duration(seconds: 1), leading: true, trailing: false);
+  ///
+  ///     source: 1-2-3---4---5-6-|
+  ///     result: 1-------4---5---|
+  ///
+  ///     source.debounce(Duration(seconds: 1), leading: true);
+  ///
+  ///     source: 1-2-3---4---5-6-|
+  ///     result: 1-----3-4---5---6|
+  ///
+  /// To collect values emitted during the debounce period see [debounceBuffer].
+  Stream<T> debounce(Duration duration,
+          {bool leading = false, bool trailing = true}) =>
+      _debounceAggregate(duration, _dropPrevious,
+          leading: leading, trailing: trailing);
+
+  /// Buffers values until this 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 this 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 this stream
+  /// closes the returned stream will wait to emit them following the debounce
+  /// period before closing. If there are no pending debounced events when this
+  /// stream closes the returned stream will close immediately.
+  ///
+  /// To keep only the most recent event during the debounce period see
+  /// [debounce].
+  Stream<List<T>> debounceBuffer(Duration duration) =>
+      _debounceAggregate(duration, _collect, leading: false, trailing: true);
+
+  /// Reduces the rate that events are emitted to at most once per [duration].
+  ///
+  /// No events will ever be emitted within [duration] of another event on the
+  /// result stream.
+  /// If this stream is a broadcast stream, the result will be as well.
+  /// Errors are forwarded immediately.
+  ///
+  /// If [trailing] is `false`, source events emitted during the [duration]
+  /// period following a result event are discarded.
+  /// The result stream will not emit an event until this stream emits an event
+  /// following the throttled period.
+  /// If this stream is consistently emitting events with less than
+  /// [duration] between events, the time between events on the result stream
+  /// may still be more than [duration].
+  /// The result stream will close immediately when this stream closes.
+  ///
+  /// If [trailing] is `true`, the latest source event emitted during the
+  /// [duration] period following an result event is held and emitted following
+  /// the period.
+  /// If this stream is consistently emitting events with less than [duration]
+  /// between events, the time between events on the result stream will be
+  /// [duration].
+  /// If this stream closes the result stream will wait to emit a pending event
+  /// before closing.
+  ///
+  /// For example:
+  ///
+  ///     source.throttle(Duration(seconds: 6));
+  ///
+  ///     source: 1-2-3---4-5-6---7-8-|
+  ///     result: 1-------4-------7---|
+  ///
+  ///     source.throttle(Duration(seconds: 6), trailing: true);
+  ///
+  ///     source: 1-2-3---4-5----6--|
+  ///     result: 1-----3-----5-----6|
+  ///
+  ///     source.throttle(Duration(seconds: 6), trailing: true);
+  ///
+  ///     source: 1-2-----------3|
+  ///     result: 1-----2-------3|
+  ///
+  /// See also:
+  /// - [audit], which emits the most recent event at the end of the period.
+  /// Compared to `audit`, `throttle` will not introduce delay to forwarded
+  /// elements, except for the [trailing] events.
+  /// - [debounce], which uses inter-event spacing instead of a fixed period
+  /// from the first event in a window. Compared to `debouce`, `throttle` cannot
+  /// be starved by having events emitted continuously within [duration].
+  Stream<T> throttle(Duration duration, {bool trailing = false}) =>
+      trailing ? _throttleTrailing(duration) : _throttle(duration);
+
+  Stream<T> _throttle(Duration duration) {
+    Timer? timer;
+
+    return transformByHandlers(onData: (data, sink) {
+      if (timer == null) {
+        sink.add(data);
+        timer = Timer(duration, () {
+          timer = null;
+        });
+      }
+    });
+  }
+
+  Stream<T> _throttleTrailing(Duration duration) {
+    Timer? timer;
+    T? pending;
+    var hasPending = false;
+    var isDone = false;
+
+    return transformByHandlers(onData: (data, sink) {
+      void onTimer() {
+        if (hasPending) {
+          sink.add(pending as T);
+          if (isDone) {
+            sink.close();
+          } else {
+            timer = Timer(duration, onTimer);
+            hasPending = false;
+            pending = null;
+          }
+        } else {
+          timer = null;
+        }
+      }
+
+      if (timer == null) {
+        sink.add(data);
+        timer = Timer(duration, onTimer);
+      } else {
+        hasPending = true;
+        pending = data;
+      }
+    }, onDone: (sink) {
+      isDone = true;
+      if (hasPending) return; // Will be closed by timer.
+      sink.close();
+      timer?.cancel();
+      timer = null;
+    });
+  }
+
+  /// Audit a single event from each [duration] length period where there are
+  /// events on this stream.
+  ///
+  /// No events will ever be emitted within [duration] of another event on the
+  /// result stream.
+  /// If this stream is a broadcast stream, the result will be as well.
+  /// Errors are forwarded immediately.
+  ///
+  /// The first event will begin the audit period. At the end of the audit
+  /// period the most recent event is emitted, and the next event restarts the
+  /// audit period.
+  ///
+  /// 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.
+  ///
+  /// If there is no pending event when this stream closes the output
+  /// stream will close immediately. If there is a pending event the output
+  /// stream will wait to emit it before closing.
+  ///
+  /// For example:
+  ///
+  ///     source.audit(Duration(seconds: 5));
+  ///
+  ///     source: a------b--c----d--|
+  ///     output: -----a------c--------d|
+  ///
+  /// See also:
+  /// - [throttle], which emits the _first_ event during the window, instead of
+  /// the last event in the window. Compared to `throttle`, `audit` will
+  /// introduce delay to forwarded events.
+  /// - [debounce], which only emits after the stream has not emitted for some
+  /// period. Compared to `debouce`, `audit` cannot be starved by having events
+  /// emitted continuously within [duration].
+  Stream<T> audit(Duration duration) {
+    Timer? timer;
+    var shouldClose = false;
+    T recentData;
+
+    return transformByHandlers(onData: (data, sink) {
+      recentData = data;
+      timer ??= Timer(duration, () {
+        sink.add(recentData);
+        timer = null;
+        if (shouldClose) {
+          sink.close();
+        }
+      });
+    }, onDone: (sink) {
+      if (timer != null) {
+        shouldClose = true;
+      } else {
+        sink.close();
+      }
+    });
+  }
+
+  /// Buffers the values emitted on this stream and emits them when [trigger]
+  /// emits an event.
+  ///
+  /// If [longPoll] is `false`, if there are no buffered values when [trigger]
+  /// emits an empty list is immediately emitted.
+  ///
+  /// If [longPoll] is `true`, and there are no buffered values when [trigger]
+  /// emits one or more events, then the *next* value from this stream is
+  /// immediately emitted on the returned stream as a single element list.
+  /// Subsequent events on [trigger] while there have been no events on this
+  /// stream are ignored.
+  ///
+  /// The result stream will close as soon as there is a guarantee it will not
+  /// emit any more events. There will not be any more events emitted if:
+  /// - [trigger] is closed and there is no waiting long poll.
+  /// - Or, this stream is closed and previously buffered events have been
+  /// delivered.
+  ///
+  /// If this stream is a broadcast stream, the result will be as well.
+  /// Errors from this stream or the trigger are immediately forwarded to the
+  /// output.
+  ///
+  /// See also:
+  /// - [sample] which use a [trigger] stream in the same way, but keeps only
+  /// the most recent source event.
+  Stream<List<T>> buffer(Stream<void> trigger, {bool longPoll = true}) =>
+      aggregateSample(
+          trigger: trigger,
+          aggregate: _collect,
+          longPoll: longPoll,
+          onEmpty: _empty);
+
+  /// Emits the most recent new value from this stream when [trigger] emits an
+  /// event.
+  ///
+  /// If [longPoll] is `false`, then an event on [trigger] when there is no
+  /// pending source event will be ignored.
+  /// If [longPoll] is `true` (the default), then an event on [trigger] when
+  /// there is no pending source event will cause the next source event
+  /// to immediately flow to the result stream.
+  ///
+  /// If [longPoll] is `false`, if there is no pending source event when
+  /// [trigger] emits, then the trigger event will be ignored.
+  ///
+  /// If [longPoll] is `true`, and there are no buffered values when [trigger]
+  /// emits one or more events, then the *next* value from this stream is
+  /// immediately emitted on the returned stream as a single element list.
+  /// Subsequent events on [trigger] while there have been no events on this
+  /// stream are ignored.
+  ///
+  /// The result stream will close as soon as there is a guarantee it will not
+  /// emit any more events. There will not be any more events emitted if:
+  /// - [trigger] is closed and there is no waiting long poll.
+  /// - Or, this source stream is closed and any pending source event has been
+  /// delivered.
+  ///
+  /// If this source stream is a broadcast stream, the result will be as well.
+  /// Errors from this source stream or the trigger are immediately forwarded to
+  /// the output.
+  ///
+  /// See also:
+  /// - [buffer] which use [trigger] stream in the same way, but keeps a list of
+  /// pending source events.
+  Stream<T> sample(Stream<void> trigger, {bool longPoll = true}) =>
+      aggregateSample(
+          trigger: trigger,
+          aggregate: _dropPrevious,
+          longPoll: longPoll,
+          onEmpty: ignoreArgument);
+
+  /// Aggregates values until this source stream does not emit for [duration],
+  /// then emits the aggregated values.
+  Stream<S> _debounceAggregate<S>(
+      Duration duration, S Function(T element, S? soFar) collect,
+      {required bool leading, required bool trailing}) {
+    Timer? timer;
+    S? soFar;
+    var hasPending = false;
+    var shouldClose = false;
+    var emittedLatestAsLeading = false;
+
+    return transformByHandlers(onData: (value, sink) {
+      void emit() {
+        sink.add(soFar as S);
+        soFar = null;
+        hasPending = false;
+      }
+
+      timer?.cancel();
+      soFar = collect(value, soFar);
+      hasPending = true;
+      if (timer == null && leading) {
+        emittedLatestAsLeading = true;
+        emit();
+      } else {
+        emittedLatestAsLeading = false;
+      }
+      timer = Timer(duration, () {
+        if (trailing && !emittedLatestAsLeading) emit();
+        if (shouldClose) sink.close();
+        timer = null;
+      });
+    }, onDone: (EventSink<S> sink) {
+      if (hasPending && trailing) {
+        shouldClose = true;
+      } else {
+        timer?.cancel();
+        sink.close();
+      }
+    });
+  }
+}
+
+T _dropPrevious<T>(T element, _) => element;
+List<T> _collect<T>(T event, List<T>? soFar) => (soFar ?? <T>[])..add(event);
+void _empty<T>(Sink<List<T>> sink) => sink.add([]);
diff --git a/pkgs/stream_transform/lib/src/scan.dart b/pkgs/stream_transform/lib/src/scan.dart
new file mode 100644
index 0000000..acd3c76
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/scan.dart
@@ -0,0 +1,31 @@
+// 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';
+
+/// A utility similar to [fold] which emits intermediate accumulations.
+extension Scan<T> on Stream<T> {
+  /// Emits a sequence of the accumulated values from repeatedly applying
+  /// [combine].
+  ///
+  /// Like [fold], but instead of producing a single value it yields each
+  /// intermediate result.
+  ///
+  /// If [combine] returns a future it will not be called again for subsequent
+  /// events from the source until it completes, therefore [combine] is always
+  /// called for elements in order, and the result stream always maintains the
+  /// same order as this stream.
+  Stream<S> scan<S>(
+      S initialValue, FutureOr<S> Function(S soFar, T element) combine) {
+    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;
+      }
+    });
+  }
+}
diff --git a/pkgs/stream_transform/lib/src/switch.dart b/pkgs/stream_transform/lib/src/switch.dart
new file mode 100644
index 0000000..546036e
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/switch.dart
@@ -0,0 +1,135 @@
+// 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 'async_expand.dart';
+import 'common_callbacks.dart';
+
+/// A utility to take events from the most recent sub stream returned by a
+/// callback.
+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.
+  /// Like [asyncExpand] but the [Stream] emitted by a previous element
+  /// will be ignored as soon as the source stream emits a new event.
+  ///
+  /// This means that the source stream is not paused until a sub stream
+  /// returned from the [convert] callback is done. Instead, the subscription
+  /// to the sub stream is canceled as soon as the source stream emits a new
+  /// event.
+  ///
+  /// Errors from [convert], the source stream, or any of the sub streams are
+  /// forwarded to the result stream.
+  ///
+  /// The result stream will not close until the source stream closes and
+  /// the current sub stream have closed.
+  ///
+  /// If the source stream is a broadcast stream, the result will be as well,
+  /// regardless of the types of streams created by [convert]. In this case,
+  /// some care should be taken:
+  ///
+  ///  * If [convert] returns a single subscription stream it may be listened to
+  /// and never canceled.
+  ///
+  /// See also:
+  /// - [concurrentAsyncExpand], which emits events from all sub streams
+  ///   concurrently instead of cancelling subscriptions to previous subs
+  ///   streams.
+  Stream<S> switchMap<S>(Stream<S> Function(T) convert) {
+    return map(convert).switchLatest();
+  }
+}
+
+/// A utility to take events from the most recent sub stream.
+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.
+  ///
+  /// Whether the source stream is a single-subscription stream or a
+  /// broadcast stream, the result stream will be the same kind of stream,
+  /// regardless of the types of streams emitted.
+  Stream<T> switchLatest() {
+    var controller = isBroadcast
+        ? StreamController<T>.broadcast(sync: true)
+        : StreamController<T>(sync: true);
+
+    controller.onListen = () {
+      StreamSubscription<T>? innerSubscription;
+      var outerStreamDone = false;
+
+      void listenToInnerStream(Stream<T> innerStream) {
+        assert(innerSubscription == null);
+        var subscription = innerStream
+            .listen(controller.add, onError: controller.addError, onDone: () {
+          innerSubscription = null;
+          if (outerStreamDone) controller.close();
+        });
+        // If a pause happens during an innerSubscription.cancel,
+        // we still listen to the next stream when the cancel is done.
+        // Then we immediately pause it again here.
+        if (controller.isPaused) subscription.pause();
+        innerSubscription = subscription;
+      }
+
+      var addError = controller.addError;
+      final outerSubscription = listen(null, onError: addError, onDone: () {
+        outerStreamDone = true;
+        if (innerSubscription == null) controller.close();
+      });
+      outerSubscription.onData((innerStream) async {
+        var currentSubscription = innerSubscription;
+        if (currentSubscription == null) {
+          listenToInnerStream(innerStream);
+          return;
+        }
+        innerSubscription = null;
+        outerSubscription.pause();
+        try {
+          await currentSubscription.cancel();
+        } catch (error, stack) {
+          controller.addError(error, stack);
+        } finally {
+          if (!isBroadcast && !controller.hasListener) {
+            // Result single-subscription stream subscription was cancelled
+            // while waiting for previous innerStream cancel.
+            //
+            // Ensure that the last received stream is also listened to and
+            // cancelled, then do nothing further.
+            innerStream.listen(null).cancel().ignore();
+          } else {
+            outerSubscription.resume();
+            listenToInnerStream(innerStream);
+          }
+        }
+      });
+      if (!isBroadcast) {
+        controller
+          ..onPause = () {
+            innerSubscription?.pause();
+            outerSubscription.pause();
+          }
+          ..onResume = () {
+            innerSubscription?.resume();
+            outerSubscription.resume();
+          };
+      }
+      controller.onCancel = () {
+        var sub = innerSubscription;
+        var cancels = [
+          if (!outerStreamDone) outerSubscription.cancel(),
+          if (sub != null) sub.cancel(),
+        ];
+        if (cancels.isEmpty) return null;
+        return cancels.wait.then(ignoreArgument);
+      };
+    };
+    return controller.stream;
+  }
+}
diff --git a/pkgs/stream_transform/lib/src/take_until.dart b/pkgs/stream_transform/lib/src/take_until.dart
new file mode 100644
index 0000000..e6deaa1
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/take_until.dart
@@ -0,0 +1,64 @@
+// 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';
+
+/// A utility to end a stream based on an external trigger.
+extension TakeUntil<T> on Stream<T> {
+  /// Takes values from this stream which are emitted before [trigger]
+  /// completes.
+  ///
+  /// 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.
+  ///
+  /// If [trigger] completes as an error, the error will be forwarded through
+  /// the result stream before the result stream closes.
+  ///
+  /// If [trigger] completes as a value or as an error after this stream has
+  /// already ended, the completion will be ignored.
+  Stream<T> takeUntil(Future<void> trigger) {
+    var controller = isBroadcast
+        ? StreamController<T>.broadcast(sync: true)
+        : StreamController<T>(sync: true);
+
+    StreamSubscription<T>? subscription;
+    var isDone = false;
+    trigger.then((_) {
+      if (isDone) return;
+      isDone = true;
+      subscription?.cancel();
+      controller.close();
+    }, onError: (Object error, StackTrace stackTrace) {
+      if (isDone) return;
+      isDone = true;
+      controller
+        ..addError(error, stackTrace)
+        ..close();
+    });
+
+    controller.onListen = () {
+      if (isDone) return;
+      subscription =
+          listen(controller.add, onError: controller.addError, onDone: () {
+        if (isDone) return;
+        isDone = true;
+        controller.close();
+      });
+      if (!isBroadcast) {
+        controller
+          ..onPause = subscription!.pause
+          ..onResume = subscription!.resume;
+      }
+      controller.onCancel = () {
+        if (isDone) return null;
+        var toCancel = subscription!;
+        subscription = null;
+        return toCancel.cancel();
+      };
+    };
+    return controller.stream;
+  }
+}
diff --git a/pkgs/stream_transform/lib/src/tap.dart b/pkgs/stream_transform/lib/src/tap.dart
new file mode 100644
index 0000000..4b16ab5
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/tap.dart
@@ -0,0 +1,44 @@
+// 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 'from_handlers.dart';
+
+/// A utility to chain extra behavior on a stream.
+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 this 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 this stream
+  /// before it is forwarded to listeners on the resulting stream.
+  ///
+  /// The [onDone] callback will be called after this 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}) =>
+      transformByHandlers(onData: (value, sink) {
+        try {
+          onValue?.call(value);
+        } catch (_) {/*Ignore*/}
+        sink.add(value);
+      }, onError: (error, stackTrace, sink) {
+        try {
+          onError?.call(error, stackTrace);
+        } catch (_) {/*Ignore*/}
+        sink.addError(error, stackTrace);
+      }, onDone: (sink) {
+        try {
+          onDone?.call();
+        } catch (_) {/*Ignore*/}
+        sink.close();
+      });
+}
diff --git a/pkgs/stream_transform/lib/src/where.dart b/pkgs/stream_transform/lib/src/where.dart
new file mode 100644
index 0000000..76aa28a
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/where.dart
@@ -0,0 +1,71 @@
+// 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';
+
+import 'from_handlers.dart';
+
+/// Utilities to filter events.
+extension Where<T> on Stream<T> {
+  /// Discards events from this stream that are not of 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>() => transformByHandlers(onData: (event, sink) {
+        if (event is S) sink.add(event);
+      });
+
+  /// Discards events from this stream based on an asynchronous [test] callback.
+  ///
+  /// 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 this 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 [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> Function(T) test) {
+    var valuesWaiting = 0;
+    var sourceDone = false;
+    return transformByHandlers(onData: (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();
+      }();
+    }, onDone: (sink) {
+      sourceDone = true;
+      if (valuesWaiting <= 0) sink.close();
+    });
+  }
+}
+
+extension WhereNotNull<T extends Object> on Stream<T?> {
+  /// Discards `null` events from this stream.
+  ///
+  /// 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.
+  Stream<T> whereNotNull() => transformByHandlers(onData: (event, sink) {
+        if (event != null) sink.add(event);
+      });
+}
diff --git a/pkgs/stream_transform/lib/stream_transform.dart b/pkgs/stream_transform/lib/stream_transform.dart
new file mode 100644
index 0000000..edf4df9
--- /dev/null
+++ b/pkgs/stream_transform/lib/stream_transform.dart
@@ -0,0 +1,15 @@
+// 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.
+
+export 'src/async_expand.dart';
+export 'src/async_map.dart';
+export 'src/combine_latest.dart';
+export 'src/concatenate.dart';
+export 'src/merge.dart';
+export 'src/rate_limit.dart';
+export 'src/scan.dart';
+export 'src/switch.dart';
+export 'src/take_until.dart';
+export 'src/tap.dart';
+export 'src/where.dart';
diff --git a/pkgs/stream_transform/pubspec.yaml b/pkgs/stream_transform/pubspec.yaml
new file mode 100644
index 0000000..1e2298a
--- /dev/null
+++ b/pkgs/stream_transform/pubspec.yaml
@@ -0,0 +1,13 @@
+name: stream_transform
+version: 2.1.1
+description: A collection of utilities to transform and manipulate streams.
+repository: https://github.com/dart-lang/tools/tree/main/pkgs/stream_transform
+
+environment:
+  sdk: ^3.1.0
+
+dev_dependencies:
+  async: ^2.5.0
+  dart_flutter_team_lints: ^2.0.0
+  fake_async: ^1.3.0
+  test: ^1.16.0
diff --git a/pkgs/stream_transform/test/async_expand_test.dart b/pkgs/stream_transform/test/async_expand_test.dart
new file mode 100644
index 0000000..8d84300
--- /dev/null
+++ b/pkgs/stream_transform/test/async_expand_test.dart
@@ -0,0 +1,195 @@
+// 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';
+
+import 'package:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  test('forwards errors from the convert callback', () async {
+    var errors = <String>[];
+    var source = Stream.fromIterable([1, 2, 3]);
+    source.concurrentAsyncExpand<void>((i) {
+      // ignore: only_throw_errors
+      throw 'Error: $i';
+    }).listen((_) {}, onError: errors.add);
+    await Future<void>(() {});
+    expect(errors, ['Error: 1', 'Error: 2', 'Error: 3']);
+  });
+
+  for (var outerType in streamTypes) {
+    for (var innerType in streamTypes) {
+      group('concurrentAsyncExpand $outerType to $innerType', () {
+        late StreamController<int> outerController;
+        late bool outerCanceled;
+        late List<StreamController<String>> innerControllers;
+        late List<bool> innerCanceled;
+        late List<String> emittedValues;
+        late bool isDone;
+        late List<String> errors;
+        late Stream<String> transformed;
+        late StreamSubscription<String> subscription;
+
+        setUp(() {
+          outerController = createController(outerType)
+            ..onCancel = () {
+              outerCanceled = true;
+            };
+          outerCanceled = false;
+          innerControllers = [];
+          innerCanceled = [];
+          emittedValues = [];
+          errors = [];
+          isDone = false;
+          transformed = outerController.stream.concurrentAsyncExpand((i) {
+            var index = innerControllers.length;
+            innerCanceled.add(false);
+            innerControllers.add(createController<String>(innerType)
+              ..onCancel = () {
+                innerCanceled[index] = true;
+              });
+            return innerControllers.last.stream;
+          });
+          subscription = transformed
+              .listen(emittedValues.add, onError: errors.add, onDone: () {
+            isDone = true;
+          });
+        });
+
+        test('interleaves events from sub streams', () async {
+          outerController
+            ..add(1)
+            ..add(2);
+          await Future<void>(() {});
+          expect(emittedValues, isEmpty);
+          expect(innerControllers, hasLength(2));
+          innerControllers[0].add('First');
+          innerControllers[1].add('Second');
+          innerControllers[0].add('First again');
+          await Future<void>(() {});
+          expect(emittedValues, ['First', 'Second', 'First again']);
+        });
+
+        test('forwards errors from outer stream', () async {
+          outerController.addError('Error');
+          await Future<void>(() {});
+          expect(errors, ['Error']);
+        });
+
+        test('forwards errors from inner streams', () async {
+          outerController
+            ..add(1)
+            ..add(2);
+          await Future<void>(() {});
+          innerControllers[0].addError('Error 1');
+          innerControllers[1].addError('Error 2');
+          await Future<void>(() {});
+          expect(errors, ['Error 1', 'Error 2']);
+        });
+
+        test('can continue handling events after an error in outer stream',
+            () async {
+          outerController
+            ..addError('Error')
+            ..add(1);
+          await Future<void>(() {});
+          innerControllers[0].add('First');
+          await Future<void>(() {});
+          expect(emittedValues, ['First']);
+          expect(errors, ['Error']);
+        });
+
+        test('cancels outer subscription if output canceled', () async {
+          await subscription.cancel();
+          expect(outerCanceled, true);
+        });
+
+        if (outerType != 'broadcast' || innerType != 'single subscription') {
+          // A single subscription inner stream in a broadcast outer stream is
+          // not canceled.
+          test('cancels inner subscriptions if output canceled', () async {
+            outerController
+              ..add(1)
+              ..add(2);
+            await Future<void>(() {});
+            await subscription.cancel();
+            expect(innerCanceled, [true, true]);
+          });
+        }
+
+        test('stays open if any inner stream is still open', () async {
+          outerController.add(1);
+          await outerController.close();
+          await Future<void>(() {});
+          expect(isDone, false);
+        });
+
+        test('stays open if outer stream is still open', () async {
+          outerController.add(1);
+          await Future<void>(() {});
+          await innerControllers[0].close();
+          await Future<void>(() {});
+          expect(isDone, false);
+        });
+
+        test('closes after all inner streams and outer stream close', () async {
+          outerController.add(1);
+          await Future<void>(() {});
+          await innerControllers[0].close();
+          await outerController.close();
+          await Future<void>(() {});
+          expect(isDone, true);
+        });
+
+        if (outerType == 'broadcast') {
+          test('multiple listerns all get values', () async {
+            var otherValues = <String>[];
+            transformed.listen(otherValues.add);
+            outerController.add(1);
+            await Future<void>(() {});
+            innerControllers[0].add('First');
+            await Future<void>(() {});
+            expect(emittedValues, ['First']);
+            expect(otherValues, ['First']);
+          });
+
+          test('multiple listeners get closed', () async {
+            var otherDone = false;
+            transformed.listen(null, onDone: () => otherDone = true);
+            outerController.add(1);
+            await Future<void>(() {});
+            await innerControllers[0].close();
+            await outerController.close();
+            await Future<void>(() {});
+            expect(isDone, true);
+            expect(otherDone, true);
+          });
+
+          test('can cancel and relisten', () async {
+            outerController
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            innerControllers[0].add('First');
+            innerControllers[1].add('Second');
+            await Future(() {});
+            await subscription.cancel();
+            innerControllers[0].add('Ignored');
+            await Future(() {});
+            subscription = transformed.listen(emittedValues.add);
+            innerControllers[0].add('Also ignored');
+            outerController.add(3);
+            await Future(() {});
+            innerControllers[2].add('More');
+            await Future(() {});
+            expect(emittedValues, ['First', 'Second', 'More']);
+          });
+        }
+      });
+    }
+  }
+}
diff --git a/pkgs/stream_transform/test/async_map_buffer_test.dart b/pkgs/stream_transform/test/async_map_buffer_test.dart
new file mode 100644
index 0000000..2386217
--- /dev/null
+++ b/pkgs/stream_transform/test/async_map_buffer_test.dart
@@ -0,0 +1,204 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  late StreamController<int> values;
+  late List<String> emittedValues;
+  late bool valuesCanceled;
+  late bool isDone;
+  late List<String> errors;
+  late Stream<String> transformed;
+  late StreamSubscription<String> subscription;
+
+  Completer<String>? finishWork;
+  List<int>? workArgument;
+
+  /// Represents the async `convert` function and asserts that is is only called
+  /// after the previous iteration has completed.
+  Future<String> work(List<int> values) {
+    expect(finishWork, isNull,
+        reason: 'See $values befor previous work is complete');
+    workArgument = values;
+    finishWork = Completer()
+      ..future.then((_) {
+        workArgument = null;
+        finishWork = null;
+      }).catchError((_) {
+        workArgument = null;
+        finishWork = null;
+      });
+    return finishWork!.future;
+  }
+
+  for (var streamType in streamTypes) {
+    group('asyncMapBuffer for stream type: [$streamType]', () {
+      setUp(() {
+        valuesCanceled = false;
+        values = createController(streamType)
+          ..onCancel = () {
+            valuesCanceled = true;
+          };
+        emittedValues = [];
+        errors = [];
+        isDone = false;
+        finishWork = null;
+        workArgument = null;
+        transformed = values.stream.asyncMapBuffer(work);
+        subscription = transformed
+            .listen(emittedValues.add, onError: errors.add, onDone: () {
+          isDone = true;
+        });
+      });
+
+      test('does not emit before work finishes', () async {
+        values.add(1);
+        await Future(() {});
+        expect(emittedValues, isEmpty);
+        expect(workArgument, [1]);
+        finishWork!.complete('result');
+        await Future(() {});
+        expect(emittedValues, ['result']);
+      });
+
+      test('buffers values while work is ongoing', () async {
+        values.add(1);
+        await Future(() {});
+        values
+          ..add(2)
+          ..add(3);
+        await Future(() {});
+        finishWork!.complete('');
+        await Future(() {});
+        expect(workArgument, [2, 3]);
+      });
+
+      test('forwards errors without waiting for work', () async {
+        values.add(1);
+        await Future(() {});
+        values.addError('error');
+        await Future(() {});
+        expect(errors, ['error']);
+      });
+
+      test('forwards errors which occur during the work', () async {
+        values.add(1);
+        await Future(() {});
+        finishWork!.completeError('error');
+        await Future(() {});
+        expect(errors, ['error']);
+      });
+
+      test('can continue handling events after an error', () async {
+        values.add(1);
+        await Future(() {});
+        finishWork!.completeError('error');
+        values.add(2);
+        await Future(() {});
+        expect(workArgument, [2]);
+        finishWork!.completeError('another');
+        await Future(() {});
+        expect(errors, ['error', 'another']);
+      });
+
+      test('does not start next work early due to an error in values',
+          () async {
+        values.add(1);
+        await Future(() {});
+        values
+          ..addError('error')
+          ..add(2);
+        await Future(() {});
+        expect(errors, ['error']);
+        // [work] will assert that the second iteration is not called because
+        // the first has not completed.
+      });
+
+      test('cancels value subscription when output canceled', () async {
+        expect(valuesCanceled, false);
+        await subscription.cancel();
+        expect(valuesCanceled, true);
+      });
+
+      test('closes when values end if no work is pending', () async {
+        expect(isDone, false);
+        await values.close();
+        await Future(() {});
+        expect(isDone, true);
+      });
+
+      test('waits for pending work when values close', () async {
+        values.add(1);
+        await Future(() {});
+        expect(isDone, false);
+        values.add(2);
+        await values.close();
+        expect(isDone, false);
+        finishWork!.complete('');
+        await Future(() {});
+        // Still a pending value
+        expect(isDone, false);
+        finishWork!.complete('');
+        await Future(() {});
+        expect(isDone, true);
+      });
+
+      test('forwards errors from values', () async {
+        values.addError('error');
+        await Future(() {});
+        expect(errors, ['error']);
+      });
+
+      if (streamType == 'broadcast') {
+        test('multiple listeners all get values', () async {
+          var otherValues = <String>[];
+          transformed.listen(otherValues.add);
+          values.add(1);
+          await Future(() {});
+          finishWork!.complete('result');
+          await Future(() {});
+          expect(emittedValues, ['result']);
+          expect(otherValues, ['result']);
+        });
+
+        test('multiple listeners get done when values end', () async {
+          var otherDone = false;
+          transformed.listen(null, onDone: () => otherDone = true);
+          values.add(1);
+          await Future(() {});
+          await values.close();
+          expect(isDone, false);
+          expect(otherDone, false);
+          finishWork!.complete('');
+          await Future(() {});
+          expect(isDone, true);
+          expect(otherDone, true);
+        });
+
+        test('can cancel and relisten', () async {
+          values.add(1);
+          await Future(() {});
+          finishWork!.complete('first');
+          await Future(() {});
+          await subscription.cancel();
+          values.add(2);
+          await Future(() {});
+          subscription = transformed.listen(emittedValues.add);
+          values.add(3);
+          await Future(() {});
+          expect(workArgument, [3]);
+          finishWork!.complete('second');
+          await Future(() {});
+          expect(emittedValues, ['first', 'second']);
+        });
+      }
+    });
+  }
+}
diff --git a/pkgs/stream_transform/test/async_map_sample_test.dart b/pkgs/stream_transform/test/async_map_sample_test.dart
new file mode 100644
index 0000000..62b1b92
--- /dev/null
+++ b/pkgs/stream_transform/test/async_map_sample_test.dart
@@ -0,0 +1,209 @@
+// 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';
+
+import 'package:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  late StreamController<int> values;
+  late List<String> emittedValues;
+  late bool valuesCanceled;
+  late bool isDone;
+  late List<String> errors;
+  late Stream<String> transformed;
+  late StreamSubscription<String> subscription;
+
+  Completer<String>? finishWork;
+  int? workArgument;
+
+  /// Represents the async `convert` function and asserts that is is only called
+  /// after the previous iteration has completed.
+  Future<String> work(int value) {
+    expect(finishWork, isNull,
+        reason: 'See $values befor previous work is complete');
+    workArgument = value;
+    finishWork = Completer()
+      ..future.then((_) {
+        workArgument = null;
+        finishWork = null;
+      }).catchError((_) {
+        workArgument = null;
+        finishWork = null;
+      });
+    return finishWork!.future;
+  }
+
+  for (var streamType in streamTypes) {
+    group('asyncMapSample for stream type: [$streamType]', () {
+      setUp(() {
+        valuesCanceled = false;
+        values = createController(streamType)
+          ..onCancel = () {
+            valuesCanceled = true;
+          };
+        emittedValues = [];
+        errors = [];
+        isDone = false;
+        finishWork = null;
+        workArgument = null;
+        transformed = values.stream.asyncMapSample(work);
+        subscription = transformed
+            .listen(emittedValues.add, onError: errors.add, onDone: () {
+          isDone = true;
+        });
+      });
+
+      test('does not emit before work finishes', () async {
+        values.add(1);
+        await Future(() {});
+        expect(emittedValues, isEmpty);
+        expect(workArgument, 1);
+        finishWork!.complete('result');
+        await Future(() {});
+        expect(emittedValues, ['result']);
+      });
+
+      test('buffers values while work is ongoing', () async {
+        values.add(1);
+        await Future(() {});
+        values
+          ..add(2)
+          ..add(3);
+        await Future(() {});
+        finishWork!.complete('');
+        await Future(() {});
+        expect(workArgument, 3);
+      });
+
+      test('forwards errors without waiting for work', () async {
+        values.add(1);
+        await Future(() {});
+        values.addError('error');
+        await Future(() {});
+        expect(errors, ['error']);
+      });
+
+      test('forwards errors which occur during the work', () async {
+        values.add(1);
+        await Future(() {});
+        finishWork!.completeError('error');
+        await Future(() {});
+        expect(errors, ['error']);
+      });
+
+      test('can continue handling events after an error', () async {
+        values.add(1);
+        await Future(() {});
+        finishWork!.completeError('error');
+        values.add(2);
+        await Future(() {});
+        expect(workArgument, 2);
+        finishWork!.completeError('another');
+        await Future(() {});
+        expect(errors, ['error', 'another']);
+      });
+
+      test('does not start next work early due to an error in values',
+          () async {
+        values.add(1);
+        await Future(() {});
+        values
+          ..addError('error')
+          ..add(2);
+        await Future(() {});
+        expect(errors, ['error']);
+        // [work] will assert that the second iteration is not called because
+        // the first has not completed.
+      });
+
+      test('cancels value subscription when output canceled', () async {
+        expect(valuesCanceled, false);
+        await subscription.cancel();
+        expect(valuesCanceled, true);
+      });
+
+      test('closes when values end if no work is pending', () async {
+        expect(isDone, false);
+        await values.close();
+        await Future(() {});
+        expect(isDone, true);
+      });
+
+      test('waits for pending work when values close', () async {
+        values.add(1);
+        await Future(() {});
+        expect(isDone, false);
+        values.add(2);
+        await values.close();
+        expect(isDone, false);
+        finishWork!.complete('');
+        await Future(() {});
+        // Still a pending value
+        expect(isDone, false);
+        finishWork!.complete('');
+        await Future(() {});
+        expect(isDone, true);
+      });
+
+      test('forwards errors from values', () async {
+        values.addError('error');
+        await Future(() {});
+        expect(errors, ['error']);
+      });
+
+      if (streamType == 'broadcast') {
+        test('multiple listeners all get values', () async {
+          var otherValues = <String>[];
+          transformed.listen(otherValues.add);
+          values.add(1);
+          await Future(() {});
+          finishWork!.complete('result');
+          await Future(() {});
+          expect(emittedValues, ['result']);
+          expect(otherValues, ['result']);
+        });
+
+        test('multiple listeners get done when values end', () async {
+          var otherDone = false;
+          transformed.listen(null, onDone: () => otherDone = true);
+          values.add(1);
+          await Future(() {});
+          await values.close();
+          expect(isDone, false);
+          expect(otherDone, false);
+          finishWork!.complete('');
+          await Future(() {});
+          expect(isDone, true);
+          expect(otherDone, true);
+        });
+
+        test('can cancel and relisten', () async {
+          values.add(1);
+          await Future(() {});
+          finishWork!.complete('first');
+          await Future(() {});
+          await subscription.cancel();
+          values.add(2);
+          await Future(() {});
+          subscription = transformed.listen(emittedValues.add);
+          values.add(3);
+          await Future(() {});
+          expect(workArgument, 3);
+          finishWork!.complete('second');
+          await Future(() {});
+          expect(emittedValues, ['first', 'second']);
+        });
+      }
+    });
+  }
+
+  test('allows nulls', () async {
+    var stream = Stream<int?>.value(null);
+    await stream.asyncMapSample(expectAsync1((_) async {})).drain<void>();
+  });
+}
diff --git a/pkgs/stream_transform/test/async_where_test.dart b/pkgs/stream_transform/test/async_where_test.dart
new file mode 100644
index 0000000..6ea4e76
--- /dev/null
+++ b/pkgs/stream_transform/test/async_where_test.dart
@@ -0,0 +1,90 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+void main() {
+  test('forwards only events that pass the predicate', () async {
+    var values = Stream.fromIterable([1, 2, 3, 4]);
+    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.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.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.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.asyncWhere((e) async => e > 2);
+    var firstValues = <int>[];
+    var secondValues = <int>[];
+    filtered
+      ..listen(firstValues.add)
+      ..listen(secondValues.add);
+    values
+      ..add(1)
+      ..add(2)
+      ..add(3)
+      ..add(4);
+    await Future(() {});
+    expect(firstValues, [3, 4]);
+    expect(secondValues, [3, 4]);
+  });
+
+  test('closes streams with multiple listeners', () async {
+    var values = StreamController<int>.broadcast();
+    var predicate = Completer<bool>();
+    var filtered = values.stream.asyncWhere((_) => predicate.future);
+    var firstDone = false;
+    var secondDone = false;
+    filtered
+      ..listen(null, onDone: () => firstDone = true)
+      ..listen(null, onDone: () => secondDone = true);
+    values.add(1);
+    await values.close();
+    expect(firstDone, false);
+    expect(secondDone, false);
+
+    predicate.complete(true);
+    await Future(() {});
+    expect(firstDone, true);
+    expect(secondDone, true);
+  });
+
+  test('forwards errors emitted by the test callback', () async {
+    var errors = <Object>[];
+    var emitted = <Object>[];
+    var values = Stream.fromIterable([1, 2, 3, 4]);
+    var filtered = values.asyncWhere((e) async {
+      await Future(() {});
+      if (e.isEven) throw Exception('$e');
+      return true;
+    });
+    var done = Completer<Object?>();
+    filtered.listen(emitted.add, onError: errors.add, onDone: done.complete);
+    await done.future;
+    expect(emitted, [1, 3]);
+    expect(errors.map((e) => '$e'), ['Exception: 2', 'Exception: 4']);
+  });
+}
diff --git a/pkgs/stream_transform/test/audit_test.dart b/pkgs/stream_transform/test/audit_test.dart
new file mode 100644
index 0000000..28537db
--- /dev/null
+++ b/pkgs/stream_transform/test/audit_test.dart
@@ -0,0 +1,140 @@
+// 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:fake_async/fake_async.dart';
+import 'package:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  for (var streamType in streamTypes) {
+    group('Stream type [$streamType]', () {
+      late StreamController<int> values;
+      late List<int> emittedValues;
+      late bool valuesCanceled;
+      late bool isDone;
+      late List<String> errors;
+      late Stream<int> transformed;
+      late StreamSubscription<int> subscription;
+
+      group('audit', () {
+        setUp(() {
+          valuesCanceled = false;
+          values = createController(streamType)
+            ..onCancel = () {
+              valuesCanceled = true;
+            };
+          emittedValues = [];
+          errors = [];
+          isDone = false;
+          transformed = values.stream.audit(const Duration(milliseconds: 6));
+        });
+
+        void listen() {
+          subscription = transformed
+              .listen(emittedValues.add, onError: errors.add, onDone: () {
+            isDone = true;
+          });
+        }
+
+        test('cancels values', () async {
+          listen();
+          await subscription.cancel();
+          expect(valuesCanceled, true);
+        });
+
+        test('swallows values that come faster than duration', () {
+          fakeAsync((async) {
+            listen();
+            values
+              ..add(1)
+              ..add(2)
+              ..close();
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [2]);
+          });
+        });
+
+        test('outputs multiple values spaced further than duration', () {
+          fakeAsync((async) {
+            listen();
+            values.add(1);
+            async.elapse(const Duration(milliseconds: 6));
+            values.add(2);
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [1, 2]);
+          });
+        });
+
+        test('waits for pending value to close', () {
+          fakeAsync((async) {
+            listen();
+            values
+              ..add(1)
+              ..close();
+            expect(isDone, false);
+            async.elapse(const Duration(milliseconds: 6));
+            expect(isDone, true);
+          });
+        });
+
+        test('closes output if there are no pending values', () {
+          fakeAsync((async) {
+            listen();
+            values.add(1);
+            async.elapse(const Duration(milliseconds: 6));
+            values
+              ..add(2)
+              ..close();
+            expect(isDone, false);
+            expect(emittedValues, [1]);
+            async.elapse(const Duration(milliseconds: 6));
+            expect(isDone, true);
+            expect(emittedValues, [1, 2]);
+          });
+        });
+
+        test('does not starve output if many values come closer than duration',
+            () {
+          fakeAsync((async) {
+            listen();
+            values.add(1);
+            async.elapse(const Duration(milliseconds: 3));
+            values.add(2);
+            async.elapse(const Duration(milliseconds: 3));
+            values.add(3);
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [2, 3]);
+          });
+        });
+
+        if (streamType == 'broadcast') {
+          test('multiple listeners all get the values', () {
+            fakeAsync((async) {
+              listen();
+              values.add(1);
+              async.elapse(const Duration(milliseconds: 3));
+              values.add(2);
+              var otherValues = <int>[];
+              transformed.listen(otherValues.add);
+              values.add(3);
+              async.elapse(const Duration(milliseconds: 3));
+              values.add(4);
+              async.elapse(const Duration(milliseconds: 3));
+              values
+                ..add(5)
+                ..close();
+              async.elapse(const Duration(milliseconds: 6));
+              expect(emittedValues, [3, 5]);
+              expect(otherValues, [3, 5]);
+            });
+          });
+        }
+      });
+    });
+  }
+}
diff --git a/pkgs/stream_transform/test/buffer_test.dart b/pkgs/stream_transform/test/buffer_test.dart
new file mode 100644
index 0000000..830f555
--- /dev/null
+++ b/pkgs/stream_transform/test/buffer_test.dart
@@ -0,0 +1,305 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  late StreamController<void> trigger;
+  late StreamController<int> values;
+  late List<List<int>> emittedValues;
+  late bool valuesCanceled;
+  late bool triggerCanceled;
+  late bool triggerPaused;
+  late bool isDone;
+  late List<String> errors;
+  late Stream<List<int>> transformed;
+  late StreamSubscription<List<int>> subscription;
+
+  void setUpForStreamTypes(String triggerType, String valuesType,
+      {required bool longPoll}) {
+    valuesCanceled = false;
+    triggerCanceled = false;
+    triggerPaused = false;
+    trigger = createController(triggerType)
+      ..onCancel = () {
+        triggerCanceled = true;
+      };
+    if (triggerType == 'single subscription') {
+      trigger.onPause = () {
+        triggerPaused = true;
+      };
+    }
+    values = createController(valuesType)
+      ..onCancel = () {
+        valuesCanceled = true;
+      };
+    emittedValues = [];
+    errors = [];
+    isDone = false;
+    transformed = values.stream.buffer(trigger.stream, longPoll: longPoll);
+    subscription =
+        transformed.listen(emittedValues.add, onError: errors.add, onDone: () {
+      isDone = true;
+    });
+  }
+
+  for (var triggerType in streamTypes) {
+    for (var valuesType in streamTypes) {
+      group('Trigger type: [$triggerType], Values type: [$valuesType]', () {
+        group('general behavior', () {
+          setUp(() {
+            setUpForStreamTypes(triggerType, valuesType, longPoll: true);
+          });
+
+          test('does not emit before `trigger`', () async {
+            values.add(1);
+            await Future(() {});
+            expect(emittedValues, isEmpty);
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, [
+              [1]
+            ]);
+          });
+
+          test('groups values between trigger', () async {
+            values
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            trigger.add(null);
+            values
+              ..add(3)
+              ..add(4);
+            await Future(() {});
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, [
+              [1, 2],
+              [3, 4]
+            ]);
+          });
+
+          test('cancels value subscription when output canceled', () async {
+            expect(valuesCanceled, false);
+            await subscription.cancel();
+            expect(valuesCanceled, true);
+          });
+
+          test('closes when trigger ends', () async {
+            expect(isDone, false);
+            await trigger.close();
+            await Future(() {});
+            expect(isDone, true);
+          });
+
+          test('closes after outputting final values when source closes',
+              () async {
+            expect(isDone, false);
+            values.add(1);
+            await values.close();
+            expect(isDone, false);
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, [
+              [1]
+            ]);
+            expect(isDone, true);
+          });
+
+          test('closes when source closes and there are no buffered', () async {
+            expect(isDone, false);
+            await values.close();
+            await Future(() {});
+            expect(isDone, true);
+          });
+
+          test('forwards errors from trigger', () async {
+            trigger.addError('error');
+            await Future(() {});
+            expect(errors, ['error']);
+          });
+
+          test('forwards errors from values', () async {
+            values.addError('error');
+            await Future(() {});
+            expect(errors, ['error']);
+          });
+        });
+
+        group('long polling', () {
+          setUp(() {
+            setUpForStreamTypes(triggerType, valuesType, longPoll: true);
+          });
+
+          test('emits immediately if trigger emits before a value', () async {
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, isEmpty);
+            values.add(1);
+            await Future(() {});
+            expect(emittedValues, [
+              [1]
+            ]);
+          });
+
+          test('two triggers in a row - emit buffere then emit next value',
+              () async {
+            values
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            trigger
+              ..add(null)
+              ..add(null);
+            await Future(() {});
+            values.add(3);
+            await Future(() {});
+            expect(emittedValues, [
+              [1, 2],
+              [3]
+            ]);
+          });
+
+          test('pre-emptive trigger then trigger after values', () async {
+            trigger.add(null);
+            await Future(() {});
+            values
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, [
+              [1],
+              [2]
+            ]);
+          });
+
+          test('multiple pre-emptive triggers, only emits first value',
+              () async {
+            trigger
+              ..add(null)
+              ..add(null);
+            await Future(() {});
+            values
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            expect(emittedValues, [
+              [1]
+            ]);
+          });
+
+          test('closes if there is no waiting long poll when source closes',
+              () async {
+            expect(isDone, false);
+            values.add(1);
+            trigger.add(null);
+            await values.close();
+            await Future(() {});
+            expect(isDone, true);
+          });
+
+          test('waits to emit if there waiting long poll when trigger closes',
+              () async {
+            trigger.add(null);
+            await trigger.close();
+            expect(isDone, false);
+            values.add(1);
+            await Future(() {});
+            expect(emittedValues, [
+              [1]
+            ]);
+            expect(isDone, true);
+          });
+        });
+
+        group('immediate polling', () {
+          setUp(() {
+            setUpForStreamTypes(triggerType, valuesType, longPoll: false);
+          });
+
+          test('emits empty list before values', () async {
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, [<int>[]]);
+          });
+
+          test('emits empty list after emitting values', () async {
+            values
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            trigger
+              ..add(null)
+              ..add(null);
+            await Future(() {});
+            expect(emittedValues, [
+              [1, 2],
+              <int>[]
+            ]);
+          });
+        });
+      });
+    }
+  }
+
+  test('always cancels trigger if values is singlesubscription', () async {
+    setUpForStreamTypes('broadcast', 'single subscription', longPoll: true);
+    expect(triggerCanceled, false);
+    await subscription.cancel();
+    expect(triggerCanceled, true);
+
+    setUpForStreamTypes('single subscription', 'single subscription',
+        longPoll: true);
+    expect(triggerCanceled, false);
+    await subscription.cancel();
+    expect(triggerCanceled, true);
+  });
+
+  test('cancels trigger if trigger is broadcast', () async {
+    setUpForStreamTypes('broadcast', 'broadcast', longPoll: true);
+    expect(triggerCanceled, false);
+    await subscription.cancel();
+    expect(triggerCanceled, true);
+  });
+
+  test('pauses single subscription trigger for broadcast values', () async {
+    setUpForStreamTypes('single subscription', 'broadcast', longPoll: true);
+    expect(triggerCanceled, false);
+    expect(triggerPaused, false);
+    await subscription.cancel();
+    expect(triggerCanceled, false);
+    expect(triggerPaused, true);
+  });
+
+  for (var triggerType in streamTypes) {
+    test('cancel and relisten with [$triggerType] trigger', () async {
+      setUpForStreamTypes(triggerType, 'broadcast', longPoll: true);
+      values.add(1);
+      trigger.add(null);
+      await Future(() {});
+      expect(emittedValues, [
+        [1]
+      ]);
+      await subscription.cancel();
+      values.add(2);
+      trigger.add(null);
+      await Future(() {});
+      subscription = transformed.listen(emittedValues.add);
+      values.add(3);
+      trigger.add(null);
+      await Future(() {});
+      expect(emittedValues, [
+        [1],
+        [3]
+      ]);
+    });
+  }
+}
diff --git a/pkgs/stream_transform/test/combine_latest_all_test.dart b/pkgs/stream_transform/test/combine_latest_all_test.dart
new file mode 100644
index 0000000..f4b719c
--- /dev/null
+++ b/pkgs/stream_transform/test/combine_latest_all_test.dart
@@ -0,0 +1,166 @@
+// 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';
+
+import 'package:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+Future<void> tick() => Future(() {});
+
+void main() {
+  group('combineLatestAll', () {
+    test('emits latest values', () async {
+      final first = StreamController<String>();
+      final second = StreamController<String>();
+      final third = StreamController<String>();
+      final combined = first.stream.combineLatestAll(
+          [second.stream, third.stream]).map((data) => data.join());
+
+      // first:    a----b------------------c--------d---|
+      // second:   --1---------2-----------------|
+      // third:    -------&----------%---|
+      // combined: -------b1&--b2&---b2%---c2%------d2%-|
+
+      expect(combined,
+          emitsInOrder(['b1&', 'b2&', 'b2%', 'c2%', 'd2%', emitsDone]));
+
+      first.add('a');
+      await tick();
+      second.add('1');
+      await tick();
+      first.add('b');
+      await tick();
+      third.add('&');
+      await tick();
+      second.add('2');
+      await tick();
+      third.add('%');
+      await tick();
+      await third.close();
+      await tick();
+      first.add('c');
+      await tick();
+      await second.close();
+      await tick();
+      first.add('d');
+      await tick();
+      await first.close();
+    });
+
+    test('ends if a Stream closes without ever emitting a value', () async {
+      final first = StreamController<String>();
+      final second = StreamController<String>();
+      final combined = first.stream.combineLatestAll([second.stream]);
+
+      // first:    -a------b-------|
+      // second:   -----|
+      // combined: -----|
+
+      expect(combined, emits(emitsDone));
+
+      first.add('a');
+      await tick();
+      await second.close();
+      await tick();
+      first.add('b');
+    });
+
+    test('forwards errors', () async {
+      final first = StreamController<String>();
+      final second = StreamController<String>();
+      final combined = first.stream
+          .combineLatestAll([second.stream]).map((data) => data.join());
+
+      // first:    -a---------|
+      // second:   ----1---#
+      // combined: ----a1--#
+
+      expect(combined, emitsThrough(emitsError('doh')));
+
+      first.add('a');
+      await tick();
+      second.add('1');
+      await tick();
+      second.addError('doh');
+    });
+
+    test('ends after both streams have ended', () async {
+      final first = StreamController<String>();
+      final second = StreamController<String>();
+
+      var done = false;
+      first.stream.combineLatestAll([second.stream]).listen(null,
+          onDone: () => done = true);
+
+      // first:    -a---|
+      // second:   --------1--|
+      // combined: --------a1-|
+
+      first.add('a');
+      await tick();
+      await first.close();
+      await tick();
+
+      expect(done, isFalse);
+
+      second.add('1');
+      await tick();
+      await second.close();
+      await tick();
+
+      expect(done, isTrue);
+    });
+
+    group('broadcast source', () {
+      test('can cancel and relisten to broadcast stream', () async {
+        final first = StreamController<String>.broadcast();
+        final second = StreamController<String>.broadcast();
+        final combined = first.stream
+            .combineLatestAll([second.stream]).map((data) => data.join());
+
+        // first:    a------b----------------c------d----e---|
+        // second:   --1---------2---3---4------5-|
+        // combined: --a1---b1---b2--b3--b4-----c5--d5---e5--|
+        // sub1:     ^-----------------!
+        // sub2:     ----------------------^-----------------|
+
+        expect(combined.take(4), emitsInOrder(['a1', 'b1', 'b2', 'b3']));
+
+        first.add('a');
+        await tick();
+        second.add('1');
+        await tick();
+        first.add('b');
+        await tick();
+        second.add('2');
+        await tick();
+        second.add('3');
+        await tick();
+
+        // First subscription is canceled here by .take(4)
+        expect(first.hasListener, isFalse);
+        expect(second.hasListener, isFalse);
+
+        // This emit is thrown away because there are no subscribers
+        second.add('4');
+        await tick();
+
+        expect(combined, emitsInOrder(['c5', 'd5', 'e5', emitsDone]));
+
+        first.add('c');
+        await tick();
+        second.add('5');
+        await tick();
+        await second.close();
+        await tick();
+        first.add('d');
+        await tick();
+        first.add('e');
+        await tick();
+        await first.close();
+      });
+    });
+  });
+}
diff --git a/pkgs/stream_transform/test/combine_latest_test.dart b/pkgs/stream_transform/test/combine_latest_test.dart
new file mode 100644
index 0000000..1985c75
--- /dev/null
+++ b/pkgs/stream_transform/test/combine_latest_test.dart
@@ -0,0 +1,179 @@
+// 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';
+
+import 'package:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+void main() {
+  group('combineLatest', () {
+    test('flows through combine callback', () async {
+      var source = StreamController<int>();
+      var other = StreamController<int>();
+      int sum(int a, int b) => a + b;
+
+      var results = <int>[];
+      unawaited(
+          source.stream.combineLatest(other.stream, sum).forEach(results.add));
+
+      source.add(1);
+      await Future(() {});
+      expect(results, isEmpty);
+
+      other.add(2);
+      await Future(() {});
+      expect(results, [3]);
+
+      source.add(3);
+      await Future(() {});
+      expect(results, [3, 5]);
+
+      source.add(4);
+      await Future(() {});
+      expect(results, [3, 5, 6]);
+
+      other.add(5);
+      await Future(() {});
+      expect(results, [3, 5, 6, 9]);
+    });
+
+    test('can combine different typed streams', () async {
+      var source = StreamController<String>();
+      var other = StreamController<int>();
+      String times(String a, int b) => a * b;
+
+      var results = <String>[];
+      unawaited(source.stream
+          .combineLatest(other.stream, times)
+          .forEach(results.add));
+
+      source
+        ..add('a')
+        ..add('b');
+      await Future(() {});
+      expect(results, isEmpty);
+
+      other.add(2);
+      await Future(() {});
+      expect(results, ['bb']);
+
+      other.add(3);
+      await Future(() {});
+      expect(results, ['bb', 'bbb']);
+
+      source.add('c');
+      await Future(() {});
+      expect(results, ['bb', 'bbb', 'ccc']);
+    });
+
+    test('ends after both streams have ended', () async {
+      var source = StreamController<int>();
+      var other = StreamController<int>();
+      int sum(int a, int b) => a + b;
+
+      var done = false;
+      source.stream
+          .combineLatest(other.stream, sum)
+          .listen(null, onDone: () => done = true);
+
+      source.add(1);
+
+      await source.close();
+      await Future(() {});
+      expect(done, false);
+
+      await other.close();
+      await Future(() {});
+      expect(done, true);
+    });
+
+    test('ends if source stream closes without ever emitting a value',
+        () async {
+      var source = const Stream<int>.empty();
+      var other = StreamController<int>();
+
+      int sum(int a, int b) => a + b;
+
+      var done = false;
+      source
+          .combineLatest(other.stream, sum)
+          .listen(null, onDone: () => done = true);
+
+      await Future(() {});
+      // Nothing can ever be emitted on the result, may as well close.
+      expect(done, true);
+    });
+
+    test('ends if other stream closes without ever emitting a value', () async {
+      var source = StreamController<int>();
+      var other = const Stream<int>.empty();
+
+      int sum(int a, int b) => a + b;
+
+      var done = false;
+      source.stream
+          .combineLatest(other, sum)
+          .listen(null, onDone: () => done = true);
+
+      await Future(() {});
+      // Nothing can ever be emitted on the result, may as well close.
+      expect(done, true);
+    });
+
+    test('forwards errors', () async {
+      var source = StreamController<int>();
+      var other = StreamController<int>();
+      int sum(int a, int b) => throw _NumberedException(3);
+
+      var errors = <Object>[];
+      source.stream
+          .combineLatest(other.stream, sum)
+          .listen(null, onError: errors.add);
+
+      source.addError(_NumberedException(1));
+      other.addError(_NumberedException(2));
+
+      source.add(1);
+      other.add(2);
+
+      await Future(() {});
+
+      expect(errors, [_isException(1), _isException(2), _isException(3)]);
+    });
+
+    group('broadcast source', () {
+      test('can cancel and relisten to broadcast stream', () async {
+        var source = StreamController<int>.broadcast();
+        var other = StreamController<int>();
+        int combine(int a, int b) => a + b;
+
+        var emittedValues = <int>[];
+        var transformed = source.stream.combineLatest(other.stream, combine);
+
+        var subscription = transformed.listen(emittedValues.add);
+
+        source.add(1);
+        other.add(2);
+        await Future(() {});
+        expect(emittedValues, [3]);
+
+        await subscription.cancel();
+
+        subscription = transformed.listen(emittedValues.add);
+        source.add(3);
+        await Future(() {});
+        expect(emittedValues, [3, 5]);
+      });
+    });
+  });
+}
+
+class _NumberedException implements Exception {
+  final int id;
+  _NumberedException(this.id);
+}
+
+Matcher _isException(int id) =>
+    const TypeMatcher<_NumberedException>().having((n) => n.id, 'id', id);
diff --git a/pkgs/stream_transform/test/concurrent_async_map_test.dart b/pkgs/stream_transform/test/concurrent_async_map_test.dart
new file mode 100644
index 0000000..1807f9f
--- /dev/null
+++ b/pkgs/stream_transform/test/concurrent_async_map_test.dart
@@ -0,0 +1,157 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  late StreamController<int> controller;
+  late List<String> emittedValues;
+  late bool valuesCanceled;
+  late bool isDone;
+  late List<String> errors;
+  late Stream<String> transformed;
+  late StreamSubscription<String> subscription;
+
+  late List<Completer<String>> finishWork;
+  late List<dynamic> values;
+
+  Future<String> convert(int value) {
+    values.add(value);
+    var completer = Completer<String>();
+    finishWork.add(completer);
+    return completer.future;
+  }
+
+  for (var streamType in streamTypes) {
+    group('concurrentAsyncMap for stream type: [$streamType]', () {
+      setUp(() {
+        valuesCanceled = false;
+        controller = createController(streamType)
+          ..onCancel = () {
+            valuesCanceled = true;
+          };
+        emittedValues = [];
+        errors = [];
+        isDone = false;
+        finishWork = [];
+        values = [];
+        transformed = controller.stream.concurrentAsyncMap(convert);
+        subscription = transformed
+            .listen(emittedValues.add, onError: errors.add, onDone: () {
+          isDone = true;
+        });
+      });
+
+      test('does not emit before convert finishes', () async {
+        controller.add(1);
+        await Future(() {});
+        expect(emittedValues, isEmpty);
+        expect(values, [1]);
+        finishWork.first.complete('result');
+        await Future(() {});
+        expect(emittedValues, ['result']);
+      });
+
+      test('allows calls to convert before the last one finished', () async {
+        controller
+          ..add(1)
+          ..add(2)
+          ..add(3);
+        await Future(() {});
+        expect(values, [1, 2, 3]);
+      });
+
+      test('forwards errors directly without waiting for previous convert',
+          () async {
+        controller.add(1);
+        await Future(() {});
+        controller.addError('error');
+        await Future(() {});
+        expect(errors, ['error']);
+      });
+
+      test('forwards errors which occur during the convert', () async {
+        controller.add(1);
+        await Future(() {});
+        finishWork.first.completeError('error');
+        await Future(() {});
+        expect(errors, ['error']);
+      });
+
+      test('can continue handling events after an error', () async {
+        controller.add(1);
+        await Future(() {});
+        finishWork[0].completeError('error');
+        controller.add(2);
+        await Future(() {});
+        expect(values, [1, 2]);
+        finishWork[1].completeError('another');
+        await Future(() {});
+        expect(errors, ['error', 'another']);
+      });
+
+      test('cancels value subscription when output canceled', () async {
+        expect(valuesCanceled, false);
+        await subscription.cancel();
+        expect(valuesCanceled, true);
+      });
+
+      test('closes when values end if no conversion is pending', () async {
+        expect(isDone, false);
+        await controller.close();
+        await Future(() {});
+        expect(isDone, true);
+      });
+
+      if (streamType == 'broadcast') {
+        test('multiple listeners all get values', () async {
+          var otherValues = <String>[];
+          transformed.listen(otherValues.add);
+          controller.add(1);
+          await Future(() {});
+          finishWork.first.complete('result');
+          await Future(() {});
+          expect(emittedValues, ['result']);
+          expect(otherValues, ['result']);
+        });
+
+        test('multiple listeners get done when values end', () async {
+          var otherDone = false;
+          transformed.listen(null, onDone: () => otherDone = true);
+          controller.add(1);
+          await Future(() {});
+          await controller.close();
+          expect(isDone, false);
+          expect(otherDone, false);
+          finishWork.first.complete('');
+          await Future(() {});
+          expect(isDone, true);
+          expect(otherDone, true);
+        });
+
+        test('can cancel and relisten', () async {
+          controller.add(1);
+          await Future(() {});
+          finishWork.first.complete('first');
+          await Future(() {});
+          await subscription.cancel();
+          controller.add(2);
+          await Future(() {});
+          subscription = transformed.listen(emittedValues.add);
+          controller.add(3);
+          await Future(() {});
+          expect(values, [1, 3]);
+          finishWork[1].complete('second');
+          await Future(() {});
+          expect(emittedValues, ['first', 'second']);
+        });
+      }
+    });
+  }
+}
diff --git a/pkgs/stream_transform/test/debounce_test.dart b/pkgs/stream_transform/test/debounce_test.dart
new file mode 100644
index 0000000..19de055
--- /dev/null
+++ b/pkgs/stream_transform/test/debounce_test.dart
@@ -0,0 +1,310 @@
+// 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:fake_async/fake_async.dart';
+import 'package:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  for (var streamType in streamTypes) {
+    group('Stream type [$streamType]', () {
+      group('debounce - trailing', () {
+        late StreamController<int> values;
+        late List<int> emittedValues;
+        late bool valuesCanceled;
+        late bool isDone;
+        late List<String> errors;
+        late StreamSubscription<int> subscription;
+        late Stream<int> transformed;
+
+        setUp(() async {
+          valuesCanceled = false;
+          values = createController(streamType)
+            ..onCancel = () {
+              valuesCanceled = true;
+            };
+          emittedValues = [];
+          errors = [];
+          isDone = false;
+          transformed = values.stream.debounce(const Duration(milliseconds: 5));
+        });
+
+        void listen() {
+          subscription = transformed
+              .listen(emittedValues.add, onError: errors.add, onDone: () {
+            isDone = true;
+          });
+        }
+
+        test('cancels values', () async {
+          listen();
+          await subscription.cancel();
+          expect(valuesCanceled, true);
+        });
+
+        test('swallows values that come faster than duration', () {
+          fakeAsync((async) {
+            listen();
+            values
+              ..add(1)
+              ..add(2)
+              ..close();
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [2]);
+          });
+        });
+
+        test('outputs multiple values spaced further than duration', () {
+          fakeAsync((async) {
+            listen();
+            values.add(1);
+            async.elapse(const Duration(milliseconds: 6));
+            values.add(2);
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [1, 2]);
+          });
+        });
+
+        test('waits for pending value to close', () {
+          fakeAsync((async) {
+            listen();
+            values.add(1);
+            async.elapse(const Duration(milliseconds: 6));
+            values.close();
+            async.flushMicrotasks();
+            expect(isDone, true);
+          });
+        });
+
+        test('closes output if there are no pending values', () {
+          fakeAsync((async) {
+            listen();
+            values.add(1);
+            async.elapse(const Duration(milliseconds: 6));
+            values
+              ..add(2)
+              ..close();
+            async.flushMicrotasks();
+            expect(isDone, false);
+            async.elapse(const Duration(milliseconds: 6));
+            expect(isDone, true);
+          });
+        });
+
+        if (streamType == 'broadcast') {
+          test('multiple listeners all get values', () {
+            fakeAsync((async) {
+              listen();
+              var otherValues = <int>[];
+              transformed.listen(otherValues.add);
+              values
+                ..add(1)
+                ..add(2);
+              async.elapse(const Duration(milliseconds: 6));
+              expect(emittedValues, [2]);
+              expect(otherValues, [2]);
+            });
+          });
+        }
+      });
+
+      group('debounce - leading', () {
+        late StreamController<int> values;
+        late List<int> emittedValues;
+        late Stream<int> transformed;
+        late bool isDone;
+
+        setUp(() async {
+          values = createController(streamType);
+          emittedValues = [];
+          isDone = false;
+          transformed = values.stream.debounce(const Duration(milliseconds: 5),
+              leading: true, trailing: false);
+        });
+
+        void listen() {
+          transformed.listen(emittedValues.add, onDone: () {
+            isDone = true;
+          });
+        }
+
+        test('swallows values that come faster than duration', () async {
+          listen();
+          values
+            ..add(1)
+            ..add(2);
+          await values.close();
+          expect(emittedValues, [1]);
+        });
+
+        test('outputs multiple values spaced further than duration', () {
+          fakeAsync((async) {
+            listen();
+            values.add(1);
+            async.elapse(const Duration(milliseconds: 6));
+            values.add(2);
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [1, 2]);
+          });
+        });
+
+        if (streamType == 'broadcast') {
+          test('multiple listeners all get values', () {
+            fakeAsync((async) {
+              listen();
+              var otherValues = <int>[];
+              transformed.listen(otherValues.add);
+              values
+                ..add(1)
+                ..add(2);
+              async.elapse(const Duration(milliseconds: 6));
+              expect(emittedValues, [1]);
+              expect(otherValues, [1]);
+            });
+          });
+        }
+
+        test('closes output immediately if not waiting for trailing value',
+            () async {
+          listen();
+          values.add(1);
+          await values.close();
+          expect(isDone, true);
+        });
+      });
+
+      group('debounce - leading and trailing', () {
+        late StreamController<int> values;
+        late List<int> emittedValues;
+        late Stream<int> transformed;
+
+        setUp(() async {
+          values = createController(streamType);
+          emittedValues = [];
+          transformed = values.stream.debounce(const Duration(milliseconds: 5),
+              leading: true, trailing: true);
+        });
+        void listen() {
+          transformed.listen(emittedValues.add);
+        }
+
+        test('swallows values that come faster than duration', () {
+          fakeAsync((async) {
+            listen();
+            values
+              ..add(1)
+              ..add(2)
+              ..add(3)
+              ..close();
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [1, 3]);
+          });
+        });
+
+        test('outputs multiple values spaced further than duration', () {
+          fakeAsync((async) {
+            listen();
+            values.add(1);
+            async.elapse(const Duration(milliseconds: 6));
+            values.add(2);
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [1, 2]);
+          });
+        });
+
+        if (streamType == 'broadcast') {
+          test('multiple listeners all get values', () {
+            fakeAsync((async) {
+              listen();
+              var otherValues = <int>[];
+              transformed.listen(otherValues.add);
+              values
+                ..add(1)
+                ..add(2);
+              async.elapse(const Duration(milliseconds: 6));
+              expect(emittedValues, [1, 2]);
+              expect(otherValues, [1, 2]);
+            });
+          });
+        }
+      });
+
+      group('debounceBuffer', () {
+        late StreamController<int> values;
+        late List<List<int>> emittedValues;
+        late List<String> errors;
+        late Stream<List<int>> transformed;
+
+        setUp(() async {
+          values = createController(streamType);
+          emittedValues = [];
+          errors = [];
+          transformed =
+              values.stream.debounceBuffer(const Duration(milliseconds: 5));
+        });
+        void listen() {
+          transformed.listen(emittedValues.add, onError: errors.add);
+        }
+
+        test('Emits all values as a list', () {
+          fakeAsync((async) {
+            listen();
+            values
+              ..add(1)
+              ..add(2)
+              ..close();
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [
+              [1, 2]
+            ]);
+          });
+        });
+
+        test('separate lists for multiple values spaced further than duration',
+            () {
+          fakeAsync((async) {
+            listen();
+            values.add(1);
+            async.elapse(const Duration(milliseconds: 6));
+            values.add(2);
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [
+              [1],
+              [2]
+            ]);
+          });
+        });
+
+        if (streamType == 'broadcast') {
+          test('multiple listeners all get values', () {
+            fakeAsync((async) {
+              listen();
+              var otherValues = <List<int>>[];
+              transformed.listen(otherValues.add);
+              values
+                ..add(1)
+                ..add(2);
+              async.elapse(const Duration(milliseconds: 6));
+              expect(emittedValues, [
+                [1, 2]
+              ]);
+              expect(otherValues, [
+                [1, 2]
+              ]);
+            });
+          });
+        }
+      });
+    });
+  }
+  test('allows nulls', () async {
+    final values = Stream<int?>.fromIterable([null]);
+    final transformed = values.debounce(const Duration(milliseconds: 1));
+    expect(await transformed.toList(), [null]);
+  });
+}
diff --git a/pkgs/stream_transform/test/followd_by_test.dart b/pkgs/stream_transform/test/followd_by_test.dart
new file mode 100644
index 0000000..d600d13
--- /dev/null
+++ b/pkgs/stream_transform/test/followd_by_test.dart
@@ -0,0 +1,159 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  for (var firstType in streamTypes) {
+    for (var secondType in streamTypes) {
+      group('followedBy [$firstType] with [$secondType]', () {
+        late StreamController<int> first;
+        late StreamController<int> second;
+
+        late List<int> emittedValues;
+        late bool firstCanceled;
+        late bool secondCanceled;
+        late bool secondListened;
+        late bool isDone;
+        late List<String> errors;
+        late Stream<int> transformed;
+        late StreamSubscription<int> subscription;
+
+        setUp(() async {
+          firstCanceled = false;
+          secondCanceled = false;
+          secondListened = false;
+          first = createController(firstType)
+            ..onCancel = () {
+              firstCanceled = true;
+            };
+          second = createController(secondType)
+            ..onCancel = () {
+              secondCanceled = true;
+            }
+            ..onListen = () {
+              secondListened = true;
+            };
+          emittedValues = [];
+          errors = [];
+          isDone = false;
+          transformed = first.stream.followedBy(second.stream);
+          subscription = transformed
+              .listen(emittedValues.add, onError: errors.add, onDone: () {
+            isDone = true;
+          });
+        });
+
+        test('adds all values from both streams', () async {
+          first
+            ..add(1)
+            ..add(2);
+          await first.close();
+          await Future(() {});
+          second
+            ..add(3)
+            ..add(4);
+          await Future(() {});
+          expect(emittedValues, [1, 2, 3, 4]);
+        });
+
+        test('Does not listen to second stream before first stream finishes',
+            () async {
+          expect(secondListened, false);
+          await first.close();
+          expect(secondListened, true);
+        });
+
+        test('closes stream after both inputs close', () async {
+          await first.close();
+          await second.close();
+          expect(isDone, true);
+        });
+
+        test('cancels any type of first stream on cancel', () async {
+          await subscription.cancel();
+          expect(firstCanceled, true);
+        });
+
+        if (firstType == 'single subscription') {
+          test(
+              'cancels any type of second stream on cancel if first is '
+              'broadcast', () async {
+            await first.close();
+            await subscription.cancel();
+            expect(secondCanceled, true);
+          });
+
+          if (secondType == 'broadcast') {
+            test('can pause and resume during second stream - dropping values',
+                () async {
+              await first.close();
+              subscription.pause();
+              second.add(1);
+              await Future(() {});
+              subscription.resume();
+              second.add(2);
+              await Future(() {});
+              expect(emittedValues, [2]);
+            });
+          } else {
+            test('can pause and resume during second stream - buffering values',
+                () async {
+              await first.close();
+              subscription.pause();
+              second.add(1);
+              await Future(() {});
+              subscription.resume();
+              second.add(2);
+              await Future(() {});
+              expect(emittedValues, [1, 2]);
+            });
+          }
+        }
+
+        if (firstType == 'broadcast') {
+          test('can cancel and relisten during first stream', () async {
+            await subscription.cancel();
+            first.add(1);
+            subscription = transformed.listen(emittedValues.add);
+            first.add(2);
+            await Future(() {});
+            expect(emittedValues, [2]);
+          });
+
+          test('can cancel and relisten during second stream', () async {
+            await first.close();
+            await subscription.cancel();
+            second.add(2);
+            await Future(() {});
+            subscription = transformed.listen(emittedValues.add);
+            second.add(3);
+            await Future(() {});
+            expect(emittedValues, [3]);
+          });
+
+          test('forwards values to multiple listeners', () async {
+            var otherValues = <int>[];
+            transformed.listen(otherValues.add);
+            first.add(1);
+            await first.close();
+            second.add(2);
+            await Future(() {});
+            var thirdValues = <int>[];
+            transformed.listen(thirdValues.add);
+            second.add(3);
+            await Future(() {});
+            expect(emittedValues, [1, 2, 3]);
+            expect(otherValues, [1, 2, 3]);
+            expect(thirdValues, [3]);
+          });
+        }
+      });
+    }
+  }
+}
diff --git a/pkgs/stream_transform/test/from_handlers_test.dart b/pkgs/stream_transform/test/from_handlers_test.dart
new file mode 100644
index 0000000..694199c
--- /dev/null
+++ b/pkgs/stream_transform/test/from_handlers_test.dart
@@ -0,0 +1,183 @@
+// 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:stream_transform/src/from_handlers.dart';
+import 'package:test/test.dart';
+
+void main() {
+  late StreamController<int> values;
+  late List<int> emittedValues;
+  late bool valuesCanceled;
+  late bool isDone;
+  late List<String> errors;
+  late Stream<int> transformed;
+  late StreamSubscription<int> subscription;
+
+  void setUpForController(StreamController<int> controller,
+      Stream<int> Function(Stream<int>) transform) {
+    valuesCanceled = false;
+    values = controller
+      ..onCancel = () {
+        valuesCanceled = true;
+      };
+    emittedValues = [];
+    errors = [];
+    isDone = false;
+    transformed = transform(values.stream);
+    subscription =
+        transformed.listen(emittedValues.add, onError: errors.add, onDone: () {
+      isDone = true;
+    });
+  }
+
+  group('default from_handlers', () {
+    group('Single subscription stream', () {
+      setUp(() {
+        setUpForController(StreamController(),
+            (s) => s.transformByHandlers(onData: (e, sink) => sink.add(e)));
+      });
+
+      test('has correct stream type', () {
+        expect(transformed.isBroadcast, false);
+      });
+
+      test('forwards values', () async {
+        values
+          ..add(1)
+          ..add(2);
+        await Future(() {});
+        expect(emittedValues, [1, 2]);
+      });
+
+      test('forwards errors', () async {
+        values.addError('error');
+        await Future(() {});
+        expect(errors, ['error']);
+      });
+
+      test('forwards done', () async {
+        await values.close();
+        expect(isDone, true);
+      });
+
+      test('forwards cancel', () async {
+        await subscription.cancel();
+        expect(valuesCanceled, true);
+      });
+    });
+
+    group('broadcast stream with muliple listeners', () {
+      late List<int> emittedValues2;
+      late List<String> errors2;
+      late bool isDone2;
+      late StreamSubscription<int> subscription2;
+
+      setUp(() {
+        setUpForController(StreamController.broadcast(),
+            (s) => s.transformByHandlers(onData: (e, sink) => sink.add(e)));
+        emittedValues2 = [];
+        errors2 = [];
+        isDone2 = false;
+        subscription2 = transformed
+            .listen(emittedValues2.add, onError: errors2.add, onDone: () {
+          isDone2 = true;
+        });
+      });
+
+      test('has correct stream type', () {
+        expect(transformed.isBroadcast, true);
+      });
+
+      test('forwards values', () async {
+        values
+          ..add(1)
+          ..add(2);
+        await Future(() {});
+        expect(emittedValues, [1, 2]);
+        expect(emittedValues2, [1, 2]);
+      });
+
+      test('forwards errors', () async {
+        values.addError('error');
+        await Future(() {});
+        expect(errors, ['error']);
+        expect(errors2, ['error']);
+      });
+
+      test('forwards done', () async {
+        await values.close();
+        expect(isDone, true);
+        expect(isDone2, true);
+      });
+
+      test('forwards cancel', () async {
+        await subscription.cancel();
+        expect(valuesCanceled, false);
+        await subscription2.cancel();
+        expect(valuesCanceled, true);
+      });
+    });
+  });
+
+  group('custom handlers', () {
+    group('single subscription', () {
+      setUp(() async {
+        setUpForController(
+            StreamController(),
+            (s) => s.transformByHandlers(onData: (value, sink) {
+                  sink.add(value + 1);
+                }));
+      });
+      test('uses transform from handleData', () async {
+        values
+          ..add(1)
+          ..add(2);
+        await Future(() {});
+        expect(emittedValues, [2, 3]);
+      });
+    });
+
+    group('broadcast stream with multiple listeners', () {
+      late int dataCallCount;
+      late int doneCallCount;
+      late int errorCallCount;
+
+      setUp(() async {
+        dataCallCount = 0;
+        doneCallCount = 0;
+        errorCallCount = 0;
+        setUpForController(
+            StreamController.broadcast(),
+            (s) => s.transformByHandlers(onData: (value, sink) {
+                  dataCallCount++;
+                }, onError: (error, stackTrace, sink) {
+                  errorCallCount++;
+                  sink.addError(error, stackTrace);
+                }, onDone: (sink) {
+                  doneCallCount++;
+                }));
+        transformed.listen((_) {}, onError: (_, __) {});
+      });
+
+      test('handles data once', () async {
+        values.add(1);
+        await Future(() {});
+        expect(dataCallCount, 1);
+      });
+
+      test('handles done once', () async {
+        await values.close();
+        expect(doneCallCount, 1);
+      });
+
+      test('handles errors once', () async {
+        values.addError('error');
+        await Future(() {});
+        expect(errorCallCount, 1);
+      });
+    });
+  });
+}
diff --git a/pkgs/stream_transform/test/merge_test.dart b/pkgs/stream_transform/test/merge_test.dart
new file mode 100644
index 0000000..ecbf97f
--- /dev/null
+++ b/pkgs/stream_transform/test/merge_test.dart
@@ -0,0 +1,140 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+void main() {
+  group('merge', () {
+    test('includes all values', () async {
+      var first = Stream.fromIterable([1, 2, 3]);
+      var second = Stream.fromIterable([4, 5, 6]);
+      var allValues = await first.merge(second).toList();
+      expect(allValues, containsAllInOrder([1, 2, 3]));
+      expect(allValues, containsAllInOrder([4, 5, 6]));
+      expect(allValues, hasLength(6));
+    });
+
+    test('cancels both sources', () async {
+      var firstCanceled = false;
+      var first = StreamController<int>()
+        ..onCancel = () {
+          firstCanceled = true;
+        };
+      var secondCanceled = false;
+      var second = StreamController<int>()
+        ..onCancel = () {
+          secondCanceled = true;
+        };
+      var subscription = first.stream.merge(second.stream).listen((_) {});
+      await subscription.cancel();
+      expect(firstCanceled, true);
+      expect(secondCanceled, true);
+    });
+
+    test('completes when both sources complete', () async {
+      var first = StreamController<int>();
+      var second = StreamController<int>();
+      var isDone = false;
+      first.stream.merge(second.stream).listen((_) {}, onDone: () {
+        isDone = true;
+      });
+      await first.close();
+      expect(isDone, false);
+      await second.close();
+      expect(isDone, true);
+    });
+
+    test('can cancel and relisten to broadcast stream', () async {
+      var first = StreamController<int>.broadcast();
+      var second = StreamController<int>();
+      var emittedValues = <int>[];
+      var transformed = first.stream.merge(second.stream);
+      var subscription = transformed.listen(emittedValues.add);
+      first.add(1);
+      second.add(2);
+      await Future(() {});
+      expect(emittedValues, contains(1));
+      expect(emittedValues, contains(2));
+      await subscription.cancel();
+      emittedValues = [];
+      subscription = transformed.listen(emittedValues.add);
+      first.add(3);
+      second.add(4);
+      await Future(() {});
+      expect(emittedValues, contains(3));
+      expect(emittedValues, contains(4));
+    });
+  });
+
+  group('mergeAll', () {
+    test('includes all values', () async {
+      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.mergeAll([second, third]).toList();
+      expect(allValues, containsAllInOrder([1, 2, 3]));
+      expect(allValues, containsAllInOrder([4, 5, 6]));
+      expect(allValues, containsAllInOrder([7, 8, 9]));
+      expect(allValues, hasLength(9));
+    });
+
+    test('handles mix of broadcast and single-subscription', () async {
+      var firstCanceled = false;
+      var first = StreamController<int>.broadcast()
+        ..onCancel = () {
+          firstCanceled = true;
+        };
+      var secondBroadcastCanceled = false;
+      var secondBroadcast = StreamController<int>.broadcast()
+        ..onCancel = () {
+          secondBroadcastCanceled = true;
+        };
+      var secondSingleCanceled = false;
+      var secondSingle = StreamController<int>()
+        ..onCancel = () {
+          secondSingleCanceled = true;
+        };
+
+      var merged =
+          first.stream.mergeAll([secondBroadcast.stream, secondSingle.stream]);
+
+      var firstListenerValues = <int>[];
+      var secondListenerValues = <int>[];
+
+      var firstSubscription = merged.listen(firstListenerValues.add);
+      var secondSubscription = merged.listen(secondListenerValues.add);
+
+      first.add(1);
+      secondBroadcast.add(2);
+      secondSingle.add(3);
+
+      await Future(() {});
+      await firstSubscription.cancel();
+
+      expect(firstCanceled, false);
+      expect(secondBroadcastCanceled, false);
+      expect(secondSingleCanceled, false);
+
+      first.add(4);
+      secondBroadcast.add(5);
+      secondSingle.add(6);
+
+      await Future(() {});
+      await secondSubscription.cancel();
+
+      await Future(() {});
+      expect(firstCanceled, true);
+      expect(secondBroadcastCanceled, true);
+      expect(secondSingleCanceled, false,
+          reason: 'Single subscription streams merged into broadcast streams '
+              'are not canceled');
+
+      expect(firstListenerValues, [1, 2, 3]);
+      expect(secondListenerValues, [1, 2, 3, 4, 5, 6]);
+    });
+  });
+}
diff --git a/pkgs/stream_transform/test/sample_test.dart b/pkgs/stream_transform/test/sample_test.dart
new file mode 100644
index 0000000..66ca09d
--- /dev/null
+++ b/pkgs/stream_transform/test/sample_test.dart
@@ -0,0 +1,291 @@
+// Copyright (c) 2022, 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  late StreamController<void> trigger;
+  late StreamController<int> values;
+  late List<int> emittedValues;
+  late bool valuesCanceled;
+  late bool triggerCanceled;
+  late bool triggerPaused;
+  late bool isDone;
+  late List<String> errors;
+  late Stream<int> transformed;
+  late StreamSubscription<int> subscription;
+
+  void setUpForStreamTypes(String triggerType, String valuesType,
+      {required bool longPoll}) {
+    valuesCanceled = false;
+    triggerCanceled = false;
+    triggerPaused = false;
+    trigger = createController(triggerType)
+      ..onCancel = () {
+        triggerCanceled = true;
+      };
+    if (triggerType == 'single subscription') {
+      trigger.onPause = () {
+        triggerPaused = true;
+      };
+    }
+    values = createController(valuesType)
+      ..onCancel = () {
+        valuesCanceled = true;
+      };
+    emittedValues = [];
+    errors = [];
+    isDone = false;
+    transformed = values.stream.sample(trigger.stream, longPoll: longPoll);
+    subscription =
+        transformed.listen(emittedValues.add, onError: errors.add, onDone: () {
+      isDone = true;
+    });
+  }
+
+  for (var triggerType in streamTypes) {
+    for (var valuesType in streamTypes) {
+      group('Trigger type: [$triggerType], Values type: [$valuesType]', () {
+        group('general behavior', () {
+          setUp(() {
+            setUpForStreamTypes(triggerType, valuesType, longPoll: true);
+          });
+
+          test('does not emit before `trigger`', () async {
+            values.add(1);
+            await Future(() {});
+            expect(emittedValues, isEmpty);
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, [1]);
+          });
+
+          test('keeps most recent event between triggers', () async {
+            values
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            trigger.add(null);
+            values
+              ..add(3)
+              ..add(4);
+            await Future(() {});
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, [2, 4]);
+          });
+
+          test('cancels value subscription when output canceled', () async {
+            expect(valuesCanceled, false);
+            await subscription.cancel();
+            expect(valuesCanceled, true);
+          });
+
+          test('closes when trigger ends', () async {
+            expect(isDone, false);
+            await trigger.close();
+            await Future(() {});
+            expect(isDone, true);
+          });
+
+          test('closes after outputting final values when source closes',
+              () async {
+            expect(isDone, false);
+            values.add(1);
+            await values.close();
+            expect(isDone, false);
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, [1]);
+            expect(isDone, true);
+          });
+
+          test('closes when source closes and there is no pending', () async {
+            expect(isDone, false);
+            await values.close();
+            await Future(() {});
+            expect(isDone, true);
+          });
+
+          test('forwards errors from trigger', () async {
+            trigger.addError('error');
+            await Future(() {});
+            expect(errors, ['error']);
+          });
+
+          test('forwards errors from values', () async {
+            values.addError('error');
+            await Future(() {});
+            expect(errors, ['error']);
+          });
+        });
+
+        group('long polling', () {
+          setUp(() {
+            setUpForStreamTypes(triggerType, valuesType, longPoll: true);
+          });
+
+          test('emits immediately if trigger emits before a value', () async {
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, isEmpty);
+            values.add(1);
+            await Future(() {});
+            expect(emittedValues, [1]);
+          });
+
+          test('two triggers in a row - emit buffere then emit next value',
+              () async {
+            values
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            trigger
+              ..add(null)
+              ..add(null);
+            await Future(() {});
+            values.add(3);
+            await Future(() {});
+            expect(emittedValues, [2, 3]);
+          });
+
+          test('pre-emptive trigger then trigger after values', () async {
+            trigger.add(null);
+            await Future(() {});
+            values
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, [1, 2]);
+          });
+
+          test('multiple pre-emptive triggers, only emits first value',
+              () async {
+            trigger
+              ..add(null)
+              ..add(null);
+            await Future(() {});
+            values
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            expect(emittedValues, [1]);
+          });
+
+          test('closes if there is no waiting long poll when source closes',
+              () async {
+            expect(isDone, false);
+            values.add(1);
+            trigger.add(null);
+            await values.close();
+            await Future(() {});
+            expect(isDone, true);
+          });
+
+          test('waits to emit if there waiting long poll when trigger closes',
+              () async {
+            trigger.add(null);
+            await trigger.close();
+            expect(isDone, false);
+            values.add(1);
+            await Future(() {});
+            expect(emittedValues, [1]);
+            expect(isDone, true);
+          });
+        });
+
+        group('immediate polling', () {
+          setUp(() {
+            setUpForStreamTypes(triggerType, valuesType, longPoll: false);
+          });
+
+          test('ignores trigger before values', () async {
+            trigger.add(null);
+            await Future(() {});
+            values
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, [2]);
+          });
+
+          test('ignores trigger if no pending values', () async {
+            values
+              ..add(1)
+              ..add(2);
+            await Future(() {});
+            trigger
+              ..add(null)
+              ..add(null);
+            await Future(() {});
+            values
+              ..add(3)
+              ..add(4);
+            await Future(() {});
+            trigger.add(null);
+            await Future(() {});
+            expect(emittedValues, [2, 4]);
+          });
+        });
+      });
+    }
+  }
+
+  test('always cancels trigger if values is singlesubscription', () async {
+    setUpForStreamTypes('broadcast', 'single subscription', longPoll: true);
+    expect(triggerCanceled, false);
+    await subscription.cancel();
+    expect(triggerCanceled, true);
+
+    setUpForStreamTypes('single subscription', 'single subscription',
+        longPoll: true);
+    expect(triggerCanceled, false);
+    await subscription.cancel();
+    expect(triggerCanceled, true);
+  });
+
+  test('cancels trigger if trigger is broadcast', () async {
+    setUpForStreamTypes('broadcast', 'broadcast', longPoll: true);
+    expect(triggerCanceled, false);
+    await subscription.cancel();
+    expect(triggerCanceled, true);
+  });
+
+  test('pauses single subscription trigger for broadcast values', () async {
+    setUpForStreamTypes('single subscription', 'broadcast', longPoll: true);
+    expect(triggerCanceled, false);
+    expect(triggerPaused, false);
+    await subscription.cancel();
+    expect(triggerCanceled, false);
+    expect(triggerPaused, true);
+  });
+
+  for (var triggerType in streamTypes) {
+    test('cancel and relisten with [$triggerType] trigger', () async {
+      setUpForStreamTypes(triggerType, 'broadcast', longPoll: true);
+      values.add(1);
+      trigger.add(null);
+      await Future(() {});
+      expect(emittedValues, [1]);
+      await subscription.cancel();
+      values.add(2);
+      trigger.add(null);
+      await Future(() {});
+      subscription = transformed.listen(emittedValues.add);
+      values.add(3);
+      trigger.add(null);
+      await Future(() {});
+      expect(emittedValues, [1, 3]);
+    });
+  }
+}
diff --git a/pkgs/stream_transform/test/scan_test.dart b/pkgs/stream_transform/test/scan_test.dart
new file mode 100644
index 0000000..3c749e7
--- /dev/null
+++ b/pkgs/stream_transform/test/scan_test.dart
@@ -0,0 +1,109 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+void main() {
+  group('Scan', () {
+    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.scan(0, sum).toList();
+
+      expect(result, [1, 3, 6, 10]);
+    });
+
+    test('can create a broadcast stream', () {
+      var source = StreamController<int>.broadcast();
+
+      var transformed = source.stream.scan(null, (_, __) {});
+
+      expect(transformed.isBroadcast, true);
+    });
+
+    test('forwards errors from source', () async {
+      var source = StreamController<int>();
+
+      int sum(int x, int y) => x + y;
+
+      var errors = <Object>[];
+
+      source.stream.scan(0, sum).listen(null, onError: errors.add);
+
+      source.addError(StateError('fail'));
+      await Future(() {});
+
+      expect(errors, [isStateError]);
+    });
+
+    group('with async combine', () {
+      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.scan(0, sum).toList();
+
+        expect(result, [1, 3, 6, 10]);
+      });
+
+      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.scan<Future<int>>(Future.value(0), sum).toList();
+
+        expect(result, [
+          const TypeMatcher<Future<void>>(),
+          const TypeMatcher<Future<void>>()
+        ]);
+        expect(await result.wait, [1, 3]);
+      });
+
+      test('does not call for subsequent values while waiting', () async {
+        var source = StreamController<int>();
+
+        var calledWith = <int>[];
+        var block = Completer<void>();
+        Future<int> combine(int x, int y) async {
+          calledWith.add(y);
+          await block.future;
+          return x + y;
+        }
+
+        var results = <int>[];
+
+        unawaited(source.stream.scan(0, combine).forEach(results.add));
+
+        source
+          ..add(1)
+          ..add(2);
+        await Future(() {});
+        expect(calledWith, [1]);
+        expect(results, isEmpty);
+
+        block.complete();
+        await Future(() {});
+        expect(calledWith, [1, 2]);
+        expect(results, [1, 3]);
+      });
+
+      test('forwards async errors', () async {
+        var source = StreamController<int>();
+
+        Future<int> combine(int x, int y) async => throw StateError('fail');
+
+        var errors = <Object>[];
+
+        source.stream.scan(0, combine).listen(null, onError: errors.add);
+
+        source.add(1);
+        await Future(() {});
+
+        expect(errors, [isStateError]);
+      });
+    });
+  });
+}
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..35f0330
--- /dev/null
+++ b/pkgs/stream_transform/test/start_with_test.dart
@@ -0,0 +1,167 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  late StreamController<int> values;
+  late Stream<int> transformed;
+  late StreamSubscription<int> subscription;
+
+  late List<int> emittedValues;
+  late bool isDone;
+
+  void setupForStreamType(
+      String streamType, Stream<int> Function(Stream<int>) transform) {
+    emittedValues = [];
+    isDone = false;
+    values = createController(streamType);
+    transformed = transform(values.stream);
+    subscription =
+        transformed.listen(emittedValues.add, onDone: () => isDone = true);
+  }
+
+  for (var streamType in streamTypes) {
+    group('startWith then [$streamType]', () {
+      setUp(() => setupForStreamType(streamType, (s) => s.startWith(1)));
+
+      test('outputs all values', () async {
+        values
+          ..add(2)
+          ..add(3);
+        await 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 Future(() {});
+          await subscription.cancel();
+          subscription = transformed.listen(emittedValues.add);
+          values.add(3);
+          await Future(() {});
+          await Future(() {});
+          expect(emittedValues, [1, 2, 3]);
+        });
+      }
+    });
+
+    group('startWithMany then [$streamType]', () {
+      setUp(() async {
+        setupForStreamType(streamType, (s) => s.startWithMany([1, 2]));
+        // Ensure all initial values go through
+        await Future(() {});
+      });
+
+      test('outputs all values', () async {
+        values
+          ..add(3)
+          ..add(4);
+        await 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 Future(() {});
+          await subscription.cancel();
+          subscription = transformed.listen(emittedValues.add);
+          values.add(4);
+          await Future(() {});
+          expect(emittedValues, [1, 2, 3, 4]);
+        });
+      }
+    });
+
+    for (var startingStreamType in streamTypes) {
+      group('startWithStream [$startingStreamType] then [$streamType]', () {
+        late StreamController<int> starting;
+        setUp(() async {
+          starting = createController(startingStreamType);
+          setupForStreamType(
+              streamType, (s) => s.startWithStream(starting.stream));
+        });
+
+        test('outputs all values', () async {
+          starting
+            ..add(1)
+            ..add(2);
+          await starting.close();
+          values
+            ..add(3)
+            ..add(4);
+          await 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 Future(() {});
+            await subscription.cancel();
+            subscription = transformed.listen(emittedValues.add);
+            starting.add(2);
+            await starting.close();
+            values
+              ..add(3)
+              ..add(4);
+            await 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 Future(() {});
+            await subscription.cancel();
+            subscription = transformed.listen(emittedValues.add);
+            values.add(4);
+            await Future(() {});
+            expect(emittedValues, [1, 2, 3, 4]);
+          });
+        }
+      });
+    }
+  }
+}
diff --git a/pkgs/stream_transform/test/switch_test.dart b/pkgs/stream_transform/test/switch_test.dart
new file mode 100644
index 0000000..9e70c08
--- /dev/null
+++ b/pkgs/stream_transform/test/switch_test.dart
@@ -0,0 +1,229 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  for (var outerType in streamTypes) {
+    for (var innerType in streamTypes) {
+      group('Outer type: [$outerType], Inner type: [$innerType]', () {
+        late StreamController<int> first;
+        late StreamController<int> second;
+        late StreamController<int> third;
+        late StreamController<Stream<int>> outer;
+
+        late List<int> emittedValues;
+        late bool firstCanceled;
+        late bool outerCanceled;
+        late bool isDone;
+        late List<String> errors;
+        late StreamSubscription<int> subscription;
+
+        setUp(() async {
+          firstCanceled = false;
+          outerCanceled = false;
+          outer = createController(outerType)
+            ..onCancel = () {
+              outerCanceled = true;
+            };
+          first = createController(innerType)
+            ..onCancel = () {
+              firstCanceled = true;
+            };
+          second = createController(innerType);
+          third = createController(innerType);
+          emittedValues = [];
+          errors = [];
+          isDone = false;
+          subscription = outer.stream
+              .switchLatest()
+              .listen(emittedValues.add, onError: errors.add, onDone: () {
+            isDone = true;
+          });
+        });
+
+        test('forwards events', () async {
+          outer.add(first.stream);
+          await Future(() {});
+          first
+            ..add(1)
+            ..add(2);
+          await Future(() {});
+
+          outer.add(second.stream);
+          await Future(() {});
+          second
+            ..add(3)
+            ..add(4);
+          await Future(() {});
+
+          expect(emittedValues, [1, 2, 3, 4]);
+        });
+
+        test('forwards errors from outer Stream', () async {
+          outer.addError('error');
+          await Future(() {});
+          expect(errors, ['error']);
+        });
+
+        test('forwards errors from inner Stream', () async {
+          outer.add(first.stream);
+          await Future(() {});
+          first.addError('error');
+          await Future(() {});
+          expect(errors, ['error']);
+        });
+
+        test('closes when final stream is done', () async {
+          outer.add(first.stream);
+          await Future(() {});
+
+          outer.add(second.stream);
+          await Future(() {});
+
+          await outer.close();
+          expect(isDone, false);
+
+          await second.close();
+          expect(isDone, true);
+        });
+
+        test(
+            'closes when outer stream closes if latest inner stream already '
+            'closed', () async {
+          outer.add(first.stream);
+          await Future(() {});
+          await first.close();
+          expect(isDone, false);
+
+          await outer.close();
+          expect(isDone, true);
+        });
+
+        test('cancels listeners on previous streams', () async {
+          outer.add(first.stream);
+          await Future(() {});
+
+          outer.add(second.stream);
+          await Future(() {});
+          expect(firstCanceled, true);
+        });
+
+        if (innerType != 'broadcast') {
+          test('waits for cancel before listening to subsequent stream',
+              () async {
+            var cancelWork = Completer<void>();
+            first.onCancel = () => cancelWork.future;
+            outer.add(first.stream);
+            await Future(() {});
+
+            var cancelDone = false;
+            second.onListen = expectAsync0(() {
+              expect(cancelDone, true);
+            });
+            outer.add(second.stream);
+            await Future(() {});
+            cancelWork.complete();
+            cancelDone = true;
+          });
+
+          test('all streams are listened to, even while cancelling', () async {
+            var cancelWork = Completer<void>();
+            first.onCancel = () => cancelWork.future;
+            outer.add(first.stream);
+            await Future(() {});
+
+            var cancelDone = false;
+            second.onListen = expectAsync0(() {
+              expect(cancelDone, true);
+            });
+            third.onListen = expectAsync0(() {
+              expect(cancelDone, true);
+            });
+            outer
+              ..add(second.stream)
+              ..add(third.stream);
+            await Future(() {});
+            cancelWork.complete();
+            cancelDone = true;
+          });
+        }
+
+        if (outerType != 'broadcast' && innerType != 'broadcast') {
+          test('pausing while cancelling an inner stream is respected',
+              () async {
+            var cancelWork = Completer<void>();
+            first.onCancel = () => cancelWork.future;
+            outer.add(first.stream);
+            await Future(() {});
+
+            var cancelDone = false;
+            second.onListen = expectAsync0(() {
+              expect(cancelDone, true);
+            });
+            outer.add(second.stream);
+            await Future(() {});
+            subscription.pause();
+            cancelWork.complete();
+            cancelDone = true;
+            await Future(() {});
+            expect(second.isPaused, true);
+            subscription.resume();
+          });
+        }
+
+        test('cancels listener on current and outer stream on cancel',
+            () async {
+          outer.add(first.stream);
+          await Future(() {});
+          await subscription.cancel();
+
+          await Future(() {});
+          expect(outerCanceled, true);
+          expect(firstCanceled, true);
+        });
+      });
+    }
+  }
+
+  group('switchMap', () {
+    test('uses map function', () async {
+      var outer = StreamController<List<int>>();
+
+      var values = <int>[];
+      outer.stream.switchMap(Stream.fromIterable).listen(values.add);
+
+      outer.add([1, 2, 3]);
+      await Future(() {});
+      outer.add([4, 5, 6]);
+      await Future(() {});
+      expect(values, [1, 2, 3, 4, 5, 6]);
+    });
+
+    test('can create a broadcast stream', () async {
+      var outer = StreamController<int>.broadcast();
+
+      var transformed =
+          outer.stream.switchMap((_) => const Stream<int>.empty());
+
+      expect(transformed.isBroadcast, true);
+    });
+
+    test('forwards errors from the convert callback', () async {
+      var errors = <String>[];
+      var source = Stream.fromIterable([1, 2, 3]);
+      source.switchMap<int>((i) {
+        // ignore: only_throw_errors
+        throw 'Error: $i';
+      }).listen((_) {}, onError: errors.add);
+      await Future<void>(() {});
+      expect(errors, ['Error: 1', 'Error: 2', 'Error: 3']);
+    });
+  });
+}
diff --git a/pkgs/stream_transform/test/take_until_test.dart b/pkgs/stream_transform/test/take_until_test.dart
new file mode 100644
index 0000000..982b3da
--- /dev/null
+++ b/pkgs/stream_transform/test/take_until_test.dart
@@ -0,0 +1,135 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  for (var streamType in streamTypes) {
+    group('takeUntil on Stream type [$streamType]', () {
+      late StreamController<int> values;
+      late List<int> emittedValues;
+      late bool valuesCanceled;
+      late bool isDone;
+      late List<String> errors;
+      late Stream<int> transformed;
+      late StreamSubscription<int> subscription;
+      late Completer<void> closeTrigger;
+
+      setUp(() {
+        valuesCanceled = false;
+        values = createController(streamType)
+          ..onCancel = () {
+            valuesCanceled = true;
+          };
+        emittedValues = [];
+        errors = [];
+        isDone = false;
+        closeTrigger = Completer();
+        transformed = values.stream.takeUntil(closeTrigger.future);
+        subscription = transformed
+            .listen(emittedValues.add, onError: errors.add, onDone: () {
+          isDone = true;
+        });
+      });
+
+      test('forwards cancellation', () async {
+        await subscription.cancel();
+        expect(valuesCanceled, true);
+      });
+
+      test('lets values through before trigger', () async {
+        values
+          ..add(1)
+          ..add(2);
+        await Future(() {});
+        expect(emittedValues, [1, 2]);
+      });
+
+      test('forwards errors', () async {
+        values.addError('error');
+        await Future(() {});
+        expect(errors, ['error']);
+      });
+
+      test('sends done if original strem ends', () async {
+        await values.close();
+        expect(isDone, true);
+      });
+
+      test('sends done when trigger fires', () async {
+        closeTrigger.complete();
+        await Future(() {});
+        expect(isDone, true);
+      });
+
+      test('forwards errors from the close trigger', () async {
+        closeTrigger.completeError('sad');
+        await Future(() {});
+        expect(errors, ['sad']);
+        expect(isDone, true);
+      });
+
+      test('ignores errors from the close trigger after stream closed',
+          () async {
+        await values.close();
+        closeTrigger.completeError('sad');
+        await Future(() {});
+        expect(errors, <Object>[]);
+      });
+
+      test('cancels value subscription when trigger fires', () async {
+        closeTrigger.complete();
+        await Future(() {});
+        expect(valuesCanceled, true);
+      });
+
+      if (streamType == 'broadcast') {
+        test('multiple listeners all get values', () async {
+          var otherValues = <Object>[];
+          transformed.listen(otherValues.add);
+          values
+            ..add(1)
+            ..add(2);
+          await Future(() {});
+          expect(emittedValues, [1, 2]);
+          expect(otherValues, [1, 2]);
+        });
+
+        test('multiple listeners get done when trigger fires', () async {
+          var otherDone = false;
+          transformed.listen(null, onDone: () => otherDone = true);
+          closeTrigger.complete();
+          await Future(() {});
+          expect(otherDone, true);
+          expect(isDone, true);
+        });
+
+        test('multiple listeners get done when values end', () async {
+          var otherDone = false;
+          transformed.listen(null, onDone: () => otherDone = true);
+          await values.close();
+          expect(otherDone, true);
+          expect(isDone, true);
+        });
+
+        test('can cancel and relisten before trigger fires', () async {
+          values.add(1);
+          await Future(() {});
+          await subscription.cancel();
+          values.add(2);
+          await Future(() {});
+          subscription = transformed.listen(emittedValues.add);
+          values.add(3);
+          await Future(() {});
+          expect(emittedValues, [1, 3]);
+        });
+      }
+    });
+  }
+}
diff --git a/pkgs/stream_transform/test/tap_test.dart b/pkgs/stream_transform/test/tap_test.dart
new file mode 100644
index 0000000..f2b4346
--- /dev/null
+++ b/pkgs/stream_transform/test/tap_test.dart
@@ -0,0 +1,116 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+void main() {
+  test('calls function for values', () async {
+    var valuesSeen = <int>[];
+    var stream = Stream.fromIterable([1, 2, 3]);
+    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.tap((_) {}).toList();
+    expect(values, [1, 2, 3]);
+  });
+
+  test('calls function for errors', () async {
+    dynamic error;
+    var source = StreamController<int>();
+    source.stream.tap((_) {}, onError: (e, st) {
+      error = e;
+    }).listen((_) {}, onError: (_) {});
+    source.addError('error');
+    await Future(() {});
+    expect(error, 'error');
+  });
+
+  test('forwards errors', () async {
+    dynamic error;
+    var source = StreamController<int>();
+    source.stream.tap((_) {}, onError: (e, st) {}).listen((_) {},
+        onError: (Object e) {
+      error = e;
+    });
+    source.addError('error');
+    await Future(() {});
+    expect(error, 'error');
+  });
+
+  test('calls function on done', () async {
+    var doneCalled = false;
+    var source = StreamController<int>();
+    source.stream.tap((_) {}, onDone: () {
+      doneCalled = true;
+    }).listen((_) {});
+    await source.close();
+    expect(doneCalled, true);
+  });
+
+  test('forwards only once with multiple listeners on a broadcast stream',
+      () async {
+    var dataCallCount = 0;
+    var source = StreamController<int>.broadcast();
+    source.stream.tap((_) {
+      dataCallCount++;
+    })
+      ..listen((_) {})
+      ..listen((_) {});
+    source.add(1);
+    await Future(() {});
+    expect(dataCallCount, 1);
+  });
+
+  test(
+      'forwards errors only once with multiple listeners on a broadcast stream',
+      () async {
+    var errorCallCount = 0;
+    var source = StreamController<int>.broadcast();
+    source.stream.tap((_) {}, onError: (_, __) {
+      errorCallCount++;
+    })
+      ..listen((_) {}, onError: (_, __) {})
+      ..listen((_) {}, onError: (_, __) {});
+    source.addError('error');
+    await Future(() {});
+    expect(errorCallCount, 1);
+  });
+
+  test('calls onDone only once with multiple listeners on a broadcast stream',
+      () async {
+    var doneCallCount = 0;
+    var source = StreamController<int>.broadcast();
+    source.stream.tap((_) {}, onDone: () {
+      doneCallCount++;
+    })
+      ..listen((_) {})
+      ..listen((_) {});
+    await source.close();
+    expect(doneCallCount, 1);
+  });
+
+  test('forwards values to multiple listeners', () async {
+    var source = StreamController<int>.broadcast();
+    var emittedValues1 = <int>[];
+    var emittedValues2 = <int>[];
+    source.stream.tap((_) {})
+      ..listen(emittedValues1.add)
+      ..listen(emittedValues2.add);
+    source.add(1);
+    await Future(() {});
+    expect(emittedValues1, [1]);
+    expect(emittedValues2, [1]);
+  });
+
+  test('allows null callback', () async {
+    var stream = Stream.fromIterable([1, 2, 3]);
+    await stream.tap(null).last;
+  });
+}
diff --git a/pkgs/stream_transform/test/throttle_test.dart b/pkgs/stream_transform/test/throttle_test.dart
new file mode 100644
index 0000000..07f607a
--- /dev/null
+++ b/pkgs/stream_transform/test/throttle_test.dart
@@ -0,0 +1,193 @@
+// 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:fake_async/fake_async.dart';
+import 'package:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  for (var streamType in streamTypes) {
+    group('Stream type [$streamType]', () {
+      late StreamController<int> values;
+      late List<int> emittedValues;
+      late bool valuesCanceled;
+      late bool isDone;
+      late Stream<int> transformed;
+      late StreamSubscription<int> subscription;
+
+      group('throttle - trailing: false', () {
+        setUp(() async {
+          valuesCanceled = false;
+          values = createController(streamType)
+            ..onCancel = () {
+              valuesCanceled = true;
+            };
+          emittedValues = [];
+          isDone = false;
+          transformed = values.stream.throttle(const Duration(milliseconds: 5));
+        });
+
+        void listen() {
+          subscription = transformed.listen(emittedValues.add, onDone: () {
+            isDone = true;
+          });
+        }
+
+        test('cancels values', () async {
+          listen();
+          await subscription.cancel();
+          expect(valuesCanceled, true);
+        });
+
+        test('swallows values that come faster than duration', () {
+          fakeAsync((async) {
+            listen();
+            values
+              ..add(1)
+              ..add(2)
+              ..close();
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [1]);
+          });
+        });
+
+        test('outputs multiple values spaced further than duration', () {
+          fakeAsync((async) {
+            listen();
+            values.add(1);
+            async.elapse(const Duration(milliseconds: 6));
+            values.add(2);
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [1, 2]);
+            async.elapse(const Duration(milliseconds: 6));
+          });
+        });
+
+        test('closes output immediately', () {
+          fakeAsync((async) {
+            listen();
+            values.add(1);
+            async.elapse(const Duration(milliseconds: 6));
+            values
+              ..add(2)
+              ..close();
+            async.flushMicrotasks();
+            expect(isDone, true);
+          });
+        });
+
+        if (streamType == 'broadcast') {
+          test('multiple listeners all get values', () {
+            fakeAsync((async) {
+              listen();
+              var otherValues = <int>[];
+              transformed.listen(otherValues.add);
+              values.add(1);
+              async.flushMicrotasks();
+              expect(emittedValues, [1]);
+              expect(otherValues, [1]);
+            });
+          });
+        }
+      });
+
+      group('throttle - trailing: true', () {
+        setUp(() async {
+          valuesCanceled = false;
+          values = createController(streamType)
+            ..onCancel = () {
+              valuesCanceled = true;
+            };
+          emittedValues = [];
+          isDone = false;
+          transformed = values.stream
+              .throttle(const Duration(milliseconds: 5), trailing: true);
+        });
+        void listen() {
+          subscription = transformed.listen(emittedValues.add, onDone: () {
+            isDone = true;
+          });
+        }
+
+        test('emits both first and last in a period', () {
+          fakeAsync((async) {
+            listen();
+            values
+              ..add(1)
+              ..add(2)
+              ..close();
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [1, 2]);
+          });
+        });
+
+        test('swallows values that are not the latest in a period', () {
+          fakeAsync((async) {
+            listen();
+            values
+              ..add(1)
+              ..add(2)
+              ..add(3)
+              ..close();
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [1, 3]);
+          });
+        });
+
+        test('waits to output the last value even if the stream closes',
+            () async {
+          fakeAsync((async) {
+            listen();
+            values
+              ..add(1)
+              ..add(2)
+              ..close();
+            async.flushMicrotasks();
+            expect(isDone, false);
+            expect(emittedValues, [1],
+                reason: 'Should not be emitted until after duration');
+            async.elapse(const Duration(milliseconds: 6));
+            expect(emittedValues, [1, 2]);
+            expect(isDone, true);
+            async.elapse(const Duration(milliseconds: 6));
+          });
+        });
+
+        test('closes immediately if there is no pending value', () {
+          fakeAsync((async) {
+            listen();
+            values
+              ..add(1)
+              ..close();
+            async.flushMicrotasks();
+            expect(isDone, true);
+          });
+        });
+
+        if (streamType == 'broadcast') {
+          test('multiple listeners all get values', () {
+            fakeAsync((async) {
+              listen();
+              var otherValues = <int>[];
+              transformed.listen(otherValues.add);
+              values
+                ..add(1)
+                ..add(2);
+              async.flushMicrotasks();
+              expect(emittedValues, [1]);
+              expect(otherValues, [1]);
+              async.elapse(const Duration(milliseconds: 6));
+              expect(emittedValues, [1, 2]);
+              expect(otherValues, [1, 2]);
+            });
+          });
+        }
+      });
+    });
+  }
+}
diff --git a/pkgs/stream_transform/test/utils.dart b/pkgs/stream_transform/test/utils.dart
new file mode 100644
index 0000000..42d9613
--- /dev/null
+++ b/pkgs/stream_transform/test/utils.dart
@@ -0,0 +1,19 @@
+// 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';
+
+StreamController<T> createController<T>(String streamType) {
+  switch (streamType) {
+    case 'single subscription':
+      return StreamController<T>();
+    case 'broadcast':
+      return StreamController<T>.broadcast();
+    default:
+      throw ArgumentError.value(
+          streamType, 'streamType', 'Must be one of $streamTypes');
+  }
+}
+
+const streamTypes = ['single subscription', 'broadcast'];
diff --git a/pkgs/stream_transform/test/where_not_null_test.dart b/pkgs/stream_transform/test/where_not_null_test.dart
new file mode 100644
index 0000000..c9af794
--- /dev/null
+++ b/pkgs/stream_transform/test/where_not_null_test.dart
@@ -0,0 +1,56 @@
+// Copyright (c) 2022, 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+void main() {
+  test('forwards only events that match the type', () async {
+    var values = Stream.fromIterable([null, 'a', null, 'b']);
+    var filtered = values.whereNotNull();
+    expect(await filtered.toList(), ['a', 'b']);
+  });
+
+  test('can result in empty stream', () async {
+    var values = Stream<Object?>.fromIterable([null, null]);
+    var filtered = values.whereNotNull();
+    expect(await filtered.isEmpty, true);
+  });
+
+  test('forwards values to multiple listeners', () async {
+    var values = StreamController<Object?>.broadcast();
+    var filtered = values.stream.whereNotNull();
+    var firstValues = <Object>[];
+    var secondValues = <Object>[];
+    filtered
+      ..listen(firstValues.add)
+      ..listen(secondValues.add);
+    values
+      ..add(null)
+      ..add('a')
+      ..add(null)
+      ..add('b');
+    await Future(() {});
+    expect(firstValues, ['a', 'b']);
+    expect(secondValues, ['a', 'b']);
+  });
+
+  test('closes streams with multiple listeners', () async {
+    var values = StreamController<Object?>.broadcast();
+    var filtered = values.stream.whereNotNull();
+    var firstDone = false;
+    var secondDone = false;
+    filtered
+      ..listen(null, onDone: () => firstDone = true)
+      ..listen(null, onDone: () => secondDone = true);
+    values
+      ..add(null)
+      ..add('a');
+    await values.close();
+    expect(firstDone, true);
+    expect(secondDone, true);
+  });
+}
diff --git a/pkgs/stream_transform/test/where_type_test.dart b/pkgs/stream_transform/test/where_type_test.dart
new file mode 100644
index 0000000..4cbea37
--- /dev/null
+++ b/pkgs/stream_transform/test/where_type_test.dart
@@ -0,0 +1,56 @@
+// 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:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+void main() {
+  test('forwards only events that match the type', () async {
+    var values = Stream.fromIterable([1, 'a', 2, 'b']);
+    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.whereType<String>();
+    expect(await filtered.isEmpty, true);
+  });
+
+  test('forwards values to multiple listeners', () async {
+    var values = StreamController<Object>.broadcast();
+    var filtered = values.stream.whereType<String>();
+    var firstValues = <Object>[];
+    var secondValues = <Object>[];
+    filtered
+      ..listen(firstValues.add)
+      ..listen(secondValues.add);
+    values
+      ..add(1)
+      ..add('a')
+      ..add(2)
+      ..add('b');
+    await Future(() {});
+    expect(firstValues, ['a', 'b']);
+    expect(secondValues, ['a', 'b']);
+  });
+
+  test('closes streams with multiple listeners', () async {
+    var values = StreamController<Object>.broadcast();
+    var filtered = values.stream.whereType<String>();
+    var firstDone = false;
+    var secondDone = false;
+    filtered
+      ..listen(null, onDone: () => firstDone = true)
+      ..listen(null, onDone: () => secondDone = true);
+    values
+      ..add(1)
+      ..add('a');
+    await values.close();
+    expect(firstDone, true);
+    expect(secondDone, true);
+  });
+}