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

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

Issue 23469003: Add an AssetStream class to Barback. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Add tests and fix a few bugs. Created 7 years, 3 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
OLDNEW
(Empty)
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
3 // BSD-style license that can be found in the LICENSE file.
4
5 library barback.stream_replayer;
6
7 import 'dart:async';
8 import 'dart:collection';
9
10 import 'utils.dart';
11
12 /// Records the values and errors that are sent through a stream and allows them
13 /// to be replayed arbitrarily many times.
14 class StreamReplayer<T> {
15 /// The wrapped stream.
16 final Stream<T> _stream;
17
18 /// Whether or not [_stream] has been closed.
19 bool _isClosed = false;
20
21 /// The buffer of events or errors that have already been emitted by
22 /// [_stream].
23 ///
24 /// Each element is a [Union] that's either a value or an error sent through
25 /// the stream.
26 final _buffer = new Queue<Union<T, dynamic>>();
27
28 /// The controllers are listening for future events from [_stream].
29 final _controllers = new Set<StreamController<T>>();
30
31 StreamReplayer(this._stream) {
32 _stream.listen((data) {
33 _buffer.add(new Union<T, dynamic>.withType1(data));
34 for (var controller in _controllers) {
35 controller.add(data);
36 }
37 }, onError: (error) {
38 _buffer.add(new Union<T, dynamic>.withType2(error));
39 for (var controller in _controllers) {
40 controller.addError(error);
41 }
42 }, onDone: () {
43 _isClosed = true;
44 for (var controller in _controllers) {
45 controller.close();
46 }
47 _controllers.clear();
48 });
49 }
50
51 /// Returns a stream that replays the values and errors of the input stream.
52 ///
53 /// This stream is a buffered stream regardless of whether the input stream
54 /// was broadcast or buffered.
55 Stream<T> getReplay() {
56 var controller = new StreamController<T>();
57 for (var eventOrError in _buffer) {
58 if (eventOrError.isType1) {
59 controller.add(eventOrError.type1);
60 } else {
61 controller.add(eventOrError.type2);
Bob Nystrom 2013/08/27 17:20:29 addError
nweiz 2013/08/27 17:47:52 Done.
62 }
63 }
64 if (_isClosed) {
65 controller.close();
66 } else {
67 _controllers.add(controller);
68 }
69 return controller.stream;
70 }
71 }
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698