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 part of dart.async; | 5 part of dart.async; |
6 | 6 |
7 /** | 7 /** |
8 * Wraps an [_EventSink] so it exposes only the [EventSink] interface. | 8 * Wraps an [_EventSink] so it exposes only the [EventSink] interface. |
9 */ | 9 */ |
10 class _EventSinkWrapper<T> implements EventSink<T> { | 10 class _EventSinkWrapper<T> implements EventSink<T> { |
(...skipping 191 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
202 * Wraps handlers (from [StreamTransformer.fromHandlers]) into an `EventSink`. | 202 * Wraps handlers (from [StreamTransformer.fromHandlers]) into an `EventSink`. |
203 * | 203 * |
204 * This way we can reuse the code from [_StreamSinkTransformer]. | 204 * This way we can reuse the code from [_StreamSinkTransformer]. |
205 */ | 205 */ |
206 class _HandlerEventSink<S, T> implements EventSink<S> { | 206 class _HandlerEventSink<S, T> implements EventSink<S> { |
207 final _TransformDataHandler<S, T> _handleData; | 207 final _TransformDataHandler<S, T> _handleData; |
208 final _TransformErrorHandler<T> _handleError; | 208 final _TransformErrorHandler<T> _handleError; |
209 final _TransformDoneHandler<T> _handleDone; | 209 final _TransformDoneHandler<T> _handleDone; |
210 | 210 |
211 /// The output sink where the handlers should send their data into. | 211 /// The output sink where the handlers should send their data into. |
212 final EventSink<T> _sink; | 212 EventSink<T> _sink; |
213 | 213 |
214 _HandlerEventSink( | 214 _HandlerEventSink( |
215 this._handleData, this._handleError, this._handleDone, this._sink); | 215 this._handleData, this._handleError, this._handleDone, this._sink) { |
| 216 if (_sink == null) { |
| 217 throw new ArgumentError("The provided sink must not be null."); |
| 218 } |
| 219 } |
| 220 |
| 221 bool get _isClosed => _sink == null; |
216 | 222 |
217 void add(S data) { | 223 void add(S data) { |
| 224 if (_isClosed) throw new StateError("Sink is closed"); |
218 if (_handleData != null) { | 225 if (_handleData != null) { |
219 _handleData(data, _sink); | 226 _handleData(data, _sink); |
220 } else { | 227 } else { |
221 _sink.add(data as T); | 228 _sink.add(data as T); |
222 } | 229 } |
223 } | 230 } |
224 | 231 |
225 void addError(Object error, [StackTrace stackTrace]) { | 232 void addError(Object error, [StackTrace stackTrace]) { |
| 233 if (_isClosed) throw new StateError("Sink is closed"); |
226 if (_handleError != null) { | 234 if (_handleError != null) { |
227 _handleError(error, stackTrace, _sink); | 235 _handleError(error, stackTrace, _sink); |
228 } else { | 236 } else { |
229 _sink.addError(error, stackTrace); | 237 _sink.addError(error, stackTrace); |
230 } | 238 } |
231 } | 239 } |
232 | 240 |
233 void close() { | 241 void close() { |
| 242 if (_isClosed) return; |
| 243 var sink = _sink; |
| 244 _sink = null; |
234 if (_handleDone != null) { | 245 if (_handleDone != null) { |
235 _handleDone(_sink); | 246 _handleDone(sink); |
236 } else { | 247 } else { |
237 _sink.close(); | 248 sink.close(); |
238 } | 249 } |
239 } | 250 } |
240 } | 251 } |
241 | 252 |
242 /** | 253 /** |
243 * A StreamTransformer that transformers events with the given handlers. | 254 * A StreamTransformer that transformers events with the given handlers. |
244 * | 255 * |
245 * Note that this transformer can only be used once. | 256 * Note that this transformer can only be used once. |
246 */ | 257 */ |
247 class _StreamHandlerTransformer<S, T> extends _StreamSinkTransformer<S, T> { | 258 class _StreamHandlerTransformer<S, T> extends _StreamSinkTransformer<S, T> { |
(...skipping 52 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
300 StreamSubscription<T> listen(void onData(T event), | 311 StreamSubscription<T> listen(void onData(T event), |
301 {Function onError, void onDone(), bool cancelOnError}) { | 312 {Function onError, void onDone(), bool cancelOnError}) { |
302 cancelOnError = identical(true, cancelOnError); | 313 cancelOnError = identical(true, cancelOnError); |
303 StreamSubscription<T> result = _transformer(_stream, cancelOnError); | 314 StreamSubscription<T> result = _transformer(_stream, cancelOnError); |
304 result.onData(onData); | 315 result.onData(onData); |
305 result.onError(onError); | 316 result.onError(onError); |
306 result.onDone(onDone); | 317 result.onDone(onDone); |
307 return result; | 318 return result; |
308 } | 319 } |
309 } | 320 } |
OLD | NEW |