Use Window Manager objects in streams.

Author: Lukasa

hyper/http20/connection.py

@@ -348,12 +348,11 @@ def _new_stream(self):
         """
         Returns a new stream object for this connection.
         """
+        window_size = self._settings[SettingsFrame.INITIAL_WINDOW_SIZE]
         s = Stream(
             self.next_stream_id, self._send_cb, self._recv_cb,
-            self._close_stream, self.encoder, self.decoder
-        )
-        s._out_flow_control_window = (
-            self._settings[SettingsFrame.INITIAL_WINDOW_SIZE]
+            self._close_stream, self.encoder, self.decoder,
+            self.__wm_class(window_size)
         )
         self.next_stream_id += 2
 

hyper/http20/stream.py

@@ -50,16 +50,17 @@ def __init__(self,
                  recv_cb,
                  close_cb,
                  header_encoder,
-                 header_decoder):
+                 header_decoder,
+                 window_manager):
         self.stream_id = stream_id
         self.state = STATE_IDLE
         self.headers = []
         self._queued_frames = collections.deque()
 
         # There are two flow control windows: one for data we're sending,
         # one for data being sent to us.
+        self._in_window_manager = window_manager
         self._out_flow_control_window = 65535
-        self._in_flow_control_window = 65535
 
         # This is the callback handed to the stream by its parent connection.
         # It is called when the stream wants to send data. It expects to
@@ -148,9 +149,10 @@ def listlen(list):
 
             # Increase the window size. Only do this if the data frame contains
             # actual data.
-            if len(frame.data):
+            increment = self._in_window_manager._handle_frame(len(frame.data))
+            if increment:
                 w = WindowUpdateFrame(self.stream_id)
-                w.window_increment = len(frame.data)
+                w.window_increment = increment
                 self._data_cb(w)
 
         return b''.join(data)

test/test_hyper.py

@@ -13,6 +13,7 @@
 )
 from hyper.http20.response import HTTP20Response
 from hyper.http20.exceptions import HPACKDecodingError
+from hyper.http20.window import FlowControlManager
 from hyper.contrib import HTTP20Adapter
 import pytest
 import zlib
@@ -973,15 +974,15 @@ def test_connections_handle_resizing_header_tables_properly(self):
 
 class TestHyperStream(object):
     def test_streams_have_ids(self):
-        s = Stream(1, None, None, None, None, None)
+        s = Stream(1, None, None, None, None, None, None)
         assert s.stream_id == 1
 
     def test_streams_initially_have_no_headers(self):
-        s = Stream(1, None, None, None, None, None)
+        s = Stream(1, None, None, None, None, None, None)
         assert s.headers == []
 
     def test_streams_can_have_headers(self):
-        s = Stream(1, None, None, None, None, None)
+        s = Stream(1, None, None, None, None, None, None)
         s.add_header("name", "value")
         assert s.headers == [("name", "value")]
 
@@ -991,14 +992,14 @@ def data_callback(frame):
             assert frame.data == 'testkeyTestVal'
             assert frame.flags == set(['END_STREAM', 'END_HEADERS'])
 
-        s = Stream(1, data_callback, None, None, NullEncoder, None)
+        s = Stream(1, data_callback, None, None, NullEncoder, None, None)
         s.add_header("TestKey", "TestVal")
         s.open(True)
 
         assert s.state == STATE_HALF_CLOSED_LOCAL
 
     def test_receiving_a_frame_queues_it(self):
-        s = Stream(1, None, None, None, None, None)
+        s = Stream(1, None, None, None, None, None, None)
         s.receive_frame(Frame(0))
         assert len(s._queued_frames) == 1
 
@@ -1008,7 +1009,7 @@ def data_callback(frame):
             assert frame.data == b'Hi there!'
             assert frame.flags == set(['END_STREAM'])
 
-        s = Stream(1, data_callback, None, None, NullEncoder, None)
+        s = Stream(1, data_callback, None, None, NullEncoder, None, None)
         s.state = STATE_OPEN
         s.send_data(BytesIO(b'Hi there!'), True)
 
@@ -1027,7 +1028,7 @@ def data_callback(frame):
 
         data = b'test' * (MAX_CHUNK + 1)
 
-        s = Stream(1, data_callback, None, None, NullEncoder, None)
+        s = Stream(1, data_callback, None, None, NullEncoder, None, None)
         s.state = STATE_OPEN
         s.send_data(BytesIO(data), True)
 
@@ -1042,7 +1043,7 @@ def data_callback(frame):
             assert frame.data == b'Hi there!'
             assert frame.flags == set(['END_STREAM'])
 
-        s = Stream(1, data_callback, None, None, NullEncoder, None)
+        s = Stream(1, data_callback, None, None, NullEncoder, None, None)
         s.state = STATE_OPEN
         s.send_data(b'Hi there!', True)
 
@@ -1061,7 +1062,7 @@ def data_callback(frame):
 
         data = b'test' * (MAX_CHUNK + 1)
 
-        s = Stream(1, data_callback, None, None, NullEncoder, None)
+        s = Stream(1, data_callback, None, None, NullEncoder, None, None)
         s.state = STATE_OPEN
         s.send_data(data, True)
 
@@ -1071,7 +1072,7 @@ def data_callback(frame):
         assert s._out_flow_control_window == 65535 - len(data)
 
     def test_windowupdate_frames_update_windows(self):
-        s = Stream(1, None, None, None, None, None)
+        s = Stream(1, None, None, None, None, None, None)
         f = WindowUpdateFrame(1)
         f.window_increment = 1000
         s.receive_frame(f)
@@ -1090,7 +1091,7 @@ def inner():
                 s.receive_frame(in_frames.pop(0))
             return inner
 
-        s = Stream(1, send_cb, None, None, None, None)
+        s = Stream(1, send_cb, None, None, None, None, None)
         s._recv_cb = recv_cb(s)
         s.state = STATE_HALF_CLOSED_LOCAL
 
@@ -1116,7 +1117,7 @@ def inner():
                 s.receive_frame(in_frames.pop(0))
             return inner
 
-        s = Stream(1, send_cb, None, None, None, None)
+        s = Stream(1, send_cb, None, None, None, None, FlowControlManager(65535))
         s._recv_cb = recv_cb(s)
         s.state = STATE_HALF_CLOSED_LOCAL
 
@@ -1148,7 +1149,7 @@ def inner():
                 s.receive_frame(in_frames.pop(0))
             return inner
 
-        s = Stream(1, send_cb, None, None, None, None)
+        s = Stream(1, send_cb, None, None, None, None, FlowControlManager(65535))
         s._recv_cb = recv_cb(s)
         s.state = STATE_HALF_CLOSED_LOCAL