Close sink on trailing throttle without pending (dart-lang/stream_transform#125)

pkgs/stream_transform/lib/src/rate_limit.dart

diff --git a/pkgs/stream_transform/lib/src/rate_limit.dart b/pkgs/stream_transform/lib/src/rate_limit.dart
index ebb55a2..b29f4f5 100644
--- a/pkgs/stream_transform/lib/src/rate_limit.dart
+++ b/pkgs/stream_transform/lib/src/rate_limit.dart
@@ -165,6 +165,7 @@
     }, onDone: (sink) {
       isDone = true;
       if (hasPending) return; // Will be closed by timer.
+      sink.close();
       timer?.cancel();
       timer = null;
     });

pkgs/stream_transform/test/throttle_test.dart

diff --git a/pkgs/stream_transform/test/throttle_test.dart b/pkgs/stream_transform/test/throttle_test.dart
index 27d9b11..bbd0cfb 100644
--- a/pkgs/stream_transform/test/throttle_test.dart
+++ b/pkgs/stream_transform/test/throttle_test.dart
@@ -117,6 +117,13 @@
           expect(isDone, true);
         });
 
+        test('closes immediately if there is no pending value', () async {
+          values.add(1);
+          await values.close();
+          await Future(() {});
+          expect(isDone, true);
+        });
+
         if (streamType == 'broadcast') {
           test('multiple listeners all get values', () async {
             var otherValues = <int>[];