Checks whether test accepts any element provided by this stream.

Completes the Future when the answer is known. If this stream reports an error, the Future will report that error.

Future<bool> any(bool test(T element)) {
 _FutureImpl<bool> future = new _FutureImpl<bool>();
 StreamSubscription subscription;
 subscription = this.listen(
     (T element) {
       _runUserCode(
         () => test(element),
         (bool isMatch) {
           if (isMatch) {
             subscription.cancel();
             future._setValue(true);
           }
         },
         _cancelAndError(subscription, future)
       );
     },
     onError: future._setError,
     onDone: () {
       future._setValue(false);
     },
     cancelOnError: true);
 return future;
}

Stream<T> asBroadcastStream({void onListen(StreamSubscription subscription),
                            void onCancel(StreamSubscription subscription)})
   => _stream.asBroadcastStream(onListen: onListen, onCancel: onCancel);

Checks whether needle occurs in the elements provided by this stream.

Completes the Future when the answer is known. If this stream reports an error, the Future will report that error.

Future<bool> contains(Object needle) {
 _FutureImpl<bool> future = new _FutureImpl<bool>();
 StreamSubscription subscription;
 subscription = this.listen(
     (T element) {
       _runUserCode(
         () => (element == needle),
         (bool isMatch) {
           if (isMatch) {
             subscription.cancel();
             future._setValue(true);
           }
         },
         _cancelAndError(subscription, future)
       );
     },
     onError: future._setError,
     onDone: () {
       future._setValue(false);
     },
     cancelOnError: true);
 return future;
}

Skips data events if they are equal to the previous data event.

The returned stream provides the same events as this stream, except that it never provides two consequtive data events that are equal.

Equality is determined by the provided equals method. If that is omitted, the '==' operator on the last provided data element is used.

Stream<T> distinct([bool equals(T previous, T next)]) {
 return new _DistinctStream(this, equals);
}

Discards all data on the stream, but signals when it's done or an error occured.

When subscribing using drain, cancelOnError will be true. This means that the future will complete with the first error on the stream and then cancel the subscription.

In case of a done event the future completes with the given futureValue.

Future drain([var futureValue]) => listen(null, cancelOnError: true)
   .asFuture(futureValue);

Returns the value of the indexth data event of this stream.

If an error event occurs, the future will end with this error.

If this stream provides fewer than index elements before closing, an error is reported.

Future<T> elementAt(int index) {
 if (index is! int || index < 0) throw new ArgumentError(index);
 _FutureImpl<T> future = new _FutureImpl<T>();
 StreamSubscription subscription;
 subscription = this.listen(
   (T value) {
     if (index == 0) {
       subscription.cancel();
       future._setValue(value);
       return;
     }
     index -= 1;
   },
   onError: future._setError,
   onDone: () {
     future._setError(new StateError("Not enough elements for elementAt"));
   },
   cancelOnError: true);
 return future;
}

Checks whether test accepts all elements provided by this stream.

Completes the Future when the answer is known. If this stream reports an error, the Future will report that error.

Future<bool> every(bool test(T element)) {
 _FutureImpl<bool> future = new _FutureImpl<bool>();
 StreamSubscription subscription;
 subscription = this.listen(
     (T element) {
       _runUserCode(
         () => test(element),
         (bool isMatch) {
           if (!isMatch) {
             subscription.cancel();
             future._setValue(false);
           }
         },
         _cancelAndError(subscription, future)
       );
     },
     onError: future._setError,
     onDone: () {
       future._setValue(true);
     },
     cancelOnError: true);
 return future;
}

Creates a new stream from this stream that converts each element into zero or more events.

Each incoming event is converted to an Iterable of new events, and each of these new events are then sent by the returned stream in order.

Stream expand(Iterable convert(T value)) {
 return new _ExpandStream<T, dynamic>(this, convert);
}

Finds the first element of this stream matching test.

Returns a future that is filled with the first element of this stream that test returns true for.

If no such element is found before this stream is done, and a defaultValue function is provided, the result of calling defaultValue becomes the value of the future.

If an error occurs, or if this stream ends without finding a match and with no defaultValue function provided, the future will receive an error.

Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) {
 _FutureImpl<dynamic> future = new _FutureImpl();
 StreamSubscription subscription;
 subscription = this.listen(
   (T value) {
     _runUserCode(
       () => test(value),
       (bool isMatch) {
         if (isMatch) {
           subscription.cancel();
           future._setValue(value);
         }
       },
       _cancelAndError(subscription, future)
     );
   },
   onError: future._setError,
   onDone: () {
     if (defaultValue != null) {
       _runUserCode(defaultValue, future._setValue, future._setError);
       return;
     }
     future._setError(new StateError("firstMatch ended without match"));
   },
   cancelOnError: true);
 return future;
}

Reduces a sequence of values by repeatedly applying combine.

Future fold(var initialValue, combine(var previous, T element)) {
 _FutureImpl result = new _FutureImpl();
 var value = initialValue;
 StreamSubscription subscription;
 subscription = this.listen(
   (T element) {
     _runUserCode(
       () => combine(value, element),
       (newValue) { value = newValue; },
       _cancelAndError(subscription, result)
     );
   },
   onError: (e) {
     result._setError(e);
   },
   onDone: () {
     result._setValue(value);
   },
   cancelOnError: true);
 return result;
}

Executes action on each data event of the stream.

Completes the returned Future when all events of the stream have been processed. Completes the future with an error if the stream has an error event, or if action throws.

Future forEach(void action(T element)) {
 _FutureImpl future = new _FutureImpl();
 StreamSubscription subscription;
 subscription = this.listen(
     (T element) {
       _runUserCode(
         () => action(element),
         (_) {},
         _cancelAndError(subscription, future)
       );
     },
     onError: future._setError,
     onDone: () {
       future._setValue(null);
     },
     cancelOnError: true);
 return future;
}

Creates a wrapper Stream that intercepts some errors from this stream.

If this stream sends an error that matches test, then it is intercepted by the handle function.

An [AsyncError] [:e:] is matched by a test function if test(e) returns true. If test is omitted, every error is considered matching.

If the error is intercepted, the handle function can decide what to do with it. It can throw if it wants to raise a new (or the same) error, or simply return to make the stream forget the error.

If you need to transform an error into a data event, use the more generic Stream.transformEvent to handle the event by writing a data event to the output sink

Stream<T> handleError(void handle( error), { bool test(error) }) {
 return new _HandleErrorStream<T>(this, handle, test);
}

Collects string of data events' string representations.

If separator is provided, it is inserted between any two elements.

Any error in the stream causes the future to complete with that error. Otherwise it completes with the collected string when the "done" event arrives.

Future<String> join([String separator = ""]) {
 _FutureImpl<String> result = new _FutureImpl<String>();
 StringBuffer buffer = new StringBuffer();
 StreamSubscription subscription;
 bool first = true;
 subscription = this.listen(
   (T element) {
     if (!first) {
       buffer.write(separator);
     }
     first = false;
     try {
       buffer.write(element);
     } catch (e, s) {
       subscription.cancel();
       result._setError(_asyncError(e, s));
     }
   },
   onError: (e) {
     result._setError(e);
   },
   onDone: () {
     result._setValue(buffer.toString());
   },
   cancelOnError: true);
 return result;
}

Finds the last element in this stream matching test.

As firstWhere, except that the last matching element is found. That means that the result cannot be provided before this stream is done.

Future<dynamic> lastWhere(bool test(T element), {Object defaultValue()}) {
 _FutureImpl<dynamic> future = new _FutureImpl();
 T result = null;
 bool foundResult = false;
 StreamSubscription subscription;
 subscription = this.listen(
   (T value) {
     _runUserCode(
       () => true == test(value),
       (bool isMatch) {
         if (isMatch) {
           foundResult = true;
           result = value;
         }
       },
       _cancelAndError(subscription, future)
     );
   },
   onError: future._setError,
   onDone: () {
     if (foundResult) {
       future._setValue(result);
       return;
     }
     if (defaultValue != null) {
       _runUserCode(defaultValue, future._setValue, future._setError);
       return;
     }
     future._setError(new StateError("lastMatch ended without match"));
   },
   cancelOnError: true);
 return future;
}

StreamSubscription<T> listen(void onData(T value),
                            { void onError(error),
                              void onDone(),
                              bool cancelOnError }) {
 return _stream.listen(onData, onError: onError, onDone: onDone,
                       cancelOnError: cancelOnError);
}

Creates a new stream that converts each element of this stream to a new value using the convert function.

Stream map(convert(T event)) {
 return new _MapStream<T, dynamic>(this, convert);
}

Binds this stream as the input of the provided StreamConsumer.

Future pipe(StreamConsumer<T> streamConsumer) {
 return streamConsumer.addStream(this).then((_) => streamConsumer.close());
}

Reduces a sequence of values by repeatedly applying combine.

Future<T> reduce(T combine(T previous, T element)) {
 _FutureImpl<T> result = new _FutureImpl<T>();
 bool seenFirst = false;
 T value;
 StreamSubscription subscription;
 subscription = this.listen(
   (T element) {
     if (seenFirst) {
       _runUserCode(() => combine(value, element),
                    (T newValue) { value = newValue; },
                    _cancelAndError(subscription, result));
     } else {
       value = element;
       seenFirst = true;
     }
   },
   onError: result._setError,
   onDone: () {
     if (!seenFirst) {
       result._setError(new StateError("No elements"));
     } else {
       result._setValue(value);
     }
   },
   cancelOnError: true
 );
 return result;
}

Finds the single element in this stream matching test.

Like lastMatch, except that it is an error if more than one matching element occurs in the stream.

Future<T> singleWhere(bool test(T element)) {
 _FutureImpl<T> future = new _FutureImpl<T>();
 T result = null;
 bool foundResult = false;
 StreamSubscription subscription;
 subscription = this.listen(
   (T value) {
     _runUserCode(
       () => true == test(value),
       (bool isMatch) {
         if (isMatch) {
           if (foundResult) {
             subscription.cancel();
             future._setError(
                 new StateError('Multiple matches for "single"'));
             return;
           }
           foundResult = true;
           result = value;
         }
       },
       _cancelAndError(subscription, future)
     );
   },
   onError: future._setError,
   onDone: () {
     if (foundResult) {
       future._setValue(result);
       return;
     }
     future._setError(new StateError("single ended without match"));
   },
   cancelOnError: true);
 return future;
}

Skips the first count data events from this stream.

Stream<T> skip(int count) {
 return new _SkipStream(this, count);
}

Skip data events from this stream while they are matched by test.

Error and done events are provided by the returned stream unmodified.

Starting with the first data event where test returns true for the event data, the returned stream will have the same events as this stream.

Stream<T> skipWhile(bool test(T element)) {
 return new _SkipWhileStream(this, test);
}

Provides at most the first n values of this stream.

Forwards the first n data events of this stream, and all error events, to the returned stream, and ends with a done event.

If this stream produces fewer than count values before it's done, so will the returned stream.

Stream<T> take(int count) {
 return new _TakeStream(this, count);
}

Forwards data events while test is successful.

The returned stream provides the same events as this stream as long as test returns true for the event data. The stream is done when either this stream is done, or when this stream first provides a value that test doesn't accept.

Stream<T> takeWhile(bool test(T element)) {
 return new _TakeWhileStream(this, test);
}

Collects the data of this stream in a List.

Future<List<T>> toList() {
 List<T> result = <T>[];
 _FutureImpl<List<T>> future = new _FutureImpl<List<T>>();
 this.listen(
   (T data) {
     result.add(data);
   },
   onError: future._setError,
   onDone: () {
     future._setValue(result);
   },
   cancelOnError: true);
 return future;
}

Collects the data of this stream in a Set.

Future<Set<T>> toSet() {
 Set<T> result = new Set<T>();
 _FutureImpl<Set<T>> future = new _FutureImpl<Set<T>>();
 this.listen(
   (T data) {
     result.add(data);
   },
   onError: future._setError,
   onDone: () {
     future._setValue(result);
   },
   cancelOnError: true);
 return future;
}

Chains this stream as the input of the provided StreamTransformer.

Returns the result of streamTransformer.bind itself.

Stream transform(StreamTransformer<T, dynamic> streamTransformer) {
 return streamTransformer.bind(this);
}

Creates a new stream from this stream that discards some data events.

The new stream sends the same error and done events as this stream, but it only sends the data events that satisfy the test.

Stream<T> where(bool test(T event)) {
 return new _WhereStream<T>(this, test);
}