def testPatronPipelineStream(self):
        """
        Test Patron pipeline stream
        """
        console.terse("{0}\n".format(self.testPatronPipelineStream.__doc__))



        store = storing.Store(stamp=0.0)

        wireLogAlpha = wiring.WireLog(buffify=True, same=True)
        result = wireLogAlpha.reopen()

        alpha = tcp.Server(port = 6101,
                                   bufsize=131072,
                                   wlog=wireLogAlpha,
                                   store=store)
        self.assertIs(alpha.reopen(), True)
        self.assertEqual(alpha.ha, ('0.0.0.0', 6101))
        self.assertEqual(alpha.eha, ('127.0.0.1', 6101))

        console.terse("{0}\n".format("Building Connector ...\n"))

        wireLogBeta = wiring.WireLog(buffify=True,  same=True)
        result = wireLogBeta.reopen()
        host = alpha.eha[0]
        port = alpha.eha[1]


        beta = clienting.Patron(bufsize=131072,
                                 wlog=wireLogBeta,
                                 hostname=host,
                                 port=port,
                                 store=store,
                                 reconnectable=True,
                                 )

        self.assertIs(beta.connector.reopen(), True)
        self.assertIs(beta.connector.accepted, False)
        self.assertIs(beta.connector.connected, False)
        self.assertIs(beta.connector.cutoff, False)

        console.terse("Connecting beta to server ...\n")
        while True:
            beta.serviceAll()
            alpha.serviceConnects()
            if beta.connector.connected and alpha.ixes in beta.connector.ca:
                break
            time.sleep(0.05)
            beta.connector.store.advanceStamp(0.05)

        self.assertIs(beta.connector.accepted, True)
        self.assertIs(beta.connector.connected, True)
        self.assertIs(beta.connector.cutoff, False)
        self.assertEqual(beta.connector.ca, beta.connector.cs.getsockname())
        self.assertEqual(beta.connector.ha, beta.connector.cs.getpeername())
        self.assertEqual(alpha.eha, beta.connector.ha)

        ixBeta = alpha.ixes[beta.connector.ca]
        self.assertIsNotNone(ixBeta.ca)
        self.assertIsNotNone(ixBeta.cs)
        self.assertEqual(ixBeta.cs.getsockname(), beta.connector.cs.getpeername())
        self.assertEqual(ixBeta.cs.getpeername(), beta.connector.cs.getsockname())
        self.assertEqual(ixBeta.ca, beta.connector.ca)
        self.assertEqual(ixBeta.ha, beta.connector.ha)

        console.terse("{0}\n".format("Building Request ...\n"))
        request = odict([('method', u'GET'),
                         ('path', u'/stream'),
                         ('qargs', odict()),
                         ('fragment', u''),
                         ('headers', odict([('Accept', 'application/json')])),
                         ('body', None),
                        ])

        beta.requests.append(request)

        console.terse("Beta requests to Alpha\n")
        console.terse("from {0}:{1}, {2} {3} ...\n".format(beta.connector.ha[0],
                                                         beta.connector.ha[1],
                                                         request['method'],
                                                         request['path']))

        while (beta.requests or beta.connector.txes) and not ixBeta.rxbs:
            beta.serviceAll()
            time.sleep(0.05)
            beta.connector.store.advanceStamp(0.05)
            alpha.serviceReceivesAllIx()
            time.sleep(0.05)
            beta.connector.store.advanceStamp(0.05)

        msgIn = bytes(ixBeta.rxbs)
        msgOut = b'GET /stream HTTP/1.1\r\nHost: 127.0.0.1:6101\r\nAccept-Encoding: identity\r\nAccept: application/json\r\n\r\n'

        self.assertEqual(msgIn, msgOut)
        ixBeta.clearRxbs()

        console.terse("Alpha responds to Beta\n")
        lines = [
            b'HTTP/1.0 200 OK\r\n',
            b'Server: PasteWSGIServer/0.5 Python/2.7.9\r\n',
            b'Date: Thu, 30 Apr 2015 21:35:25 GMT\r\n'
            b'Content-Type: text/event-stream\r\n',
            b'Cache-Control: no-cache\r\n',
            b'Connection: close\r\n\r\n',
            b'retry: 1000\n\n',
            b'id: 0\ndata: START\n\n',
            b'id: 1\ndata: 1\ndata: 2\n\n',
            b'id: 2\ndata: 3\ndata: 4\n\n',
            b'id: 3\ndata: 5\ndata: 6\n\n',
            b'id: 4\ndata: 7\ndata: 8\n\n',
        ]

        msgOut = b''.join(lines)
        ixBeta.tx(msgOut)
        timer = StoreTimer(store=store, duration=0.5)
        while ixBeta.txes or not timer.expired:
            alpha.serviceTxesAllIx()
            time.sleep(0.05)
            beta.connector.store.advanceStamp(0.05)
            beta.serviceAll()
            time.sleep(0.05)
            beta.connector.store.advanceStamp(0.05)

        self.assertEqual(len(beta.connector.rxbs), 0)

        #timed out while stream still open so no responses in .responses
        self.assertIs(beta.waited, True)
        self.assertIs(beta.respondent.ended, False)
        self.assertEqual(len(beta.responses), 0)

        # but are events in .events
        self.assertEqual(len(beta.events), 5)
        self.assertEqual(beta.respondent.retry, 1000)
        self.assertEqual(beta.respondent.leid, '4')
        event = beta.events.popleft()
        self.assertEqual(event, {'id': '0', 'name': '', 'data': 'START'})
        event = beta.events.popleft()
        self.assertEqual(event, {'id': '1', 'name': '', 'data': '1\n2'})
        event = beta.events.popleft()
        self.assertEqual(event, {'id': '2', 'name': '', 'data': '3\n4'})
        beta.events.clear()

        # alpha's ixBeta connection shutdown prematurely
        console.terse("Disconnecting server so beta must auto reconnect ...\n")
        alpha.closeIx(beta.connector.ca)
        alpha.removeIx(beta.connector.ca)
        while True:
            beta.serviceAll()
            if not beta.connector.connected:
                break
            time.sleep(0.1)
            beta.connector.store.advanceStamp(0.1)

        self.assertIs(beta.connector.cutoff, False)

        console.terse("Auto reconnecting beta and rerequesting...\n")
        while True:
            beta.serviceAll()
            alpha.serviceConnects()
            if beta.connector.connected and beta.connector.ca in alpha.ixes:
                break
            time.sleep(0.05)
            beta.connector.store.advanceStamp(0.05)

        self.assertIs(beta.connector.accepted, True)
        self.assertIs(beta.connector.connected, True)
        self.assertIs(beta.connector.cutoff, False)
        self.assertEqual(beta.connector.ca, beta.connector.cs.getsockname())
        self.assertEqual(beta.connector.ha, beta.connector.cs.getpeername())
        self.assertEqual(alpha.eha, beta.connector.ha)

        ixBeta = alpha.ixes[beta.connector.ca]
        self.assertIsNotNone(ixBeta.ca)
        self.assertIsNotNone(ixBeta.cs)
        self.assertEqual(ixBeta.cs.getsockname(), beta.connector.cs.getpeername())
        self.assertEqual(ixBeta.cs.getpeername(), beta.connector.cs.getsockname())
        self.assertEqual(ixBeta.ca, beta.connector.ca)
        self.assertEqual(ixBeta.ha, beta.connector.ha)

        console.terse("Server receiving...\n")
        while (beta.requests or beta.connector.txes) or not ixBeta.rxbs:
            beta.serviceAll()
            time.sleep(0.05)
            beta.connector.store.advanceStamp(0.05)
            alpha.serviceReceivesAllIx()
            time.sleep(0.05)
            beta.connector.store.advanceStamp(0.05)

        msgIn = bytes(ixBeta.rxbs)
        msgOut = b'GET /stream HTTP/1.1\r\nHost: 127.0.0.1:6101\r\nAccept-Encoding: identity\r\nAccept: application/json\r\nLast-Event-Id: 4\r\n\r\n'

        self.assertEqual(msgIn, msgOut)
        ixBeta.clearRxbs()

        console.terse("Alpha responds to Beta\n")
        lines = [
            b'HTTP/1.0 200 OK\r\n',
            b'Server: PasteWSGIServer/0.5 Python/2.7.9\r\n',
            b'Date: Thu, 30 Apr 2015 21:35:25 GMT\r\n'
            b'Content-Type: text/event-stream\r\n',
            b'Cache-Control: no-cache\r\n',
            b'Connection: close\r\n\r\n',
            b'id: 5\ndata: 9\ndata: 10\n\n',
            b'id: 6\ndata: 11\ndata: 12\n\n',
        ]

        msgOut = b''.join(lines)
        ixBeta.tx(msgOut)
        timer = StoreTimer(store=store, duration=0.5)
        while ixBeta.txes or not timer.expired:
            alpha.serviceTxesAllIx()
            time.sleep(0.05)
            beta.connector.store.advanceStamp(0.05)
            beta.serviceAll()
            time.sleep(0.05)
            beta.connector.store.advanceStamp(0.05)

        self.assertEqual(len(beta.connector.rxbs), 0)

        #timed out while stream still open so no responses in .responses
        self.assertIs(beta.waited, True)
        self.assertIs(beta.respondent.ended, False)
        self.assertEqual(len(beta.responses), 0)

        # but are events in .events
        self.assertEqual(len(beta.events), 2)
        self.assertEqual(beta.respondent.retry, 1000)
        self.assertEqual(beta.respondent.leid, '6')
        event = beta.events.popleft()
        self.assertEqual(event, {'id': '5', 'name': '', 'data': '9\n10'})
        event = beta.events.popleft()
        self.assertEqual(event, {'id': '6', 'name': '', 'data': '11\n12'})


        alpha.close()
        beta.connector.close()

        wireLogAlpha.close()
        wireLogBeta.close()
dataset/ETHPy150Open ioflo/ioflo/ioflo/aio/http/test/test_clienting.py/BasicTestCase.testPatronPipelineStream/swapped operands of `in`