def __init__(self, channel_service: 'ChannelService'):
        self._channel_service = channel_service
        self._block_manager = None
        self._blockchain = None

        self._thread_pool = ThreadPoolExecutor(1, "ChannelInnerThread")

        # Citizen
        CitizenInfo = namedtuple("CitizenInfo", "peer_id target connected_time")
        self._CitizenInfo = CitizenInfo
        self._citizens: Dict[str, CitizenInfo] = dict()
        self._citizen_condition_new_block: Condition = None
        self._citizen_condition_unregister: Condition = None

        self.__sub_processes = []
        self.__loop_for_sub_services = None 

def test_context_manager_async_with(self):
        primitives = [
            asyncio.Lock(loop=self.loop),
            asyncio.Condition(loop=self.loop),
            asyncio.Semaphore(loop=self.loop),
            asyncio.BoundedSemaphore(loop=self.loop),
        ]

        async def test(lock):
            await asyncio.sleep(0.01, loop=self.loop)
            self.assertFalse(lock.locked())
            async with lock as _lock:
                self.assertIs(_lock, None)
                self.assertTrue(lock.locked())
                await asyncio.sleep(0.01, loop=self.loop)
                self.assertTrue(lock.locked())
            self.assertFalse(lock.locked())

        for primitive in primitives:
            self.loop.run_until_complete(test(primitive))
            self.assertFalse(primitive.locked()) 

def test_context_manager_with_await(self):
        primitives = [
            asyncio.Lock(loop=self.loop),
            asyncio.Condition(loop=self.loop),
            asyncio.Semaphore(loop=self.loop),
            asyncio.BoundedSemaphore(loop=self.loop),
        ]

        async def test(lock):
            await asyncio.sleep(0.01, loop=self.loop)
            self.assertFalse(lock.locked())
            with await lock as _lock:
                self.assertIs(_lock, None)
                self.assertTrue(lock.locked())
                await asyncio.sleep(0.01, loop=self.loop)
                self.assertTrue(lock.locked())
            self.assertFalse(lock.locked())

        for primitive in primitives:
            self.loop.run_until_complete(test(primitive))
            self.assertFalse(primitive.locked()) 

def __init__(self, minsize, maxsize, echo, loop, pool_recycle, **kwargs):
        if minsize < 0:
            raise ValueError("minsize should be zero or greater")
        if maxsize < minsize:
            raise ValueError("maxsize should be not less than minsize")
        self._minsize = minsize
        self._loop = loop
        self._conn_kwargs = kwargs
        self._acquiring = 0
        self._recycle = pool_recycle
        self._free = collections.deque(maxlen=maxsize)
        self._cond = asyncio.Condition(loop=loop)
        self._used = set()
        self._closing = False
        self._closed = False
        self._echo = echo 

def print_tests():
    st = '''Available functions:
print_tests()  Print this list.
ack_test()  Test event acknowledge and Message class.
message_test() Test Message class.
event_test()  Test Event and Lock objects.
barrier_test()  Test the Barrier class.
semaphore_test(bounded=False)  Test Semaphore or BoundedSemaphore.
condition_test()  Test the Condition class.
queue_test()  Test the  Queue class

Recommended to issue ctrl-D after running each test.
'''
    print('\x1b[32m')
    print(st)
    print('\x1b[39m') 

def cond_go():
    cond = asyncio.Condition()
    ntasks = 7
    barrier = Barrier(ntasks + 1)
    t1 = asyncio.create_task(cond01_new(cond))
    t3 = asyncio.create_task(cond03_new())
    for n in range(ntasks):
        asyncio.create_task(cond02(n, cond, barrier))
    await barrier  # All instances of cond02 have completed
    # Test wait_for
    barrier = Barrier(2)
    asyncio.create_task(cond04(99, cond, barrier))
    await barrier
    # cancel continuously running coros.
    t1.cancel()
    t3.cancel()
    await asyncio.sleep(0)
    print('Done.') 

def _get_condition(self):
        """
        Creation of condition is delayed, to minimize the change of using the
        wrong loop.

        A user may create a mock with _AwaitEvent before selecting the
        execution loop.  Requiring a user to delay creation is error-prone and
        inflexible. Instead, condition is created when user actually starts to
        use the mock.
        """
        # No synchronization is needed:
        #   - asyncio is thread unsafe
        #   - there are no awaits here, method will be executed without
        #   switching asyncio context.
        if self._condition is None:
            self._condition = asyncio.Condition()

        return self._condition 

def test_context_manager_async_with(self):
        primitives = [
            asyncio.Lock(loop=self.loop),
            asyncio.Condition(loop=self.loop),
            asyncio.Semaphore(loop=self.loop),
            asyncio.BoundedSemaphore(loop=self.loop),
        ]

        async def test(lock):
            await asyncio.sleep(0.01, loop=self.loop)
            self.assertFalse(lock.locked())
            async with lock as _lock:
                self.assertIs(_lock, None)
                self.assertTrue(lock.locked())
                await asyncio.sleep(0.01, loop=self.loop)
                self.assertTrue(lock.locked())
            self.assertFalse(lock.locked())

        for primitive in primitives:
            self.loop.run_until_complete(test(primitive))
            self.assertFalse(primitive.locked()) 

def test_context_manager_with_await(self):
        primitives = [
            asyncio.Lock(loop=self.loop),
            asyncio.Condition(loop=self.loop),
            asyncio.Semaphore(loop=self.loop),
            asyncio.BoundedSemaphore(loop=self.loop),
        ]

        async def test(lock):
            await asyncio.sleep(0.01, loop=self.loop)
            self.assertFalse(lock.locked())
            with await lock as _lock:
                self.assertIs(_lock, None)
                self.assertTrue(lock.locked())
                await asyncio.sleep(0.01, loop=self.loop)
                self.assertTrue(lock.locked())
            self.assertFalse(lock.locked())

        for primitive in primitives:
            self.loop.run_until_complete(test(primitive))
            self.assertFalse(primitive.locked()) 

def __init__(self, service, name=None, executor=None):
        super().__init__(service, name)
        self._state = State.CREATE

        # A Task may want to run blocking calls in separate thread. To run a
        # method in separate thread, task can use the _run_in_executor() method.
        # User can create their own executor instead using the default one
        # created by the asyncio. This allows user control over the type of
        # executor (task/threads) and its properties (e.g. num_workers)
        self._executor = executor

        # _update_event can be used to notify coroutines about the change in
        # state in this service. e.g. run() has completed
        self._update_event = asyncio.Condition(loop=self.loop)

        self.set_state(State.INIT)

        coro = self.start()
        # fixup task name to show actual task in logs
        coro.__qualname__ = self._objname
        self._task = asyncio.ensure_future(coro, loop=self.loop)

        self._ALL_TASKS[self._objname] = self 

def __init__(self, dsn, minsize, maxsize, timeout, *,
                 enable_json, enable_hstore, enable_uuid, echo,
                 on_connect, pool_recycle, **kwargs):
        if minsize < 0:
            raise ValueError("minsize should be zero or greater")
        if maxsize < minsize and maxsize != 0:
            raise ValueError("maxsize should be not less than minsize")
        self._dsn = dsn
        self._minsize = minsize
        self._loop = get_running_loop(kwargs.pop('loop', None) is not None)
        self._timeout = timeout
        self._recycle = pool_recycle
        self._enable_json = enable_json
        self._enable_hstore = enable_hstore
        self._enable_uuid = enable_uuid
        self._echo = echo
        self._on_connect = on_connect
        self._conn_kwargs = kwargs
        self._acquiring = 0
        self._free = collections.deque(maxlen=maxsize or None)
        self._cond = asyncio.Condition(loop=self._loop)
        self._used = set()
        self._terminated = set()
        self._closing = False
        self._closed = False 

def __init__(self, server_state, fetch_chunk_size=0, capabilities=CAPABILITIES,
                 loop=asyncio.get_event_loop()):
        self.uidvalidity = int(datetime.now().timestamp())
        self.capabilities = capabilities
        self.state_to_send = list()
        self.delay_seconds = 0
        self.loop = loop
        self.fetch_chunk_size = fetch_chunk_size
        self.transport = None
        self.server_state = server_state
        self.user_login = None
        self.user_mailbox = None
        self.idle_tag = None
        self.idle_task = None
        self.state = NONAUTH
        self.state_condition = asyncio.Condition()
        self.append_literal_command = None 

def __init__(self, loop, conn_lost_cb=None):
        self.loop = loop
        self.transport = None
        self.state = STARTED
        self.state_condition = asyncio.Condition()
        self.capabilities = set()
        self.pending_async_commands = dict()
        self.pending_sync_command = None
        self.idle_queue = asyncio.Queue()
        self.imap_version = None
        self.literal_data = None
        self.incomplete_line = b''
        self.current_command = None
        self.conn_lost_cb = conn_lost_cb

        self.tagnum = 0
        self.tagpre = int2ap(random.randint(4096, 65535)) 

def main(loop):
    # Create a condition
    condition = asyncio.Condition()

    # Set up tasks watching the condition
    consumers = [
        consumer(condition, i)
        for i in range(5)
    ]

    # Schedule a task to manipulate the condition variable
    loop.create_task(manipulate_condition(condition))

    # Wait for the consumers to be done
    await asyncio.wait(consumers) 

def __init__(self, address, db=None, password=None, encoding=None,
                 *, minsize, maxsize, ssl=None, parser=None,
                 create_connection_timeout=None,
                 connection_cls=None,
                 loop=None):
        assert isinstance(minsize, int) and minsize >= 0, (
            "minsize must be int >= 0", minsize, type(minsize))
        assert maxsize is not None, "Arbitrary pool size is disallowed."
        assert isinstance(maxsize, int) and maxsize > 0, (
            "maxsize must be int > 0", maxsize, type(maxsize))
        assert minsize <= maxsize, (
            "Invalid pool min/max sizes", minsize, maxsize)
        if loop is not None and sys.version_info >= (3, 8):
            warnings.warn("The loop argument is deprecated",
                          DeprecationWarning)
        self._address = address
        self._db = db
        self._password = password
        self._ssl = ssl
        self._encoding = encoding
        self._parser_class = parser
        self._minsize = minsize
        self._create_connection_timeout = create_connection_timeout
        self._pool = collections.deque(maxlen=maxsize)
        self._used = set()
        self._acquiring = 0
        self._cond = asyncio.Condition(lock=Lock())
        self._close_state = CloseEvent(self._do_close)
        self._pubsub_conn = None
        self._connection_cls = connection_cls 

def __init__(self):
            """Initialize state."""
            self.value = 0
            self.active = False
            self.condition = asyncio.Condition()
            self.tag_sequence = [] 

def __init__(self, amqp_target, route_key, username=None, password=None, **task_kwargs):
        super().__init__(amqp_target, route_key, username, password, **task_kwargs)
        self._task._citizen_condition_new_block = Condition(loop=self.loop)
        self._task._citizen_condition_unregister = Condition(loop=self.loop) 

def __init__(self, address, db=0, password=0, encoding=None,
                 *, minsize, maxsize, commands_factory, ssl=None, loop=None):
        if loop is not None and sys.version_info >= (3, 8):
            warnings.warn("The loop argument is deprecated",
                          DeprecationWarning)
        if loop is None and sys.version_info < (3, 8):
            loop = asyncio.get_event_loop()

        self._address = address
        self._db = db
        self._password = password
        self._encoding = encoding
        self._minsize = minsize
        self._maxsize = maxsize
        self._factory = commands_factory
        self._ssl = ssl
        self._loop = loop

        # fake it here, we always only have one connection
        self._pool = collections.deque(maxlen=1)
        self._used = set()
        self._acquiring = 0

        self._cond = asyncio.Condition(loop=loop)
        self._close_state = asyncio.Event(loop=loop)
        self._close_waiter = asyncio.ensure_future(self._do_close(), loop=loop) 

def test_ctor_loop(self):
        loop = mock.Mock()
        cond = asyncio.Condition(loop=loop)
        self.assertIs(cond._loop, loop)

        cond = asyncio.Condition(loop=self.loop)
        self.assertIs(cond._loop, self.loop) 

def test_ctor_noloop(self):
        asyncio.set_event_loop(self.loop)
        cond = asyncio.Condition()
        self.assertIs(cond._loop, self.loop) 

def test_wait_cancel(self):
        cond = asyncio.Condition(loop=self.loop)
        self.loop.run_until_complete(cond.acquire())

        wait = asyncio.Task(cond.wait(), loop=self.loop)
        self.loop.call_soon(wait.cancel)
        self.assertRaises(
            asyncio.CancelledError,
            self.loop.run_until_complete, wait)
        self.assertFalse(cond._waiters)
        self.assertTrue(cond.locked()) 

def test_wait_for(self):
        cond = asyncio.Condition(loop=self.loop)
        presult = False

        def predicate():
            return presult

        result = []

        @asyncio.coroutine
        def c1(result):
            yield from cond.acquire()
            if (yield from cond.wait_for(predicate)):
                result.append(1)
                cond.release()
            return True

        t = asyncio.Task(c1(result), loop=self.loop)

        test_utils.run_briefly(self.loop)
        self.assertEqual([], result)

        self.loop.run_until_complete(cond.acquire())
        cond.notify()
        cond.release()
        test_utils.run_briefly(self.loop)
        self.assertEqual([], result)

        presult = True
        self.loop.run_until_complete(cond.acquire())
        cond.notify()
        cond.release()
        test_utils.run_briefly(self.loop)
        self.assertEqual([1], result)

        self.assertTrue(t.done())
        self.assertTrue(t.result()) 

def test_wait_for_unacquired(self):
        cond = asyncio.Condition(loop=self.loop)

        # predicate can return true immediately
        res = self.loop.run_until_complete(cond.wait_for(lambda: [1, 2, 3]))
        self.assertEqual([1, 2, 3], res)

        self.assertRaises(
            RuntimeError,
            self.loop.run_until_complete,
            cond.wait_for(lambda: False)) 

def test_notify_all(self):
        cond = asyncio.Condition(loop=self.loop)

        result = []

        @asyncio.coroutine
        def c1(result):
            yield from cond.acquire()
            if (yield from cond.wait()):
                result.append(1)
                cond.release()
            return True

        @asyncio.coroutine
        def c2(result):
            yield from cond.acquire()
            if (yield from cond.wait()):
                result.append(2)
                cond.release()
            return True

        t1 = asyncio.Task(c1(result), loop=self.loop)
        t2 = asyncio.Task(c2(result), loop=self.loop)

        test_utils.run_briefly(self.loop)
        self.assertEqual([], result)

        self.loop.run_until_complete(cond.acquire())
        cond.notify_all()
        cond.release()
        test_utils.run_briefly(self.loop)
        self.assertEqual([1, 2], result)

        self.assertTrue(t1.done())
        self.assertTrue(t1.result())
        self.assertTrue(t2.done())
        self.assertTrue(t2.result()) 

def test_notify_unacquired(self):
        cond = asyncio.Condition(loop=self.loop)
        self.assertRaises(RuntimeError, cond.notify) 

def test_repr(self):
        cond = asyncio.Condition(loop=self.loop)
        self.assertTrue('unlocked' in repr(cond))
        self.assertTrue(RGX_REPR.match(repr(cond)))

        self.loop.run_until_complete(cond.acquire())
        self.assertTrue('locked' in repr(cond))

        cond._waiters.append(mock.Mock())
        self.assertTrue('waiters:1' in repr(cond))
        self.assertTrue(RGX_REPR.match(repr(cond)))

        cond._waiters.append(mock.Mock())
        self.assertTrue('waiters:2' in repr(cond))
        self.assertTrue(RGX_REPR.match(repr(cond))) 

def test_context_manager(self):
        cond = asyncio.Condition(loop=self.loop)

        @asyncio.coroutine
        def acquire_cond():
            return (yield from cond)

        with self.loop.run_until_complete(acquire_cond()):
            self.assertTrue(cond.locked())

        self.assertFalse(cond.locked()) 

def test_context_manager_no_yield(self):
        cond = asyncio.Condition(loop=self.loop)

        try:
            with cond:
                self.fail('RuntimeError is not raised in with expression')
        except RuntimeError as err:
            self.assertEqual(
                str(err),
                '"yield from" should be used as context manager expression')

        self.assertFalse(cond.locked()) 

def test_explicit_lock(self):
        lock = asyncio.Lock(loop=self.loop)
        cond = asyncio.Condition(lock, loop=self.loop)

        self.assertIs(cond._lock, lock)
        self.assertIs(cond._loop, lock._loop) 

def test_ambiguous_loops(self):
        loop = self.new_test_loop()
        self.addCleanup(loop.close)

        lock = asyncio.Lock(loop=self.loop)
        with self.assertRaises(ValueError):
            asyncio.Condition(lock, loop=loop)