Increase size for slow-consumer test (overflows memory on X64).

sdk/lib/async/stream_impl.dart

 // Copyright (c) 2012, 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.
 
 part of dart.async;
 
 // States shared by single/multi stream implementations.
 
 /// Initial and default state where the stream can receive and send events.
 const int _STREAM_OPEN = 0;
 /// The stream has received a request to complete, but hasn't done so yet.
 /// No further events can be added to the stream.
 const int _STREAM_CLOSED = 1;
 /// The stream has completed and will no longer receive or send events.
 /// Also counts as closed. The stream must not be paused when it's completed.
 /// Always used in conjunction with [_STREAM_CLOSED].
 const int _STREAM_COMPLETE = 2;
 /// Bit that alternates between events, and listeners are updated to the
 /// current value when they are notified of the event.
 const int _STREAM_EVENT_ID = 4;
 const int _STREAM_EVENT_ID_SHIFT = 2;
 /// Bit set while firing and clear while not.
 const int _STREAM_FIRING = 8;
 /// Bit set while calling a pause-state or subscription-state change callback.
 const int _STREAM_CALLBACK = 16;
 /// The count of times a stream has paused is stored in the
 /// state, shifted by this amount.
-const int _STREAM_PAUSE_COUNT_SHIFT = 8;
+const int _STREAM_PAUSE_COUNT_SHIFT = 5;
 
 // States for listeners.
 
 /// The listener is currently not subscribed to its source stream.
 const int _LISTENER_UNSUBSCRIBED = 0;
 /// The listener is actively subscribed to its source stream.
 const int _LISTENER_SUBSCRIBED = 1;
 /// The listener is subscribed until it has been notified of the current event.
 /// This flag bit is always used in conjuction with [_LISTENER_SUBSCRIBED].
 const int _LISTENER_PENDING_UNSUBSCRIBE = 2;
@@ skipping 102 matching lines @@
 
   /** Whether the stream has been closed (a done event requested). */
   bool get _isClosed => (_state & _STREAM_CLOSED) != 0;
 
   /** Whether the stream is completed. */
   bool get _isComplete => (_state & _STREAM_COMPLETE) != 0;
 
   /** Whether one or more active subscribers have requested a pause. */
   bool get _isPaused => _state >= (1 << _STREAM_PAUSE_COUNT_SHIFT);
 
+  /** Whether we are currently executing a state-chance callback. */
+  bool get _isInCallback => (_state & _STREAM_CALLBACK) != 0;
+
   /** Check whether the pending event queue is non-empty */
   bool get _hasPendingEvent =>
       _pendingEvents != null && !_pendingEvents.isEmpty;
 
   /** Whether we are currently firing an event. */
   bool get _isFiring => (_state & _STREAM_FIRING) != 0;
 
   /** Whether the state bits allow firing. */
   bool get _mayFireState {
     // The state disallows firing if:
@@ skipping 63 matching lines @@
     // This sets the _STREAM_FIRING bit and toggles the _STREAM_EVENT_ID
     // bit. All current subscribers will now have a _LISTENER_EVENT_ID
     // that doesn't match _STREAM_EVENT_ID, and they will receive the
     // event being fired.
     _state ^= _STREAM_FIRING | _STREAM_EVENT_ID;
   }
 
   void _endFiring() {
     assert(_isFiring);
     _state ^= _STREAM_FIRING;
-
     if (!_hasSubscribers) {
       _callOnSubscriptionStateChange();
     } else if (_isPaused) {
       _callOnPauseStateChange();
     }
   }
 
   /**
    * Record that a listener wants a pause from events.
    *
@@ skipping 80 matching lines @@
    * invalid state.
    *
    * This method must not be called while [isFiring] is true.
    */
   void _forEachSubscriber(void action(_StreamSubscriptionImpl<T> subscription));
 
   /** Calls [_onPauseStateChange] while setting callback bit. */
   void _callOnPauseStateChange() {
     // After calling [_close], all pauses are handled internally by the Stream.
     if (_isClosed) return;
-    _state |= _STREAM_CALLBACK;
+    if (!_isInCallback) {
+      _state |= _STREAM_CALLBACK;
+      _onPauseStateChange();
+      _state ^= _STREAM_CALLBACK;
+      return;
+    }

[Anders Johnsen, 2013/02/20 13:17:32]

Use else.

     _onPauseStateChange();
-    _state ^= _STREAM_CALLBACK;
   }
 
   /** Calls [_onSubscriptionStateChange] while setting callback bit. */
   void _callOnSubscriptionStateChange() {
-    _state |= _STREAM_CALLBACK;
+    if (!_isInCallback) {
+      _state |= _STREAM_CALLBACK;
+      _onSubscriptionStateChange();
+      _state ^= _STREAM_CALLBACK;
+      return;
+    }
     _onSubscriptionStateChange();
-    _state ^= _STREAM_CALLBACK;
   }
 
   /**
    * Called when the first subscriber requests a pause or the last a resume.
    *
    * Read [isPaused] to see the new state.
    */
   void _onPauseStateChange() {}
 
   /**
@@ skipping 109 matching lines @@
  */
 class _SingleStreamImpl<T> extends _StreamImpl<T> {
   _StreamListener _subscriber = null;
 
   // A single-stream is considered paused when it has no subscriber.
   // Exception is when it's complete (which only matters for pause-state-change
   // callbacks), where it's not considered paused.
   bool get _isPaused => (!_hasSubscribers && !_isComplete) || super._isPaused;
 
 
-  // A single-stream is considered paused when it has no subscriber.
-  bool get _canFireEvent =>
-      _mayFireState && !_hasPendingEvent && _hasSubscribers;
+  bool get _canFireEvent {
+    // A single-stream is considered paused when it has no subscriber and
+    // isn't complete, so it can't fire if there is no subscriber.
+    // It won't try to fire when completed, so no need to check that.
+    return _mayFireState && !_hasPendingEvent && _hasSubscribers;
+  }
 
 
   /** Whether there is currently a subscriber on this [Stream]. */
   bool get _hasSubscribers => _subscriber != null;
 
   // -------------------------------------------------------------------
   // Internal implementation.
 
   /**
    * Create the new subscription object.
@@ skipping 714 matching lines @@
                                      onError: this._signalError,
                                      onDone: this._close);
     } else {
       // TODO(lrn): Check why this can happen.
       if (_subscription == null) return;
       _subscription.cancel();
       _subscription = null;
     }
   }
 }