def test_main(self):
        @gen.coroutine
        def main_task():
            with self.tracer.start_active_span('parent'):
                tasks = self.submit_callbacks()
                yield tasks

        with tracer_stack_context():
            self.loop.add_callback(main_task)

        stop_loop_when(self.loop,
                       lambda: len(self.tracer.finished_spans()) == 4)
        self.loop.start()

        spans = self.tracer.finished_spans()
        self.assertEquals(len(spans), 4)
        self.assertNamesEqual(spans, ['task', 'task', 'task', 'parent'])

        for i in range(3):
            self.assertSameTrace(spans[i], spans[-1])
            self.assertIsChildOf(spans[i], spans[-1]) 

def test_kv_missing(self, loop, consul_port):
        c = consul.tornado.Consul(port=consul_port)

        @gen.coroutine
        def main():
            yield c.kv.put('index', 'bump')
            index, data = yield c.kv.get('foo')
            assert data is None
            index, data = yield c.kv.get('foo', index=index)
            assert data['Value'] == six.b('bar')
            loop.stop()

        @gen.coroutine
        def put():
            yield c.kv.put('foo', 'bar')

        loop.add_timeout(time.time() + (2.0 / 100), put)
        loop.run_sync(main) 

def test_kv_delete(self, loop, consul_port):
        @gen.coroutine
        def main():
            c = consul.tornado.Consul(port=consul_port)
            yield c.kv.put('foo1', '1')
            yield c.kv.put('foo2', '2')
            yield c.kv.put('foo3', '3')
            index, data = yield c.kv.get('foo', recurse=True)
            assert [x['Key'] for x in data] == ['foo1', 'foo2', 'foo3']

            response = yield c.kv.delete('foo2')
            assert response is True
            index, data = yield c.kv.get('foo', recurse=True)
            assert [x['Key'] for x in data] == ['foo1', 'foo3']
            response = yield c.kv.delete('foo', recurse=True)
            assert response is True
            index, data = yield c.kv.get('foo', recurse=True)
            assert data is None
            loop.stop()

        loop.run_sync(main) 

def test_agent_services(self, loop, consul_port):
        @gen.coroutine
        def main():
            c = consul.tornado.Consul(port=consul_port)
            services = yield c.agent.services()
            assert services == {}
            response = yield c.agent.service.register('foo')
            assert response is True
            services = yield c.agent.services()
            assert services == {'foo': {'ID': 'foo',
                                        'Service': 'foo',
                                        'Tags': [],
                                        'Meta': {},
                                        'Port': 0,
                                        'Address': '',
                                        'Weights': {'Passing': 1,
                                                    'Warning': 1},
                                        'EnableTagOverride': False}}
            response = yield c.agent.service.deregister('foo')
            assert response is True
            services = yield c.agent.services()
            assert services == {}
            loop.stop()

        loop.run_sync(main) 

def test_root(self, loop, local_server):
        c = consul.tornado.Consul(port=local_server.port)

        @gen.coroutine
        def test_timeout():
            time_out = False
            yield sleep(loop, 20 / 1000.0)

            try:
                yield c.agent.services()
            except consul.Timeout:
                time_out = True
            assert time_out
            loop.stop()

        loop.run_sync(test_timeout) 

def get_links_from_url(url):
    """Download the page at `url` and parse it for links.

    Returned links have had the fragment after `#` removed, and have been made
    absolute so, e.g. the URL 'gen.html#tornado.gen.coroutine' becomes
    'http://www.tornadoweb.org/en/stable/gen.html'.
    """
    try:
        response = yield httpclient.AsyncHTTPClient().fetch(url)
        print('fetched %s' % url)

        html = response.body if isinstance(response.body, str) \
            else response.body.decode()
        urls = [urljoin(url, remove_fragment(new_url))
                for new_url in get_links(html)]
    except Exception as e:
        print('Exception: %s %s' % (e, url))
        raise gen.Return([])

    raise gen.Return(urls) 

def wait_for_messages(self, cursor=None):
        # Construct a Future to return to our caller.  This allows
        # wait_for_messages to be yielded from a coroutine even though
        # it is not a coroutine itself.  We will set the result of the
        # Future when results are available.
        result_future = Future()
        if cursor:
            new_count = 0
            for msg in reversed(self.cache):
                if msg["id"] == cursor:
                    break
                new_count += 1
            if new_count:
                result_future.set_result(self.cache[-new_count:])
                return result_future
        self.waiters.add(result_future)
        return result_future 

def authenticate_redirect(self, callback_uri=None, callback=None):
        """Just like `~OAuthMixin.authorize_redirect`, but
        auto-redirects if authorized.

        This is generally the right interface to use if you are using
        Twitter for single-sign on.

        .. versionchanged:: 3.1
           Now returns a `.Future` and takes an optional callback, for
           compatibility with `.gen.coroutine`.
        """
        http = self.get_auth_http_client()
        http.fetch(self._oauth_request_token_url(callback_uri=callback_uri),
                   functools.partial(
                       self._on_request_token, self._OAUTH_AUTHENTICATE_URL,
                       None, callback)) 

def release(self):
        """增加counter 并且唤醒一个waiter."""
        self._value += 1
        while self._waiters:
            waiter = self._waiters.popleft()
            if not waiter.done():
                self._value -= 1

                # If the waiter is a coroutine paused at
                #
                #     with (yield semaphore.acquire()):
                #
                # then the context manager's __exit__ calls release() at the end
                # of the "with" block.
                waiter.set_result(_ReleasingContextManager(self))
                break 

def test_async_await_mixed_multi_native_yieldpoint(self):
        namespace = exec_test(globals(), locals(), """
        async def f1():
            await gen.Task(self.io_loop.add_callback)
            return 42
        """)

        @gen.coroutine
        def f2():
            yield gen.Task(self.io_loop.add_callback)
            raise gen.Return(43)

        f2(callback=(yield gen.Callback('cb')))
        results = yield [namespace['f1'](), gen.Wait('cb')]
        self.assertEqual(results, [42, 43])
        self.finished = True 

def test_replace_yieldpoint_exception(self):
        # Test exception handling: a coroutine can catch one exception
        # raised by a yield point and raise a different one.
        @gen.coroutine
        def f1():
            1 / 0

        @gen.coroutine
        def f2():
            try:
                yield f1()
            except ZeroDivisionError:
                raise KeyError()

        future = f2()
        with self.assertRaises(KeyError):
            yield future
        self.finished = True 

def test_swallow_yieldpoint_exception(self):
        # Test exception handling: a coroutine can catch an exception
        # raised by a yield point and not raise a different one.
        @gen.coroutine
        def f1():
            1 / 0

        @gen.coroutine
        def f2():
            try:
                yield f1()
            except ZeroDivisionError:
                raise gen.Return(42)

        result = yield f2()
        self.assertEqual(result, 42)
        self.finished = True 

def test_replace_context_exception(self):
        # Test exception handling: exceptions thrown into the stack context
        # can be caught and replaced.
        # Note that this test and the following are for behavior that is
        # not really supported any more:  coroutines no longer create a
        # stack context automatically; but one is created after the first
        # YieldPoint (i.e. not a Future).
        @gen.coroutine
        def f2():
            (yield gen.Callback(1))()
            yield gen.Wait(1)
            self.io_loop.add_callback(lambda: 1 / 0)
            try:
                yield gen.Task(self.io_loop.add_timeout,
                               self.io_loop.time() + 10)
            except ZeroDivisionError:
                raise KeyError()

        future = f2()
        with self.assertRaises(KeyError):
            yield future
        self.finished = True 

def test_swallow_context_exception(self):
        # Test exception handling: exceptions thrown into the stack context
        # can be caught and ignored.
        @gen.coroutine
        def f2():
            (yield gen.Callback(1))()
            yield gen.Wait(1)
            self.io_loop.add_callback(lambda: 1 / 0)
            try:
                yield gen.Task(self.io_loop.add_timeout,
                               self.io_loop.time() + 10)
            except ZeroDivisionError:
                raise gen.Return(42)

        result = yield f2()
        self.assertEqual(result, 42)
        self.finished = True 

def test_py3_leak_exception_context(self):
        class LeakedException(Exception):
            pass

        @gen.coroutine
        def inner(iteration):
            raise LeakedException(iteration)

        try:
            yield inner(1)
        except LeakedException as e:
            self.assertEqual(str(e), "1")
            self.assertIsNone(e.__context__)

        try:
            yield inner(2)
        except LeakedException as e:
            self.assertEqual(str(e), "2")
            self.assertIsNone(e.__context__)

        self.finished = True 

def test_streaming_until_close_future(self):
        server, client = self.make_iostream_pair()
        try:
            chunks = []

            @gen.coroutine
            def client_task():
                yield client.read_until_close(streaming_callback=chunks.append)

            @gen.coroutine
            def server_task():
                yield server.write(b"1234")
                yield gen.sleep(0.01)
                yield server.write(b"5678")
                server.close()

            @gen.coroutine
            def f():
                yield [client_task(), server_task()]
            self.io_loop.run_sync(f)
            self.assertEqual(chunks, [b"1234", b"5678"])
        finally:
            server.close()
            client.close() 

def test_context_manager_contended(self):
        sem = locks.Semaphore()
        history = []

        @gen.coroutine
        def f(index):
            with (yield sem.acquire()):
                history.append('acquired %d' % index)
                yield gen.sleep(0.01)
                history.append('release %d' % index)

        yield [f(i) for i in range(2)]

        expected_history = []
        for i in range(2):
            expected_history.extend(['acquired %d' % i, 'release %d' % i])

        self.assertEqual(expected_history, history) 

def test_acquire_fifo(self):
        lock = locks.Lock()
        self.assertTrue(lock.acquire().done())
        N = 5
        history = []

        @gen.coroutine
        def f(idx):
            with (yield lock.acquire()):
                history.append(idx)

        futures = [f(i) for i in range(N)]
        self.assertFalse(any(future.done() for future in futures))
        lock.release()
        yield futures
        self.assertEqual(list(range(N)), history) 

def test_run_with_stack_context(self):
        @gen.coroutine
        def f1():
            self.assertEqual(self.active_contexts, ['c1'])
            yield run_with_stack_context(
                StackContext(functools.partial(self.context, 'c2')),
                f2)
            self.assertEqual(self.active_contexts, ['c1'])

        @gen.coroutine
        def f2():
            self.assertEqual(self.active_contexts, ['c1', 'c2'])
            yield gen.Task(self.io_loop.add_callback)
            self.assertEqual(self.active_contexts, ['c1', 'c2'])

        self.assertEqual(self.active_contexts, [])
        yield run_with_stack_context(
            StackContext(functools.partial(self.context, 'c1')),
            f1)
        self.assertEqual(self.active_contexts, []) 

def test_task_done(self):
        q = self.queue_class()
        for i in range(100):
            q.put_nowait(i)

        self.accumulator = 0

        @gen.coroutine
        def worker():
            while True:
                item = yield q.get()
                self.accumulator += item
                q.task_done()
                yield gen.sleep(random() * 0.01)

        # Two coroutines share work.
        worker()
        worker()
        yield q.join()
        self.assertEqual(sum(range(100)), self.accumulator) 

def test_producer_consumer(self):
        q = queues.Queue(maxsize=3)
        history = []

        # We don't yield between get() and task_done(), so get() must wait for
        # the next tick. Otherwise we'd immediately call task_done and unblock
        # join() before q.put() resumes, and we'd only process the first four
        # items.
        @gen.coroutine
        def consumer():
            while True:
                history.append((yield q.get()))
                q.task_done()

        @gen.coroutine
        def producer():
            for item in range(10):
                yield q.put(item)

        consumer()
        yield producer()
        yield q.join()
        self.assertEqual(list(range(10)), history) 

def twisted_coroutine_fetch(self, url, runner):
        body = [None]

        @gen.coroutine
        def f():
            # This is simpler than the non-coroutine version, but it cheats
            # by reading the body in one blob instead of streaming it with
            # a Protocol.
            client = Agent(self.reactor)
            response = yield client.request(b'GET', utf8(url))
            with warnings.catch_warnings():
                # readBody has a buggy DeprecationWarning in Twisted 15.0:
                # https://twistedmatrix.com/trac/changeset/43379
                warnings.simplefilter('ignore', category=DeprecationWarning)
                body[0] = yield readBody(response)
            self.stop_loop()
        self.io_loop.add_callback(f)
        runner()
        return body[0] 

def authenticate_redirect(self, callback_uri=None, callback=None):
        """Just like `~OAuthMixin.authorize_redirect`, but
        auto-redirects if authorized.

        This is generally the right interface to use if you are using
        Twitter for single-sign on.

        .. versionchanged:: 3.1
           Now returns a `.Future` and takes an optional callback, for
           compatibility with `.gen.coroutine`.
        """
        http = self.get_auth_http_client()
        http.fetch(self._oauth_request_token_url(callback_uri=callback_uri),
                   functools.partial(
                       self._on_request_token, self._OAUTH_AUTHENTICATE_URL,
                       None, callback)) 

def release(self):
        """增加counter 并且唤醒一个waiter."""
        self._value += 1
        while self._waiters:
            waiter = self._waiters.popleft()
            if not waiter.done():
                self._value -= 1

                # If the waiter is a coroutine paused at
                #
                #     with (yield semaphore.acquire()):
                #
                # then the context manager's __exit__ calls release() at the end
                # of the "with" block.
                waiter.set_result(_ReleasingContextManager(self))
                break 

def send_task(self, message):
        request_context = {}

        @gen.coroutine
        def before_handler():
            self.request_handler.before_request(message, request_context)

        @gen.coroutine
        def after_handler():
            self.request_handler.after_request(message, request_context)

        yield before_handler()
        yield after_handler()

        raise gen.Return('%s::response' % message) 

def submit(self):
        span = self.tracer.scope_manager.active.span

        @gen.coroutine
        def task1():
            self.assertEqual(span, self.tracer.scope_manager.active.span)
            span.set_tag('key1', '1')

            @gen.coroutine
            def task2():
                self.assertEqual(span,
                                 self.tracer.scope_manager.active.span)
                span.set_tag('key2', '2')

                @gen.coroutine
                def task3():
                    self.assertEqual(span,
                                     self.tracer.scope_manager.active.span)
                    span.set_tag('key3', '3')
                    span.finish()

                yield task3()

            yield task2()

        yield task1() 

def submit_subtasks(self, parent_span):
        @gen.coroutine
        def task(name):
            logger.info('Running %s' % name)
            with self.tracer.scope_manager.activate(parent_span, False):
                with self.tracer.start_active_span(name):
                    gen.sleep(0.1)

        self.loop.add_callback(task, 'task1')
        self.loop.add_callback(task, 'task2') 

def is_tornado_coroutine(func):
    """
    Return whether *func* is a Tornado coroutine function.
    Running coroutines are not supported.
    """
    if 'tornado.gen' not in sys.modules:
        return False
    gen = sys.modules['tornado.gen']
    if not hasattr(gen, "is_coroutine_function"):
        # Tornado version is too old
        return False
    return gen.is_coroutine_function(func) 

def _rebuild_tornado_coroutine(func):
    from tornado import gen
    return gen.coroutine(func)


# Shorthands for legacy support 

def test_kv(self, loop, consul_port):
        @gen.coroutine
        def main():
            c = consul.tornado.Consul(port=consul_port)
            index, data = yield c.kv.get('foo')
            assert data is None
            response = yield c.kv.put('foo', 'bar')
            assert response is True
            index, data = yield c.kv.get('foo')
            assert data['Value'] == six.b('bar')
            loop.stop()

        loop.run_sync(main)