def test_flow_control(self):
        # Write more messages than the protocol is willing to pipeline. Flow
        # control should kick in and alternate scheduling of the producer and
        # the consumer.
        proto, trans = self.protocol, self.transport
        self.assertTrue(trans._reading)
        proto.max_pipeline_size = 10
        message = b'{ "id": 1, "method": "foo", "params": [] }'
        interrupted = 0
        for i in range(1000):
            proto.data_received(message)
            if not trans._reading:
                interrupted += 1
                gruvi.sleep(0)  # run dispatcher
            self.assertTrue(trans._reading)
        mm = self.get_messages()
        self.assertEqual(len(mm), 1000)
        self.assertEqual(interrupted, 100)
        message = json.loads(message.decode('utf8'))
        for m in mm:
            self.assertEqual(m, message)