Call `tap` data callback once per event (dart-lang/stream_transform#26)
This is slightly different from the other bugs. In this case the output
stream correctly forwarded values to all listeners, but the callback was
also called once per listener.
- Add `handleError` to `fromHandlers` and tests.
- Add tests for expected single-call behavior. These fail on the old
implementation.
- Add test for the values getting to multiple listeners. This passes on
both the old and new implementation.
- Switch to `fromHandlers`.
diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md
index 00e2600..7f1aedf 100644
--- a/pkgs/stream_transform/CHANGELOG.md
+++ b/pkgs/stream_transform/CHANGELOG.md
@@ -2,6 +2,8 @@
- Bug Fix: Some transformers did not correctly add data to all listeners on
broadcast streams. Fixed for `throttle`, `debounce`, and `audit`.
+- Bug Fix: Only call the `tap` data callback once per event rather than once per
+ listener.
## 0.0.5
diff --git a/pkgs/stream_transform/lib/src/from_handlers.dart b/pkgs/stream_transform/lib/src/from_handlers.dart
index 8157ea1..315de96 100644
--- a/pkgs/stream_transform/lib/src/from_handlers.dart
+++ b/pkgs/stream_transform/lib/src/from_handlers.dart
@@ -2,26 +2,43 @@
typedef void HandleData<S, T>(S value, EventSink<T> sink);
typedef void HandleDone<T>(EventSink<T> sink);
+typedef void HandleError<T>(
+ Object error, StackTrace stackTrace, EventSink<T> sink);
/// Like [new StreamTransformer.fromHandlers] but the handlers are called once
/// per event rather than once per listener for broadcast streams.
StreamTransformer<S, T> fromHandlers<S, T>(
- {HandleData<S, T> handleData, HandleDone<T> handleDone}) =>
- new _StreamTransformer(handleData: handleData, handleDone: handleDone);
+ {HandleData<S, T> handleData,
+ HandleError<T> handleError,
+ HandleDone<T> handleDone}) =>
+ new _StreamTransformer(
+ handleData: handleData,
+ handleError: handleError,
+ handleDone: handleDone);
class _StreamTransformer<S, T> implements StreamTransformer<S, T> {
final HandleData<S, T> _handleData;
final HandleDone<T> _handleDone;
+ final HandleError<T> _handleError;
- _StreamTransformer({HandleData<S, T> handleData, HandleDone<T> handleDone})
+ _StreamTransformer(
+ {HandleData<S, T> handleData,
+ HandleError<T> handleError,
+ HandleDone<T> handleDone})
: _handleData = handleData ?? _defaultHandleData,
+ _handleError = handleError ?? _defaultHandleError,
_handleDone = handleDone ?? _defaultHandleDone;
- static _defaultHandleData<S, T>(S value, EventSink<T> sink) {
+ static void _defaultHandleData<S, T>(S value, EventSink<T> sink) {
sink.add(value as T);
}
- static _defaultHandleDone<T>(EventSink<T> sink) {
+ static void _defaultHandleError<T>(
+ Object error, StackTrace stackTrace, EventSink<T> sink) {
+ sink.addError(error, stackTrace);
+ }
+
+ static void _defaultHandleDone<T>(EventSink<T> sink) {
sink.close();
}
@@ -39,7 +56,9 @@
return;
}
subscription = values.listen((value) => _handleData(value, controller),
- onError: controller.addError, onDone: () {
+ onError: (error, stackTrace) {
+ _handleError(error, stackTrace, controller);
+ }, onDone: () {
_handleDone(controller);
});
};
diff --git a/pkgs/stream_transform/lib/src/tap.dart b/pkgs/stream_transform/lib/src/tap.dart
index 2ec8cf8..8046640 100644
--- a/pkgs/stream_transform/lib/src/tap.dart
+++ b/pkgs/stream_transform/lib/src/tap.dart
@@ -3,6 +3,8 @@
// BSD-style license that can be found in the LICENSE file.
import 'dart:async';
+import 'from_handlers.dart';
+
/// Taps into a Stream to allow additional handling on a single-subscriber
/// stream without first wrapping as a broadcast stream.
///
@@ -13,7 +15,7 @@
/// listener, and may be canceled only by the listener.
StreamTransformer<T, T> tap<T>(void fn(T value),
{void onError(error, stackTrace), void onDone()}) =>
- new StreamTransformer.fromHandlers(handleData: (value, sink) {
+ fromHandlers(handleData: (value, sink) {
try {
fn(value);
} catch (_) {/*Ignore*/}
diff --git a/pkgs/stream_transform/test/from_handlers_test.dart b/pkgs/stream_transform/test/from_handlers_test.dart
index 28436bd..268b55e 100644
--- a/pkgs/stream_transform/test/from_handlers_test.dart
+++ b/pkgs/stream_transform/test/from_handlers_test.dart
@@ -136,18 +136,23 @@
group('broadcast stream with multiple listeners', () {
int dataCallCount;
int doneCallCount;
+ int errorCallCount;
setUp(() async {
dataCallCount = 0;
doneCallCount = 0;
+ errorCallCount = 0;
setUpForController(
new StreamController.broadcast(),
fromHandlers(handleData: (value, sink) {
dataCallCount++;
+ }, handleError: (error, stackTrace, sink) {
+ errorCallCount++;
+ sink.addError(error, stackTrace);
}, handleDone: (sink) {
doneCallCount++;
}));
- transformed.listen((_) {});
+ transformed.listen((_) {}, onError: (_, __) {});
});
test('handles data once', () async {
@@ -160,6 +165,12 @@
await values.close();
expect(doneCallCount, 1);
});
+
+ test('handles errors once', () async {
+ values.addError('error');
+ await new Future(() {});
+ expect(errorCallCount, 1);
+ });
});
});
}
diff --git a/pkgs/stream_transform/test/tap_test.dart b/pkgs/stream_transform/test/tap_test.dart
index 8aecee6..eb64eee 100644
--- a/pkgs/stream_transform/test/tap_test.dart
+++ b/pkgs/stream_transform/test/tap_test.dart
@@ -53,4 +53,59 @@
await source.close();
expect(doneCalled, true);
});
+
+ test('forwards only once with multiple listeners on a broadcast stream',
+ () async {
+ var dataCallCount = 0;
+ var source = new StreamController.broadcast();
+ source.stream.transform(tap((_) {
+ dataCallCount++;
+ }))
+ ..listen((_) {})
+ ..listen((_) {});
+ source.add(1);
+ await new Future(() {});
+ expect(dataCallCount, 1);
+ });
+
+ test(
+ 'forwards errors only once with multiple listeners on a broadcast stream',
+ () async {
+ var errorCallCount = 0;
+ var source = new StreamController.broadcast();
+ source.stream.transform(tap((_) {}, onError: (_, __) {
+ errorCallCount++;
+ }))
+ ..listen((_) {}, onError: (_, __) {})
+ ..listen((_) {}, onError: (_, __) {});
+ source.addError('error');
+ await new Future(() {});
+ expect(errorCallCount, 1);
+ });
+
+ test('calls onDone only once with multiple listeners on a broadcast stream',
+ () async {
+ var doneCallCount = 0;
+ var source = new StreamController.broadcast();
+ source.stream.transform(tap((_) {}, onDone: () {
+ doneCallCount++;
+ }))
+ ..listen((_) {})
+ ..listen((_) {});
+ await source.close();
+ expect(doneCallCount, 1);
+ });
+
+ test('forwards values to multiple listeners', () async {
+ var source = new StreamController.broadcast();
+ var emittedValues1 = [];
+ var emittedValues2 = [];
+ source.stream.transform(tap((_) {}))
+ ..listen(emittedValues1.add)
+ ..listen(emittedValues2.add);
+ source.add(1);
+ await new Future(() {});
+ expect(emittedValues1, [1]);
+ expect(emittedValues2, [1]);
+ });
}
