Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(183)

Side by Side Diff: pkg/barback/lib/src/stream_replayer.dart

Issue 25094002: Adapt streams for additional stackTrace argument. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Remove types in closures. Created 7 years, 2 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch | Annotate | Revision Log
« no previous file with comments | « pkg/barback/lib/src/package_graph.dart ('k') | pkg/barback/lib/src/utils.dart » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 // Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file 1 // Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
2 // for details. All rights reserved. Use of this source code is governed by a 2 // for details. All rights reserved. Use of this source code is governed by a
3 // BSD-style license that can be found in the LICENSE file. 3 // BSD-style license that can be found in the LICENSE file.
4 4
5 library barback.stream_replayer; 5 library barback.stream_replayer;
6 6
7 import 'dart:async'; 7 import 'dart:async';
8 import 'dart:collection'; 8 import 'dart:collection';
9 9
10 import 'utils.dart'; 10 import 'utils.dart';
11 11
12 /// Records the values and errors that are sent through a stream and allows them 12 /// Records the values and errors that are sent through a stream and allows them
13 /// to be replayed arbitrarily many times. 13 /// to be replayed arbitrarily many times.
14 class StreamReplayer<T> { 14 class StreamReplayer<T> {
15 /// The wrapped stream. 15 /// The wrapped stream.
16 final Stream<T> _stream; 16 final Stream<T> _stream;
17 17
18 /// Whether or not [_stream] has been closed. 18 /// Whether or not [_stream] has been closed.
19 bool _isClosed = false; 19 bool _isClosed = false;
20 20
21 /// The buffer of events or errors that have already been emitted by 21 /// The buffer of events or errors that have already been emitted by
22 /// [_stream]. 22 /// [_stream].
23 /// 23 ///
24 /// Each element is a [Either] that's either a value or an error sent through 24 /// Each element is a [Either] that's either a value or an error sent through
25 /// the stream. 25 /// the stream.
26 final _buffer = new Queue<Either<T, dynamic>>(); 26 final _buffer = new Queue<Either<T, Pair<dynamic, StackTrace>>>();
27 27
28 /// The controllers that are listening for future events from [_stream]. 28 /// The controllers that are listening for future events from [_stream].
29 final _controllers = new Set<StreamController<T>>(); 29 final _controllers = new Set<StreamController<T>>();
30 30
31 StreamReplayer(this._stream) { 31 StreamReplayer(this._stream) {
32 _stream.listen((data) { 32 _stream.listen((data) {
33 _buffer.add(new Either<T, dynamic>.withFirst(data)); 33 _buffer.add(new Either<T, dynamic>.withFirst(data));
34 for (var controller in _controllers) { 34 for (var controller in _controllers) {
35 controller.add(data); 35 controller.add(data);
36 } 36 }
37 }, onError: (error) { 37 }, onError: (error, [stackTrace]) {
38 _buffer.add(new Either<T, dynamic>.withSecond(error)); 38 _buffer.add(new Either<T, Pair<dynamic, StackTrace>>.withSecond(
39 new Pair<dynamic, StackTrace>(error, stackTrace)));
39 for (var controller in _controllers) { 40 for (var controller in _controllers) {
40 controller.addError(error); 41 controller.addError(error, stackTrace);
41 } 42 }
42 }, onDone: () { 43 }, onDone: () {
43 _isClosed = true; 44 _isClosed = true;
44 for (var controller in _controllers) { 45 for (var controller in _controllers) {
45 controller.close(); 46 controller.close();
46 } 47 }
47 _controllers.clear(); 48 _controllers.clear();
48 }); 49 });
49 } 50 }
50 51
51 /// Returns a stream that replays the values and errors of the input stream. 52 /// Returns a stream that replays the values and errors of the input stream.
52 /// 53 ///
53 /// This stream is a buffered stream regardless of whether the input stream 54 /// This stream is a buffered stream regardless of whether the input stream
54 /// was broadcast or buffered. 55 /// was broadcast or buffered.
55 Stream<T> getReplay() { 56 Stream<T> getReplay() {
56 var controller = new StreamController<T>(); 57 var controller = new StreamController<T>();
57 for (var eventOrError in _buffer) { 58 for (var eventOrError in _buffer) {
58 eventOrError.match(controller.add, controller.addError); 59 eventOrError.match(controller.add, (pair) {
60 controller.addError(pair.first, pair.second);
61 });
59 } 62 }
60 if (_isClosed) { 63 if (_isClosed) {
61 controller.close(); 64 controller.close();
62 } else { 65 } else {
63 _controllers.add(controller); 66 _controllers.add(controller);
64 } 67 }
65 return controller.stream; 68 return controller.stream;
66 } 69 }
67 } 70 }
OLDNEW
« no previous file with comments | « pkg/barback/lib/src/package_graph.dart ('k') | pkg/barback/lib/src/utils.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698