pkg/scheduled_test/lib/scheduled_stream.dart - sdk.git - Git at Google

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.

 library scheduled_test.scheduled_stream;

 import 'dart:async';
 import 'dart:collection';

 import 'package:stack_trace/stack_trace.dart';

 import 'scheduled_test.dart';
 import 'src/stream_matcher.dart';
 import 'src/utils.dart';

 export 'src/stream_matcher.dart';

 /// A wrapper for streams that supports a pull-based model of retrieving values
 /// as well as a set of [StreamMatcher]s for testing the values emitted by the
 /// stream.
 ///
 /// The only method on [ScheduledStream] that's actually scheduled is [expect],
 /// which is the method that users testing streaming code are most likely to
 /// want to use.
 class ScheduledStream<T> {
   /// The underlying stream.
   final Stream<T> _stream;

   /// The subscription to [_stream].
   StreamSubscription<T> _subscription;

   /// The completer for emitting a value requested by [next].
   ///
   /// If this is non-null, [_pendingValues] will always be empty, since any
   /// value coming in will be passed to this completer.
   Completer<T> _nextCompleter;

   /// The completer for emitting a value requested by [hasNext].
   Completer<bool> _hasNextCompleter;

   /// The set of all streams forked from this one.
   final _forks = new Set<ScheduledStream<T>>();
   final _forkControllers = new Set<StreamController<T>>();

   /// The queue of values emitted by [_stream] but not yet emitted through
   /// [next].
   final _pendingValues = new Queue<Fallible<T>>();

   /// All values emitted by this stream so far.
   ///
   /// This does not include values emitted by the underlying stream but not yet
   /// emitted through [next].
   List<T> get emittedValues => new UnmodifiableListView(_emittedValues);
   final _emittedValues = new List<T>();

   /// All values emitted by the underlying stream.
   ///
   /// This is intended primarily for providing debugging information.
   List<T> get allValues {
     var list = new List<T>.from(_emittedValues);
     list.addAll(_pendingValues.where((value) => value.hasValue)
         .map((value) => value.value));
     return new UnmodifiableListView(list);
   }

   /// Whether the wrapped stream has been closed.
   bool _isDone = false;

   /// Whether [next] has been called but has not yet returned.
   ///
   /// This is distinct from `_nextCompleter != null` when [next] is called while
   /// there are pending values available, until the future it returns completes.
   bool _isNextPending = false;

   /// Creates a new scheduled stream wrapping [stream].
   ScheduledStream(this._stream) {
     _subscription = _stream.listen((value) {
       for (var c in _forkControllers) {
         c.add(value);
       }
       if (_hasNextCompleter != null) {
         _hasNextCompleter.complete(true);
         _hasNextCompleter = null;
       }

       if (_nextCompleter != null) {
         _nextCompleter.complete(value);
         _emittedValues.add(value);
         _nextCompleter = null;
         _isNextPending = false;
       } else {
         _pendingValues.add(new Fallible.withValue(value));
       }
     }, onError: (error, stackTrace) {
       for (var c in _forkControllers) {
         c.addError(error, stackTrace);
       }
       if (_hasNextCompleter != null) {
         _hasNextCompleter.completeError(error, stackTrace);
         _hasNextCompleter = null;
       }

       if (_nextCompleter != null) {
         _nextCompleter.completeError(error, stackTrace);
         _nextCompleter = null;
       } else {
         _pendingValues.add(new Fallible.withError(error, stackTrace));
       }
     }, onDone: _onDone);
   }

   /// Enqueue an expectation that [streamMatcher] will match the value(s)
   /// emitted by the stream at this point in the schedule.
   ///
   /// If [streamMatcher] is a [StreamMatcher], it will match the stream as a
   /// whole. If it's a [Matcher] or another object, it will match the next value
   /// emitted by the stream (as though it were a [nextValue] matcher).
   ///
   /// This call is scheduled; the expectation won't be added until the schedule
   /// reaches this point, and the schedule won't continue until the matcher has
   /// matched the stream.
   void expect(streamMatcher) {
     streamMatcher = new StreamMatcher.wrap(streamMatcher);
     var description = 'stream emits $streamMatcher';
     schedule(() {
       return streamMatcher.tryMatch(this).then((description) {
         if (description == null) return;

         var expected = prefixLines(streamMatcher.toString(),
             firstPrefix: 'Expected: ',
             prefix:      '        | ');

         var actual = prefixLines(allValues.map((value) {
           return prefixLines(value.toString(), firstPrefix: '* ');
         }).join('\n'),
             firstPrefix: ' Emitted: ',
             prefix:      '          ');

         var which = '';
         if (description.length > 0) {
           which = '\n' + prefixLines(description.toString(),
               firstPrefix: '   Which: ',
               prefix:      '        | ');
         }

         fail("$expected\n$actual$which");
       });
     }, description);
   }

   /// Returns a Future that completes to the next value emitted by this stream.
   ///
   /// It's a [StateError] to call [next] when another call's Future has not yet
   /// completed, or when the stream has no more values. The latter can be
   /// checked using [hasNext].
   Future<T> next() {
     if (_isNextPending) {
       return new Future.error(
           new StateError("There's already a pending call to "
               "ScheduledStream.next."),
           new Chain.current());
     }

     if (_pendingValues.isNotEmpty) {
       _isNextPending = true;

       var valueOrError = _pendingValues.removeFirst();
       if (valueOrError.hasValue) {
         _emittedValues.add(valueOrError.value);
       }
       return valueOrError.toFuture().whenComplete(() {
         _isNextPending = false;
       });
     } else if (_isDone) {
       return new Future.error(
           new StateError("ScheduledStream has no more elements."),
           new Chain.current());
     }

     _isNextPending = true;
     _nextCompleter = new Completer();
     return _nextCompleter.future;
   }

   /// Returns a Future that completes to a boolean indicating whether the stream
   /// has additional values or not.
   Future<bool> get hasNext {
     if (_hasNextCompleter != null) return _hasNextCompleter.future;

     if (_pendingValues.isNotEmpty) {
       return _pendingValues.first.toFuture().then((_) => true);
     } else if (_isDone) {
       return new Future.value(false);
     }

     _hasNextCompleter = new Completer();
     return _hasNextCompleter.future;
   }

   /// Returns a fork of this stream.
   ///
   /// The fork begins at the same point [this] is at. Values can be read from it
   /// without consuming values in [this]. If [this] is closed, the fork will be
   /// closed at whatever point it's currently at.
   ScheduledStream<T> fork() {
     var controller = new StreamController<T>();
     for (var valueOrError in _pendingValues) {
       if (valueOrError.hasValue) {
         controller.add(valueOrError.value);
       } else {
         controller.addError(valueOrError.error, valueOrError.stackTrace);
       }
     }
     if (_isDone) {
       controller.close();
     } else {
       _forkControllers.add(controller);
     }

     var fork = new ScheduledStream<T>(controller.stream);
     _forks.add(fork);
     return fork;
   }

   /// Closes this stream.
   ///
   /// This cancels the subscription to the underlying stream and acts as though
   /// [this] was closed immediately after the current position, regardless of
   /// whether the underlying stream has emitted additional events.
   void close() {
     _subscription.cancel();
     _pendingValues.clear();

     for (var fork in _forks) {
       fork.close();
     }
     _forks.clear();

     if (!_isDone) _onDone();
   }

   /// Handles a "done" event from the underlying stream, as well as [this] being
   /// closed.
   void _onDone() {
     for (var c in _forkControllers) {
       c.close();
     }
     _forkControllers.clear();

     if (_hasNextCompleter != null) {
       _hasNextCompleter.complete(false);
       _hasNextCompleter = null;
     }

     if (_nextCompleter != null) {
       _nextCompleter.completeError(
           new StateError("ScheduledStream has no more elements."),
           new Chain.current());
       _nextCompleter = null;
       _isNextPending = false;
     }

     _isDone = true;
   }
 }
	// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
	// for details. All rights reserved. Use of this source code is governed by a
	// BSD-style license that can be found in the LICENSE file.

	library scheduled_test.scheduled_stream;

	import 'dart:async';
	import 'dart:collection';

	import 'package:stack_trace/stack_trace.dart';

	import 'scheduled_test.dart';
	import 'src/stream_matcher.dart';
	import 'src/utils.dart';

	export 'src/stream_matcher.dart';

	/// A wrapper for streams that supports a pull-based model of retrieving values
	/// as well as a set of [StreamMatcher]s for testing the values emitted by the
	/// stream.
	///
	/// The only method on [ScheduledStream] that's actually scheduled is [expect],
	/// which is the method that users testing streaming code are most likely to
	/// want to use.
	class ScheduledStream<T> {
	/// The underlying stream.
	final Stream<T> _stream;

	/// The subscription to [_stream].
	StreamSubscription<T> _subscription;

	/// The completer for emitting a value requested by [next].
	///
	/// If this is non-null, [_pendingValues] will always be empty, since any
	/// value coming in will be passed to this completer.
	Completer<T> _nextCompleter;

	/// The completer for emitting a value requested by [hasNext].
	Completer<bool> _hasNextCompleter;

	/// The set of all streams forked from this one.
	final _forks = new Set<ScheduledStream<T>>();
	final _forkControllers = new Set<StreamController<T>>();

	/// The queue of values emitted by [_stream] but not yet emitted through
	/// [next].
	final _pendingValues = new Queue<Fallible<T>>();

	/// All values emitted by this stream so far.
	///
	/// This does not include values emitted by the underlying stream but not yet
	/// emitted through [next].
	List<T> get emittedValues => new UnmodifiableListView(_emittedValues);
	final _emittedValues = new List<T>();

	/// All values emitted by the underlying stream.
	///
	/// This is intended primarily for providing debugging information.
	List<T> get allValues {
	var list = new List<T>.from(_emittedValues);
	list.addAll(_pendingValues.where((value) => value.hasValue)
	.map((value) => value.value));
	return new UnmodifiableListView(list);
	}

	/// Whether the wrapped stream has been closed.
	bool _isDone = false;

	/// Whether [next] has been called but has not yet returned.
	///
	/// This is distinct from `_nextCompleter != null` when [next] is called while
	/// there are pending values available, until the future it returns completes.
	bool _isNextPending = false;

	/// Creates a new scheduled stream wrapping [stream].
	ScheduledStream(this._stream) {
	_subscription = _stream.listen((value) {
	for (var c in _forkControllers) {
	c.add(value);
	}
	if (_hasNextCompleter != null) {
	_hasNextCompleter.complete(true);
	_hasNextCompleter = null;
	}

	if (_nextCompleter != null) {
	_nextCompleter.complete(value);
	_emittedValues.add(value);
	_nextCompleter = null;
	_isNextPending = false;
	} else {
	_pendingValues.add(new Fallible.withValue(value));
	}
	}, onError: (error, stackTrace) {
	for (var c in _forkControllers) {
	c.addError(error, stackTrace);
	}
	if (_hasNextCompleter != null) {
	_hasNextCompleter.completeError(error, stackTrace);
	_hasNextCompleter = null;
	}

	if (_nextCompleter != null) {
	_nextCompleter.completeError(error, stackTrace);
	_nextCompleter = null;
	} else {
	_pendingValues.add(new Fallible.withError(error, stackTrace));
	}
	}, onDone: _onDone);
	}

	/// Enqueue an expectation that [streamMatcher] will match the value(s)
	/// emitted by the stream at this point in the schedule.
	///
	/// If [streamMatcher] is a [StreamMatcher], it will match the stream as a
	/// whole. If it's a [Matcher] or another object, it will match the next value
	/// emitted by the stream (as though it were a [nextValue] matcher).
	///
	/// This call is scheduled; the expectation won't be added until the schedule
	/// reaches this point, and the schedule won't continue until the matcher has
	/// matched the stream.
	void expect(streamMatcher) {
	streamMatcher = new StreamMatcher.wrap(streamMatcher);
	var description = 'stream emits $streamMatcher';
	schedule(() {
	return streamMatcher.tryMatch(this).then((description) {
	if (description == null) return;

	var expected = prefixLines(streamMatcher.toString(),
	firstPrefix: 'Expected: ',
	prefix: ' \| ');

	var actual = prefixLines(allValues.map((value) {
	return prefixLines(value.toString(), firstPrefix: '* ');
	}).join('\n'),
	firstPrefix: ' Emitted: ',
	prefix: ' ');

	var which = '';
	if (description.length > 0) {
	which = '\n' + prefixLines(description.toString(),
	firstPrefix: ' Which: ',
	prefix: ' \| ');
	}

	fail("$expected\n$actual$which");
	});
	}, description);
	}

	/// Returns a Future that completes to the next value emitted by this stream.
	///
	/// It's a [StateError] to call [next] when another call's Future has not yet
	/// completed, or when the stream has no more values. The latter can be
	/// checked using [hasNext].
	Future<T> next() {
	if (_isNextPending) {
	return new Future.error(
	new StateError("There's already a pending call to "
	"ScheduledStream.next."),
	new Chain.current());
	}

	if (_pendingValues.isNotEmpty) {
	_isNextPending = true;

	var valueOrError = _pendingValues.removeFirst();
	if (valueOrError.hasValue) {
	_emittedValues.add(valueOrError.value);
	}
	return valueOrError.toFuture().whenComplete(() {
	_isNextPending = false;
	});
	} else if (_isDone) {
	return new Future.error(
	new StateError("ScheduledStream has no more elements."),
	new Chain.current());
	}

	_isNextPending = true;
	_nextCompleter = new Completer();
	return _nextCompleter.future;
	}

	/// Returns a Future that completes to a boolean indicating whether the stream
	/// has additional values or not.
	Future<bool> get hasNext {
	if (_hasNextCompleter != null) return _hasNextCompleter.future;

	if (_pendingValues.isNotEmpty) {
	return _pendingValues.first.toFuture().then((_) => true);
	} else if (_isDone) {
	return new Future.value(false);
	}

	_hasNextCompleter = new Completer();
	return _hasNextCompleter.future;
	}

	/// Returns a fork of this stream.
	///
	/// The fork begins at the same point [this] is at. Values can be read from it
	/// without consuming values in [this]. If [this] is closed, the fork will be
	/// closed at whatever point it's currently at.
	ScheduledStream<T> fork() {
	var controller = new StreamController<T>();
	for (var valueOrError in _pendingValues) {
	if (valueOrError.hasValue) {
	controller.add(valueOrError.value);
	} else {
	controller.addError(valueOrError.error, valueOrError.stackTrace);
	}
	}
	if (_isDone) {
	controller.close();
	} else {
	_forkControllers.add(controller);
	}

	var fork = new ScheduledStream<T>(controller.stream);
	_forks.add(fork);
	return fork;
	}

	/// Closes this stream.
	///
	/// This cancels the subscription to the underlying stream and acts as though
	/// [this] was closed immediately after the current position, regardless of
	/// whether the underlying stream has emitted additional events.
	void close() {
	_subscription.cancel();
	_pendingValues.clear();

	for (var fork in _forks) {
	fork.close();
	}
	_forks.clear();

	if (!_isDone) _onDone();
	}

	/// Handles a "done" event from the underlying stream, as well as [this] being
	/// closed.
	void _onDone() {
	for (var c in _forkControllers) {
	c.close();
	}
	_forkControllers.clear();

	if (_hasNextCompleter != null) {
	_hasNextCompleter.complete(false);
	_hasNextCompleter = null;
	}

	if (_nextCompleter != null) {
	_nextCompleter.completeError(
	new StateError("ScheduledStream has no more elements."),
	new Chain.current());
	_nextCompleter = null;
	_isNextPending = false;
	}

	_isDone = true;
	}
	}