addStream method
- @override
override
Consumes the elements of stream
.
Listens on stream
and does something for each event.
Returns a future which is completed when the stream is done being added, and the consumer is ready to accept a new stream. No further calls to addStream or close should happen before the returned future has completed.
The consumer may stop listening to the stream after an error, it may consume all the errors and only stop at a done event, or it may be canceled early if the receiver don't want any further events.
If the consumer stops listening because of some error preventing it
from continuing, it may report this error in the returned future,
otherwise it will just complete the future with null
.
Implementation
@override
Future addStream(Stream<List<T>> stream) {
var lastStream = _currentStream == null ? stream : _currentStream;
if (_sub != null) {
_sub.cancel();
}
_currentStream = stream;
final streamDone = new Completer<Null>();
_sub = stream.listen((items) {
_chunks.addAll(items);
_counter += items is List ? items.length : 1;
if (limited && _counter >= limit) {
_sub.pause();
}
while (_readers.isNotEmpty && _readers.first.size <= _counter) {
var waiting = _readers.removeAt(0);
waiting.completer.complete(_consume(waiting.size));
}
}, onDone: () {
// User is piping in a new stream
if (stream == lastStream && _throwOnError) {
_closed(new UnderflowError());
}
streamDone.complete();
}, onError: (e, stack) {
_closed(e, stack);
});
return streamDone.future;
}