Window updates. (#12)

Make sure the positiveWindow is marked un-buffered if the window opens
due to an increase in initial window size. Otherwise, even sub-sequent
WindowUpdate frames wouldn't cause a bufferEmpty event to get emitted.

Also, if the connection preface doesn't match, don't try to add data to
the terminated connection.

Author: jakobr-google

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## Unreleased
+
+* Fixed a bug where a closed window would not open correctly due to an increase
+  in initial window size.
+
 ## 0.1.2
 
 * The endStream bit is now set on the requested frame, instead of on an empty

lib/src/connection_preface.dart

@@ -38,15 +38,16 @@ Stream<List<int>> readConnectionPreface(Stream<List<int>> incoming) {
     terminated = true;
   }
 
-  void compareConnectionPreface(List<int> data) {
+  bool compareConnectionPreface(List<int> data) {
     for (int i = 0; i < CONNECTION_PREFACE.length; i++) {
       if (data[i] != CONNECTION_PREFACE[i]) {
         terminate('Connection preface does not match.');
-        break;
+        return false;
       }
     }
     prefaceBuffer = null;
     connectionPrefaceRead = true;
+    return true;
   }
 
   void onData(List<int> data) {
@@ -55,7 +56,7 @@ Stream<List<int>> readConnectionPreface(Stream<List<int>> incoming) {
       result.add(data);
     } else {
       if (prefaceBuffer.isEmpty && data.length > CONNECTION_PREFACE.length) {
-        compareConnectionPreface(data);
+        if (!compareConnectionPreface(data)) return;
         data = data.sublist(CONNECTION_PREFACE.length);
       } else if (prefaceBuffer.length < CONNECTION_PREFACE.length) {
         int remaining = CONNECTION_PREFACE.length - prefaceBuffer.length;
@@ -66,7 +67,7 @@ Stream<List<int>> readConnectionPreface(Stream<List<int>> incoming) {
         prefaceBuffer.addAll(part1);
 
         if (prefaceBuffer.length == CONNECTION_PREFACE.length) {
-          compareConnectionPreface(prefaceBuffer);
+          if (!compareConnectionPreface(prefaceBuffer)) return;
         }
         data = part2;
       }

lib/src/flowcontrol/window.dart

@@ -15,26 +15,13 @@ class Window {
   /// NOTE: This value can potentially become negative.
   int _size;
 
-  /// The size the window would normally have if there is no outstanding
-  /// data.
-  ///
-  /// NOTE: The peer can always increase a stream window above this default
-  /// limit.
-  int _defaultSize;
-
   Window({int initialSize: (1 << 16) - 1})
-      : _size = initialSize, _defaultSize = initialSize;
+      : _size = initialSize;
 
   /// The current size of the flow control window.
   int get size => _size;
 
   void modify(int difference) {
     _size += difference;
   }
-
-  /// This method can be e.g. called after receiving a SettingsFrame
-  /// which changes the initial window size of all streams.
-  void modifyDefaultSize(int difference) {
-    _defaultSize += difference;
-  }
 }

lib/src/flowcontrol/window_handler.dart

@@ -30,8 +30,6 @@ abstract class AbstractOutgoingWindowHandler {
 
   /// Process a window update frame received from the remote end.
   void processWindowUpdate(WindowUpdateFrame frame) {
-    int oldWindowSize = _peerWindow.size;
-
     int increment = frame.windowSizeIncrement;
     if ((_peerWindow.size + increment) > Window.MAX_WINDOW_SIZE) {
       throw new FlowControlException(
@@ -43,7 +41,7 @@ abstract class AbstractOutgoingWindowHandler {
 
     // If we transitioned from an negative/empty window to a positive window
     // we'll fire an event that more data frames can be sent now.
-    if (oldWindowSize <= 0 && _peerWindow.size > 0) {
+    if (positiveWindow.wouldBuffer && _peerWindow.size > 0) {
       positiveWindow.markUnBuffered();
     }
   }
@@ -86,6 +84,8 @@ class OutgoingStreamWindowHandler extends AbstractOutgoingWindowHandler {
       _peerWindow.modify(difference);
       if (_peerWindow.size <= 0) {
         positiveWindow.markBuffered();
+      } else if (positiveWindow.wouldBuffer) {
+        positiveWindow.markUnBuffered();
       }
     }
   }

test/src/flowcontrol/window_handler_test.dart

@@ -83,12 +83,17 @@ main() {
       handler = new OutgoingStreamWindowHandler(window);
 
       expect(handler.positiveWindow.wouldBuffer, isFalse);
-      handler.positiveWindow.bufferEmptyEvents.listen(
+      final bufferEmpty = handler.positiveWindow.bufferEmptyEvents.listen(
           expectAsync1((_) {}, count: 0));
       handler.processInitialWindowSizeSettingChange(-window.size);
       expect(handler.positiveWindow.wouldBuffer, isTrue);
       expect(handler.peerWindowSize, 0);
       expect(window.size, 0);
+      bufferEmpty.onData(expectAsync1((_) {}, count: 1));
+      handler.processInitialWindowSizeSettingChange(1);
+      expect(handler.positiveWindow.wouldBuffer, isFalse);
+      expect(handler.peerWindowSize, 1);
+      expect(window.size, 1);
 
       expect(() => handler.processInitialWindowSizeSettingChange(
           Window.MAX_WINDOW_SIZE + 1), throwsA(isFlowControlException));