class ReactorCallFromThreadTest(unittest.TestCase):
    def setUp(self):
        self._reactor = TornadoReactor(IOLoop())
        self._mainThread = thread.get_ident()

    def tearDown(self):
        self._thread.join()

    def _newThreadRun(self):
        self.assertNotEqual(self._mainThread, thread.get_ident())
        if hasattr(self._thread, 'ident'):  # new in python 2.6
            self.assertEqual(self._thread.ident, thread.get_ident())
        self._reactor.callFromThread(self._fnCalledFromThread)

    def _fnCalledFromThread(self):
        self.assertEqual(self._mainThread, thread.get_ident())
        self._reactor.stop()

    def _whenRunningCallback(self):
        self._thread = threading.Thread(target=self._newThreadRun)
        self._thread.start()

    def testCallFromThread(self):
        self._reactor.callWhenRunning(self._whenRunningCallback)
        self._reactor.run()

 def setUp(self):
     self._reactor = TornadoReactor(IOLoop())
     r, w = os.pipe()
     self._set_nonblocking(r)
     self._set_nonblocking(w)
     set_close_exec(r)
     set_close_exec(w)
     self._p1 = os.fdopen(r, "rb", 0)
     self._p2 = os.fdopen(w, "wb", 0)

class ReactorCallInThread(unittest.TestCase):
    def setUp(self):
        self._reactor = TornadoReactor(IOLoop())
        self._mainThread = thread.get_ident()

    def _fnCalledInThread(self, *args, **kwargs):
        self.assertNotEqual(thread.get_ident(), self._mainThread)
        self._reactor.callFromThread(lambda: self._reactor.stop())

    def _whenRunningCallback(self):
        self._reactor.callInThread(self._fnCalledInThread)

    def testCallInThread(self):
        self._reactor.callWhenRunning(self._whenRunningCallback)
        self._reactor.run()

    def on_start(self):
        ''' start up the WAMP frontend by creating a session factory and 
            running a local IOLoop'''
        logger.info("WAMPFrontend started successfully")

        if self.config["wampfrontend"]["enable_twisted_log"]:
            log.startLogging(sys.stdout)

        self._loop = IOLoop.instance()
        self._reactor = TornadoReactor(self._loop)
        self.schedule( self.client_connect )

class ReactorCallLaterTest(unittest.TestCase):
    def setUp(self):
        self._reactor = TornadoReactor(IOLoop())

    def test_callLater(self):
        self._laterCalled = False
        self._now = self._reactor.seconds()
        self._timeout = 0.001
        dc = self._reactor.callLater(self._timeout, self.callLaterCallback)
        self.assertEqual(self._reactor.getDelayedCalls(), [dc])
        self._reactor.run()
        self.assertTrue(self._laterCalled)
        self.assertTrue(self._called - self._now > self._timeout)
        self.assertEqual(self._reactor.getDelayedCalls(), [])

    def callLaterCallback(self):
        self._laterCalled = True
        self._called = self._reactor.seconds()
        self._reactor.stop()

class ReactorWhenRunningTest(unittest.TestCase):
    def setUp(self):
        self._reactor = TornadoReactor(IOLoop())

    def test_whenRunning(self):
        self._whenRunningCalled = False
        self._anotherWhenRunningCalled = False
        self._reactor.callWhenRunning(self.whenRunningCallback)
        self._reactor.run()
        self.assertTrue(self._whenRunningCalled)
        self.assertTrue(self._anotherWhenRunningCalled)

    def whenRunningCallback(self):
        self._whenRunningCalled = True
        self._reactor.callWhenRunning(self.anotherWhenRunningCallback)
        self._reactor.stop()

    def anotherWhenRunningCallback(self):
        self._anotherWhenRunningCalled = True

 def setUp(self):
     self.saved_signals = save_signal_handlers()
     self.io_loop = IOLoop()
     self.reactor = TornadoReactor(self.io_loop)

class CompatibilityTests(unittest.TestCase):
    def setUp(self):
        self.saved_signals = save_signal_handlers()
        self.io_loop = IOLoop()
        self.reactor = TornadoReactor(self.io_loop)

    def tearDown(self):
        self.reactor.disconnectAll()
        self.io_loop.close(all_fds=True)
        restore_signal_handlers(self.saved_signals)

    def start_twisted_server(self):
        class HelloResource(Resource):
            isLeaf = True

            def render_GET(self, request):
                return "Hello from twisted!"
        site = Site(HelloResource())
        self.twisted_port = get_unused_port()
        self.reactor.listenTCP(self.twisted_port, site, interface='127.0.0.1')

    def start_tornado_server(self):
        class HelloHandler(RequestHandler):
            def get(self):
                self.write("Hello from tornado!")
        app = Application([('/', HelloHandler)],
                          log_function=lambda x: None)
        self.tornado_port = get_unused_port()
        app.listen(self.tornado_port, address='127.0.0.1', io_loop=self.io_loop)

    def run_ioloop(self):
        self.stop_loop = self.io_loop.stop
        self.io_loop.start()
        self.reactor.fireSystemEvent('shutdown')

    def run_reactor(self):
        self.stop_loop = self.reactor.stop
        self.stop = self.reactor.stop
        self.reactor.run()

    def tornado_fetch(self, url, runner):
        responses = []
        client = AsyncHTTPClient(self.io_loop)

        def callback(response):
            responses.append(response)
            self.stop_loop()
        client.fetch(url, callback=callback)
        runner()
        self.assertEqual(len(responses), 1)
        responses[0].rethrow()
        return responses[0]

    def twisted_fetch(self, url, runner):
        # http://twistedmatrix.com/documents/current/web/howto/client.html
        chunks = []
        client = Agent(self.reactor)
        d = client.request('GET', url)

        class Accumulator(Protocol):
            def __init__(self, finished):
                self.finished = finished

            def dataReceived(self, data):
                chunks.append(data)

            def connectionLost(self, reason):
                self.finished.callback(None)

        def callback(response):
            finished = Deferred()
            response.deliverBody(Accumulator(finished))
            return finished
        d.addCallback(callback)

        def shutdown(ignored):
            self.stop_loop()
        d.addBoth(shutdown)
        runner()
        self.assertTrue(chunks)
        return ''.join(chunks)

    def testTwistedServerTornadoClientIOLoop(self):
        self.start_twisted_server()
        response = self.tornado_fetch(
            'http://localhost:%d' % self.twisted_port, self.run_ioloop)
        self.assertEqual(response.body, 'Hello from twisted!')

    def testTwistedServerTornadoClientReactor(self):
        self.start_twisted_server()
        response = self.tornado_fetch(
            'http://localhost:%d' % self.twisted_port, self.run_reactor)
        self.assertEqual(response.body, 'Hello from twisted!')

    def testTornadoServerTwistedClientIOLoop(self):
        self.start_tornado_server()
        response = self.twisted_fetch(
            'http://localhost:%d' % self.tornado_port, self.run_ioloop)
        self.assertEqual(response, 'Hello from tornado!')

#.........这里部分代码省略.........

class CompatibilityTests(unittest.TestCase):
    def setUp(self):
        self.saved_signals = save_signal_handlers()
        self.io_loop = IOLoop()
        self.io_loop.make_current()
        self.reactor = TornadoReactor(self.io_loop)

    def tearDown(self):
        self.reactor.disconnectAll()
        self.io_loop.clear_current()
        self.io_loop.close(all_fds=True)
        restore_signal_handlers(self.saved_signals)

    def start_twisted_server(self):
        class HelloResource(Resource):
            isLeaf = True

            def render_GET(self, request):
                return "Hello from twisted!"

        site = Site(HelloResource())
        port = self.reactor.listenTCP(0, site, interface="127.0.0.1")
        self.twisted_port = port.getHost().port

    def start_tornado_server(self):
        class HelloHandler(RequestHandler):
            def get(self):
                self.write("Hello from tornado!")

        app = Application([("/", HelloHandler)], log_function=lambda x: None)
        server = HTTPServer(app, io_loop=self.io_loop)
        sock, self.tornado_port = bind_unused_port()
        server.add_sockets([sock])

    def run_ioloop(self):
        self.stop_loop = self.io_loop.stop
        self.io_loop.start()
        self.reactor.fireSystemEvent("shutdown")

    def run_reactor(self):
        self.stop_loop = self.reactor.stop
        self.stop = self.reactor.stop
        self.reactor.run()

    def tornado_fetch(self, url, runner):
        responses = []
        client = AsyncHTTPClient(self.io_loop)

        def callback(response):
            responses.append(response)
            self.stop_loop()

        client.fetch(url, callback=callback)
        runner()
        self.assertEqual(len(responses), 1)
        responses[0].rethrow()
        return responses[0]

    def twisted_fetch(self, url, runner):
        # http://twistedmatrix.com/documents/current/web/howto/client.html
        chunks = []
        client = Agent(self.reactor)
        d = client.request(b"GET", utf8(url))

        class Accumulator(Protocol):
            def __init__(self, finished):
                self.finished = finished

            def dataReceived(self, data):
                chunks.append(data)

            def connectionLost(self, reason):
                self.finished.callback(None)

        def callback(response):
            finished = Deferred()
            response.deliverBody(Accumulator(finished))
            return finished

        d.addCallback(callback)

        def shutdown(failure):
            if hasattr(self, "stop_loop"):
                self.stop_loop()
            elif failure is not None:
                # loop hasn't been initialized yet; try our best to
                # get an error message out. (the runner() interaction
                # should probably be refactored).
                try:
                    failure.raiseException()
                except:
                    logging.error("exception before starting loop", exc_info=True)

        d.addBoth(shutdown)
        runner()
        self.assertTrue(chunks)
        return "".join(chunks)

    def twisted_coroutine_fetch(self, url, runner):
        body = [None]
#.........这里部分代码省略.........

 def setUp(self):
     self.io_loop = IOLoop()
     self.reactor = TornadoReactor(self.io_loop)

class WAMPFrontend(pykka.ThreadingActor, core.CoreListener):
    ''' Web Application Messaging Protocol frontend for Mopidy.
        On startup an additional thread running a local Tornado 
        IOLoop will be created '''

    def __init__(self, config, core):
        ''' initialize the frontend 
            `config` is a dict holding Mopidy configuration values
            `core` is a Mopidy core actor proxy '''
        super(WAMPFrontend, self).__init__()
        self.core = core
        self.config = config
        self._loop = None
        self._reactor = None
        self._session = None
        self._ioloop_thread = None


    def on_start(self):
        ''' start up the WAMP frontend by creating a session factory and 
            running a local IOLoop'''
        logger.info("WAMPFrontend started successfully")

        if self.config["wampfrontend"]["enable_twisted_log"]:
            log.startLogging(sys.stdout)

        self._loop = IOLoop.instance()
        self._reactor = TornadoReactor(self._loop)
        self.schedule( self.client_connect )


    def on_stop(self):
        logger.info("WAMPFrontend stopping ...")


    def on_event(self, event, **kwargs):
        ''' Schedule an incoming mopidy event to be published '''
        if self._session._client:
             self.schedule( self._session._client.on_event, event, **kwargs )
             logger.debug("WAMPFrontend: scheduled event publishing for %s" % event)


    def schedule(self, function, *args, **kwargs):
        ''' Schedule a function to be called during the next IOLoop iteration '''
        assert self._loop, \
           'WAMPFrontend has not been started yet. Unable to run IOLoop'
        assert isinstance(self._loop, IOLoop), \
           'WAMPFrontend: Invalid local IO-loop type: %s' % type(self._loop)
        logger.debug('WAMPFrontend: scheduled call for IOLoop: %s' % function)
        self._loop.add_callback( function, *args, **kwargs )
        
        
    def client_connect(self):
        client = clientFromString(self._reactor, url_to_client_string(self.config['wampfrontend']["router"]))
        transport = self._prepare_transport()
        logger.info('WAMPFrontend: connecting client (from thread: %s)' % currentThread()) 
        d = client.connect(transport)
        def on_success(result):
            pass
        def on_error(result):
            logger.warn("WAMPFrontend: %s" % result)
            self._reactor.callLater(2, self.client_connect)

        d.addCallbacks(on_success, on_error) 

    def _prepare_transport(self):
        ''' Prepare a transport factory '''
        assert self._reactor, \
            'WAMPFrontend: cannot create transport without an IOLoop and a TornadoReactor'

        config = types.ComponentConfig(
            realm = self.config['wampfrontend']['realm'],
            extra = {
                'name': self.config['wampfrontend']['name'],
                'core': self.core,
                'frontend': self })
        session = wamp.ApplicationSessionFactory(config=config)
        session.session = WAMPFrontendComponent

        # Add a reference toe the frontend object (ourself)
        session._frontend = self
        
        # Add a reference the the ApplicationSession created
        # since we are a client, there will be only one session at all
        session._client = None

        # Now also store a reference to the session factory for us
        self._session = session
        
        transport = MyClientFactory(
            session,
            url = self.config['wampfrontend']['router'],
            debug = self.config['wampfrontend']['debug_autobahn'],
            debug_wamp = self.config['wampfrontend']['debug_wamp'])
        return transport

class ReactorTwoCallLaterTest(unittest.TestCase):
    def setUp(self):
        self._reactor = TornadoReactor(IOLoop())

    def test_callLater(self):
        self._later1Called = False
        self._later2Called = False
        self._now = self._reactor.seconds()
        self._timeout1 = 0.0005
        dc1 = self._reactor.callLater(self._timeout1, self.callLaterCallback1)
        self._timeout2 = 0.001
        dc2 = self._reactor.callLater(self._timeout2, self.callLaterCallback2)
        self.assertTrue(self._reactor.getDelayedCalls() == [dc1, dc2] or
                        self._reactor.getDelayedCalls() == [dc2, dc1])
        self._reactor.run()
        self.assertTrue(self._later1Called)
        self.assertTrue(self._later2Called)
        self.assertTrue(self._called1 - self._now > self._timeout1)
        self.assertTrue(self._called2 - self._now > self._timeout2)
        self.assertEqual(self._reactor.getDelayedCalls(), [])

    def callLaterCallback1(self):
        self._later1Called = True
        self._called1 = self._reactor.seconds()

    def callLaterCallback2(self):
        self._later2Called = True
        self._called2 = self._reactor.seconds()
        self._reactor.stop()

class ReactorReaderWriterTest(unittest.TestCase):
    def _set_nonblocking(self, fd):
        flags = fcntl.fcntl(fd, fcntl.F_GETFL)
        fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)

    def setUp(self):
        self._reactor = TornadoReactor(IOLoop())
        r, w = os.pipe()
        self._set_nonblocking(r)
        self._set_nonblocking(w)
        set_close_exec(r)
        set_close_exec(w)
        self._p1 = os.fdopen(r, "rb", 0)
        self._p2 = os.fdopen(w, "wb", 0)

    def _testReadWrite(self):
        """
        In this test the writer writes an 'x' to its fd. The reader
        reads it, check the value and ends the test.
        """
        self.shouldWrite = True
        def checkReadInput(fd):
            self.assertEquals(fd.read(), 'x')
            self._reactor.stop()
        def writeOnce(fd):
            if self.shouldWrite:
                self.shouldWrite = False
                fd.write('x')
        self._reader = Reader(self._p1, checkReadInput)
        self._writer = Writer(self._p2, writeOnce)

        self._reactor.addWriter(self._writer)

        # Test that adding the reader twice adds it only once to
        # IOLoop.
        self._reactor.addReader(self._reader)
        self._reactor.addReader(self._reader)

    def testReadWrite(self):
        self._reactor.callWhenRunning(self._testReadWrite)
        self._reactor.run()

    def _testNoWriter(self):
        """
        In this test we have no writer. Make sure the reader doesn't
        read anything.
        """
        def checkReadInput(fd):
            self.fail("Must not be called.")

        def stopTest():
            # Close the writer here since the IOLoop doesn't know
            # about it.
            self._writer.close()
            self._reactor.stop()
        self._reader = Reader(self._p1, checkReadInput)

        # We create a writer, but it should never be invoked.
        self._writer = Writer(self._p2, lambda fd: fd.write('x'))

        # Test that adding and removing the writer leaves us with no writer.
        self._reactor.addWriter(self._writer)
        self._reactor.removeWriter(self._writer)

        # Test that adding and removing the reader doesn't cause
        # unintended effects.
        self._reactor.addReader(self._reader)

        # Wake up after a moment and stop the test
        self._reactor.callLater(0.001, stopTest)

    def testNoWriter(self):
        self._reactor.callWhenRunning(self._testNoWriter)
        self._reactor.run()