Add `debounce` utilities and prepare for publish (dart-lang/stream_transform#3)
- Add `debounce` and `debounceBuffer` to drop or collect values that
occur quickly
- Add tests
- Drop `-dev` from pubspec to prepare for publishing very early version
- Add LICENSE file
- Exclude dartanalyzer and dartfmt test from 1.22.1 SDK
diff --git a/pkgs/stream_transform/.travis.yml b/pkgs/stream_transform/.travis.yml
index ed6c5bb..4c02d8e 100644
--- a/pkgs/stream_transform/.travis.yml
+++ b/pkgs/stream_transform/.travis.yml
@@ -11,3 +11,9 @@
- test
- dartfmt
- dartanalyzer
+matrix:
+ exclude:
+ - dart: 1.22.1
+ dart_task: dartfmt
+ - dart: 1.22.1
+ dart_task: dartanalyzer
diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md
index af394b6..cf303c7 100644
--- a/pkgs/stream_transform/CHANGELOG.md
+++ b/pkgs/stream_transform/CHANGELOG.md
@@ -1,4 +1,6 @@
## 0.0.1
-- Add `buffer` utility. Collects events in a `List` until a `trigger` stream
- fires.
+- 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..389ce98
--- /dev/null
+++ b/pkgs/stream_transform/LICENSE
@@ -0,0 +1,26 @@
+Copyright 2017, the Dart project authors. All rights reserved.
+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 Inc. 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/lib/src/debounce.dart b/pkgs/stream_transform/lib/src/debounce.dart
new file mode 100644
index 0000000..980c371
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/debounce.dart
@@ -0,0 +1,57 @@
+// 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';
+
+/// Creates a StreamTransformer which only emits when the source stream does not
+/// emit for [duration].
+///
+/// Source values will always be delayed by at least [duration], and values
+/// which come within this time will replace the old values, only the most
+/// recent value will be emitted.
+StreamTransformer<T, T> debounce<T>(Duration duration) =>
+ _debounceAggregate(duration, _dropPrevious);
+
+/// Creates a StreamTransformer which collects values until the source stream
+/// does not emit for [duration] then emits the collected values.
+///
+/// This differs from [debounce] in that values are aggregated instead of
+/// skipped.
+StreamTransformer<T, List<T>> debounceBuffer<T>(Duration duration) =>
+ _debounceAggregate(duration, _collectToList);
+
+List<T> _collectToList<T>(T element, List<T> soFar) {
+ soFar ??= <T>[];
+ soFar.add(element);
+ return soFar;
+}
+
+T _dropPrevious<T>(T element, _) => element;
+
+/// Creates a StreamTransformer which aggregates values until the source stream
+/// does not emit for [duration], then emits the aggregated values.
+StreamTransformer<T, R> _debounceAggregate<T, R>(
+ Duration duration, R collect(T element, R soFar)) {
+ Timer timer;
+ R soFar;
+ bool shouldClose = false;
+ return new StreamTransformer.fromHandlers(
+ handleData: (T value, EventSink<R> sink) {
+ timer?.cancel();
+ timer = new Timer(duration, () {
+ sink.add(soFar);
+ if (shouldClose) {
+ sink.close();
+ }
+ soFar = null;
+ timer = null;
+ });
+ soFar = collect(value, soFar);
+ }, handleDone: (EventSink<T> sink) {
+ if (soFar != null) {
+ shouldClose = true;
+ } else {
+ sink.close();
+ }
+ });
+}
diff --git a/pkgs/stream_transform/lib/stream_transform.dart b/pkgs/stream_transform/lib/stream_transform.dart
index 02069f5..49a714e 100644
--- a/pkgs/stream_transform/lib/stream_transform.dart
+++ b/pkgs/stream_transform/lib/stream_transform.dart
@@ -3,3 +3,4 @@
// BSD-style license that can be found in the LICENSE file.
export 'src/buffer.dart';
+export 'src/debounce.dart';
diff --git a/pkgs/stream_transform/pubspec.yaml b/pkgs/stream_transform/pubspec.yaml
index b260f12..1cf4034 100644
--- a/pkgs/stream_transform/pubspec.yaml
+++ b/pkgs/stream_transform/pubspec.yaml
@@ -2,7 +2,7 @@
description: A collection of utilities to transform and manipulate streams.
author: Dart Team <misc@dartlang.org>
homepage: https://www.github.com/dart-lang/stream_transform
-version: 0.0.1-dev
+version: 0.0.1
environment:
sdk: ">=1.22.0 <2.0.0"
diff --git a/pkgs/stream_transform/test/debounce_test.dart b/pkgs/stream_transform/test/debounce_test.dart
new file mode 100644
index 0000000..a556b41
--- /dev/null
+++ b/pkgs/stream_transform/test/debounce_test.dart
@@ -0,0 +1,110 @@
+import 'dart:async';
+import 'package:test/test.dart';
+import 'package:stream_transform/stream_transform.dart';
+
+void main() {
+ var streamTypes = {
+ 'single subscription': () => new StreamController(),
+ 'broadcast': () => new StreamController.broadcast()
+ };
+ for (var streamType in streamTypes.keys) {
+ group('Stream type [$streamType]', () {
+ StreamController values;
+ List emittedValues;
+ bool valuesCanceled;
+ bool isDone;
+ List errors;
+ StreamSubscription subscription;
+
+ void setUpStreams(StreamTransformer transformer) {
+ valuesCanceled = false;
+ values = streamTypes[streamType]()
+ ..onCancel = () {
+ valuesCanceled = true;
+ };
+ emittedValues = [];
+ errors = [];
+ isDone = false;
+ subscription = values.stream
+ .transform(transformer)
+ .listen(emittedValues.add, onError: errors.add, onDone: () {
+ isDone = true;
+ });
+ }
+
+ group('debounce', () {
+ setUp(() async {
+ setUpStreams(debounce(const Duration(milliseconds: 5)));
+ });
+
+ test('cancels values', () async {
+ await subscription.cancel();
+ expect(valuesCanceled, true);
+ });
+
+ test('swallows values that come faster than duration', () async {
+ values.add(1);
+ values.add(2);
+ await values.close();
+ await new Future.delayed(const Duration(milliseconds: 10));
+ expect(emittedValues, [2]);
+ });
+
+ test('outputs multiple values spaced further than duration', () async {
+ values.add(1);
+ await new Future.delayed(const Duration(milliseconds: 10));
+ values.add(2);
+ await new Future.delayed(const Duration(milliseconds: 10));
+ expect(emittedValues, [1, 2]);
+ });
+
+ test('waits for pending value to close', () async {
+ values.add(1);
+ await new Future.delayed(const Duration(milliseconds: 10));
+ await values.close();
+ await new Future(() {});
+ expect(isDone, true);
+ });
+
+ test('closes output if there are no pending values', () async {
+ values.add(1);
+ await new Future.delayed(const Duration(milliseconds: 10));
+ values.add(2);
+ await new Future(() {});
+ await values.close();
+ expect(isDone, false);
+ await new Future.delayed(const Duration(milliseconds: 10));
+ expect(isDone, true);
+ });
+ });
+
+ group('debounceBuffer', () {
+ setUp(() async {
+ setUpStreams(debounceBuffer(const Duration(milliseconds: 5)));
+ });
+
+ test('Emits all values as a list', () async {
+ values.add(1);
+ values.add(2);
+ await values.close();
+ await new Future.delayed(const Duration(milliseconds: 10));
+ expect(emittedValues, [
+ [1, 2]
+ ]);
+ });
+
+ test('separate lists for multiple values spaced further than duration',
+ () async {
+ values.add(1);
+ await new Future.delayed(const Duration(milliseconds: 10));
+ values.add(2);
+ await new Future.delayed(const Duration(milliseconds: 10));
+ expect(emittedValues, [
+ [1],
+ [2]
+ ]);
+ });
+ });
+ });
+ }
+}
