Update PR based on feedback

  - Deprecate ignoreElements
  - Update docs for TimerInterval and TimeStamped
  - Update docs for flatMap

Author: brianegan

lib/src/transformers/flat_map.dart

@@ -113,8 +113,7 @@ extension FlatMapExtension<T> on Stream<T> {
   /// ### Example
   ///
   ///     RangeStream(4, 1)
-  ///       .flatMap((i) =>
-  ///         TimerStream(i, Duration(minutes: i))
+  ///       .flatMap((i) => TimerStream(i, Duration(minutes: i))
   ///       .listen(print); // prints 1, 2, 3, 4
   Stream<S> flatMap<S>(Stream<S> mapper(T value)) =>
       transform(FlatMapStreamTransformer<T, S>(mapper));
@@ -129,8 +128,7 @@ extension FlatMapExtension<T> on Stream<T> {
   /// ### Example
   ///
   ///     RangeStream(1, 4)
-  ///       .flatMapIterable((i) =>
-  ///         Stream.fromIterable([[]i])
+  ///       .flatMapIterable((i) => Stream.fromIterable([[i]])
   ///       .listen(print); // prints 1, 2, 3, 4
   Stream<S> flatMapIterable<S>(Stream<Iterable<S>> mapper(T value)) =>
       transform(FlatMapStreamTransformer<T, Iterable<S>>(mapper))

lib/src/transformers/ignore_elements.dart

@@ -10,6 +10,7 @@ import 'dart:async';
 ///       ErrorStream(Exception())
 ///     ])
 ///     .listen(print, onError: print); // prints Exception
+@Deprecated('Use the drain method from the Stream class instead')
 class IgnoreElementsStreamTransformer<T> extends StreamTransformerBase<T, T> {
   final StreamTransformer<T, T> _transformer;
 
@@ -55,5 +56,6 @@ extension IgnoreElementsExtension<T> on Stream<T> {
   ///      Stream.error(Exception())
   ///    ])
   ///    .listen(print, onError: print); // prints Exception
+  @Deprecated('Use the drain method from the Stream class instead')
   Stream<T> ignoreElements() => transform(IgnoreElementsStreamTransformer<T>());
 }

lib/src/transformers/time_interval.dart

@@ -98,9 +98,8 @@ class TimeInterval<T> {
   }
 }
 
-/// Extends the Stream class with the ability to wrap each item emitted by the
-/// source Stream in a [Timestamped] object that includes the emitted item
-/// and the time when the item was emitted.
+/// Extends the Stream class with the ability to record the time interval
+/// between consecutive values in an stream
 extension TimeIntervalExtension<T> on Stream<T> {
   /// Records the time interval between consecutive values in a Stream sequence.
   ///

lib/src/transformers/timestamp.dart

@@ -79,7 +79,9 @@ class Timestamped<T> {
   }
 }
 
-/// Extends the Stream class with the ability to
+/// Extends the Stream class with the ability to wrap each item emitted by the
+/// source Stream in a [Timestamped] object that includes the emitted item and
+/// the time when the item was emitted.
 extension TimeStampExtension<T> on Stream<T> {
   /// Wraps each item emitted by the source Stream in a [Timestamped] object
   /// that includes the emitted item and the time when the item was emitted.

test/streams/sequence_equals_test.dart

@@ -96,14 +96,14 @@ void main() {
   });
 
   test('Rx.sequenceEqual.asBroadcastStream', () async {
-    final stream = Rx.sequenceEqual(Stream.fromIterable(const [1, 2, 3, 4, 5]),
+    final future = Rx.sequenceEqual(Stream.fromIterable(const [1, 2, 3, 4, 5]),
             Stream.fromIterable(const [1, 2, 3, 4, 5]))
         .asBroadcastStream()
-        .ignoreElements();
+        .drain<void>();
 
     // listen twice on same stream
-    await expectLater(stream, emitsDone);
-    await expectLater(stream, emitsDone);
+    await expectLater(future, completes);
+    await expectLater(future, completes);
   });
 
   test('Rx.sequenceEqual.error.shouldThrowA', () {

test/transformers/backpressure/debounce_test.dart

@@ -50,14 +50,14 @@ void main() {
   });
 
   test('Rx.debounce.asBroadcastStream', () async {
-    final stream = _getStream()
+    final future = _getStream()
         .asBroadcastStream()
         .debounce((_) => Stream<void>.fromFuture(
             Future<void>.delayed(const Duration(milliseconds: 200))))
-        .ignoreElements();
+        .drain<void>();
 
-    await expectLater(stream, emitsDone);
-    await expectLater(stream, emitsDone);
+    await expectLater(future, completes);
+    await expectLater(future, completes);
   });
 
   test('Rx.debounce.error.shouldThrowA', () async {

test/transformers/backpressure/debounce_time_test.dart

@@ -36,13 +36,13 @@ void main() {
   });
 
   test('Rx.debounceTime.asBroadcastStream', () async {
-    final stream = _getStream()
+    final future = _getStream()
         .asBroadcastStream()
         .debounceTime(const Duration(milliseconds: 200))
-        .ignoreElements();
+        .drain<void>();
 
-    await expectLater(stream, emitsDone);
-    await expectLater(stream, emitsDone);
+    await expectLater(future, completes);
+    await expectLater(future, completes);
   });
 
   test('Rx.debounceTime.error.shouldThrowA', () async {

test/transformers/backpressure/sample_time_test.dart

@@ -20,10 +20,12 @@ void main() {
             .asBroadcastStream());
 
     await expectLater(
-        _getStream().transform(transformer).ignoreElements(), emitsDone);
+      _getStream().transform(transformer).drain<void>(),
+      completes,
+    );
     await expectLater(
-      _getStream().transform(transformer).ignoreElements(),
-      emitsDone,
+      _getStream().transform(transformer).drain<void>(),
+      completes,
     );
   });
 

test/transformers/backpressure/throttle_test.dart

@@ -60,15 +60,15 @@ void main() {
   });
 
   test('Rx.throttle.asBroadcastStream', () async {
-    final stream = _stream()
+    final future = _stream()
         .asBroadcastStream()
         .throttle(
             (_) => Stream<void>.periodic(const Duration(milliseconds: 250)))
-        .ignoreElements();
+        .drain<void>();
 
     // listen twice on same stream
-    await expectLater(stream, emitsDone);
-    await expectLater(stream, emitsDone);
+    await expectLater(future, completes);
+    await expectLater(future, completes);
   });
 
   test('Rx.throttle.error.shouldThrowA', () async {

test/transformers/backpressure/throttle_time_test.dart

@@ -33,14 +33,14 @@ void main() {
   });
 
   test('Rx.throttleTime.asBroadcastStream', () async {
-    final stream = _stream()
+    final future = _stream()
         .asBroadcastStream()
         .throttleTime(const Duration(milliseconds: 250))
-        .ignoreElements();
+        .drain<void>();
 
     // listen twice on same stream
-    await expectLater(stream, emitsDone);
-    await expectLater(stream, emitsDone);
+    await expectLater(future, completes);
+    await expectLater(future, completes);
   });
 
   test('Rx.throttleTime.error.shouldThrowA', () async {