test_tcp.py 文件源码

def _testBuildProtocol(self, portno):
        f = AServerFactory(self, IPv4Address('TCP', '127.0.0.1', portno))
        wrappedF = FireOnListenFactory(f)
        p = reactor.listenTCP(0, wrappedF)
        self.ports.append(p)

        def client(ignored):
            acf = AClientFactory(self, IPv4Address("TCP", "127.0.0.1",
                                                   p.getHost().port))
            wired = WiredFactory(acf)
            reactor.connectTCP("127.0.0.1", p.getHost().port, wired,
                               bindAddress=("127.0.0.1", portno))
            d = wired.onConnect
            def _onConnect(ignored):
                self.ports.append(acf.protocol.transport)
                self.assert_(hasattr(self, "ran"))
                return wired.onDisconnect
            def _onDisconnect(ignored):
                del self.ran
            d.addCallback(_onConnect)
            d.addCallback(_onDisconnect)
            return d

        return wrappedF.deferred.addCallback(client)