OLD | NEW |
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 stream_state_helper; | 5 library stream_state_helper; |
6 | 6 |
7 import "../../../pkg/unittest/lib/unittest.dart"; | 7 import "../../../pkg/unittest/lib/unittest.dart"; |
8 import "dart:async"; | 8 import "dart:async"; |
9 import "dart:collection"; | 9 import "dart:collection"; |
10 | 10 |
11 class StreamProtocolTest { | 11 class StreamProtocolTest { |
12 StreamController _controller; | 12 StreamController _controller; |
| 13 Stream _controllerStream; |
13 StreamSubscription _subscription; | 14 StreamSubscription _subscription; |
14 List<Event> _expectations = new List<Event>(); | 15 List<Event> _expectations = new List<Event>(); |
15 int _nextExpectationIndex = 0; | 16 int _nextExpectationIndex = 0; |
16 Function _onComplete; | 17 Function _onComplete; |
17 | 18 |
18 StreamProtocolTest([bool broadcast = false]) { | 19 StreamProtocolTest([bool broadcast = false]) { |
| 20 _controller = new StreamController( |
| 21 onPauseStateChange: _onPause, |
| 22 onSubscriptionStateChange: _onSubcription); |
| 23 // TODO(lrn): Make it work with multiple subscribers too. |
19 if (broadcast) { | 24 if (broadcast) { |
20 _controller = new StreamController.broadcast( | 25 _controllerStream = _controller.stream.asBroadcastStream(); |
21 onPauseStateChange: _onPause, | |
22 onSubscriptionStateChange: _onSubcription); | |
23 // TODO(lrn): Make it work with multiple subscribers too. | |
24 } else { | 26 } else { |
25 _controller = new StreamController( | 27 _controllerStream = _controller.stream; |
26 onPauseStateChange: _onPause, | |
27 onSubscriptionStateChange: _onSubcription); | |
28 } | 28 } |
29 _onComplete = expectAsync0((){ | 29 _onComplete = expectAsync0((){ |
30 _onComplete = null; // Being null marks the test to be complete. | 30 _onComplete = null; // Being null marks the test to be complete. |
31 }); | 31 }); |
32 } | 32 } |
33 | 33 |
34 // Actions on the stream and controller. | 34 // Actions on the stream and controller. |
35 void add(var data) { _controller.add(data); } | 35 void add(var data) { _controller.add(data); } |
36 void error(var error) { _controller.addError(error); } | 36 void error(var error) { _controller.addError(error); } |
37 void close() { _controller.close(); } | 37 void close() { _controller.close(); } |
38 | 38 |
39 void subscribe({bool unsubscribeOnError : false}) { | 39 void subscribe({bool unsubscribeOnError : false}) { |
40 // TODO(lrn): Handle more subscriptions (e.g., a subscription-id | 40 // TODO(lrn): Handle more subscriptions (e.g., a subscription-id |
41 // per subscription, and an id on event _expectations). | 41 // per subscription, and an id on event _expectations). |
42 if (_subscription != null) throw new StateError("Already subscribed"); | 42 if (_subscription != null) throw new StateError("Already subscribed"); |
43 _subscription = _controller.stream.listen(_onData, | 43 _subscription = _controllerStream.listen(_onData, |
44 onError: _onError, | 44 onError: _onError, |
45 onDone: _onDone, | 45 onDone: _onDone, |
46 unsubscribeOnError: | 46 unsubscribeOnError: |
47 unsubscribeOnError); | 47 unsubscribeOnError); |
48 } | 48 } |
49 | 49 |
50 void pause([Future resumeSignal]) { | 50 void pause([Future resumeSignal]) { |
51 if (_subscription == null) throw new StateError("Not subscribed"); | 51 if (_subscription == null) throw new StateError("Not subscribed"); |
52 _subscription.pause(resumeSignal); | 52 _subscription.pause(resumeSignal); |
53 } | 53 } |
54 | 54 |
55 void resume([Future resumeSignal]) { | 55 void resume([Future resumeSignal]) { |
56 if (_subscription == null) throw new StateError("Not subscribed"); | 56 if (_subscription == null) throw new StateError("Not subscribed"); |
57 _subscription.resume(); | 57 _subscription.resume(); |
(...skipping 224 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
282 _actual = "*[Paused:${c.isPaused}]"; | 282 _actual = "*[Paused:${c.isPaused}]"; |
283 return true; | 283 return true; |
284 } | 284 } |
285 bool _testSubcribe(StreamController c) { | 285 bool _testSubcribe(StreamController c) { |
286 _actual = "*[Has listener:${c.hasListener}, Paused:${c.isPaused}]"; | 286 _actual = "*[Has listener:${c.hasListener}, Paused:${c.isPaused}]"; |
287 return true; | 287 return true; |
288 } | 288 } |
289 | 289 |
290 String toString() => _actual; | 290 String toString() => _actual; |
291 } | 291 } |
OLD | NEW |