Add a shared ignoreArgument callback (#190)

Add a `common_callbacks.dart` library with a single top level function
`ignoreArgument` which takes a single `dynamic` argument and does
nothing. Replace all `(_) => null` function literals, as well as a few
library-private functions with unnecessarily specific signatures, with
the shared definition.

As suggested in #188

Author: natebosch

lib/src/aggregate_sample.dart

@@ -4,6 +4,8 @@
 
 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.
@@ -136,7 +138,7 @@ extension AggregateSample<T> on Stream<T> {
           triggerSub!.pause();
         }
         if (cancels.isEmpty) return null;
-        return cancels.wait.then((_) => null);
+        return cancels.wait.then(ignoreArgument);
       };
     };
     return controller.stream;

lib/src/async_expand.dart

@@ -4,6 +4,7 @@
 
 import 'dart:async';
 
+import 'common_callbacks.dart';
 import 'switch.dart';
 
 /// Alternatives to [asyncExpand].
@@ -78,7 +79,9 @@ extension AsyncExpand<T> on Stream<T> {
       }
       controller.onCancel = () {
         if (subscriptions.isEmpty) return null;
-        return [for (var s in subscriptions) s.cancel()].wait.then((_) => null);
+        return [for (var s in subscriptions) s.cancel()]
+            .wait
+            .then(ignoreArgument);
       };
     };
     return controller.stream;

lib/src/async_map.dart

@@ -5,6 +5,7 @@
 import 'dart:async';
 
 import 'aggregate_sample.dart';
+import 'common_callbacks.dart';
 import 'from_handlers.dart';
 import 'rate_limit.dart';
 
@@ -72,7 +73,7 @@ extension AsyncMap<T> on Stream<T> {
             trigger: workFinished.stream,
             aggregate: _dropPrevious,
             longPoll: true,
-            onEmpty: _ignore<T>)
+            onEmpty: ignoreArgument)
         ._asyncMapThen(convert, workFinished.add);
   }
 
@@ -133,4 +134,3 @@ extension AsyncMap<T> on Stream<T> {
 }
 
 T _dropPrevious<T>(T event, _) => event;
-void _ignore<T>(Sink<T> sink) {}

lib/src/combine_latest.dart

@@ -4,6 +4,8 @@
 
 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> {
@@ -131,7 +133,7 @@ extension CombineLatest<T> on Stream<T> {
         ];
         sourceSubscription = null;
         otherSubscription = null;
-        return cancels.wait.then((_) => null);
+        return cancels.wait.then(ignoreArgument);
       };
     };
     return controller.stream;
@@ -228,7 +230,9 @@ extension CombineLatest<T> on Stream<T> {
       }
       controller.onCancel = () {
         if (subscriptions.isEmpty) return null;
-        return [for (var s in subscriptions) s.cancel()].wait.then((_) => null);
+        return [for (var s in subscriptions) s.cancel()]
+            .wait
+            .then(ignoreArgument);
       };
     };
     return controller.stream;

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(_) {}

lib/src/merge.dart

@@ -4,6 +4,8 @@
 
 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
@@ -90,7 +92,9 @@ extension Merge<T> on Stream<T> {
       }
       controller.onCancel = () {
         if (subscriptions.isEmpty) return null;
-        return [for (var s in subscriptions) s.cancel()].wait.then((_) => null);
+        return [for (var s in subscriptions) s.cancel()]
+            .wait
+            .then(ignoreArgument);
       };
     };
     return controller.stream;

lib/src/rate_limit.dart

@@ -5,6 +5,7 @@
 import 'dart:async';
 
 import 'aggregate_sample.dart';
+import 'common_callbacks.dart';
 import 'from_handlers.dart';
 
 /// Utilities to rate limit events.
@@ -305,7 +306,7 @@ extension RateLimit<T> on Stream<T> {
           trigger: trigger,
           aggregate: _dropPrevious,
           longPoll: longPoll,
-          onEmpty: _ignore);
+          onEmpty: ignoreArgument);
 
   /// Aggregates values until this source stream does not emit for [duration],
   /// then emits the aggregated values.
@@ -353,4 +354,3 @@ extension RateLimit<T> on Stream<T> {
 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([]);
-void _ignore<T>(Sink<T> sink) {}

lib/src/switch.dart

@@ -5,6 +5,7 @@
 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.
@@ -126,12 +127,9 @@ extension SwitchLatest<T> on Stream<Stream<T>> {
           if (sub != null) sub.cancel(),
         ];
         if (cancels.isEmpty) return null;
-        return cancels.wait.then(_ignore);
+        return cancels.wait.then(ignoreArgument);
       };
     };
     return controller.stream;
   }
 }
-
-/// Helper function to ignore future callback
-void _ignore(_, [__]) {}