test_api_end_to_end.py 文件源码

def setUp(self):
    market_factory = WebSocketServerFactory('ws://localhost:8080/market_stream')
    user_factory = WebSocketServerFactory('ws://localhost:8080/user_stream')
    market_factory.protocol = MarketStreamServerProtocol
    user_factory.protocol = UserStreamServerProtocol
    market_factory.startFactory()
    user_factory.startFactory()
    root = Data('', 'text/plain')
    root.putChild(b'market_stream', WebSocketResource(market_factory))
    root.putChild(b'user_stream', WebSocketResource(user_factory))
    site = Site(root)
    reactor.listenTCP(8080, site)
    def run_server():
      reactor.run(installSignalHandlers=False)
    Thread(target=run_server).start()