def test_barrier(zk, path):
    is_lifted = False
    is_worker_started = zk.loop.create_future()

    async def start_worker():
        barrier = zk.recipes.Barrier(path)
        is_worker_started.set_result('ok')
        await barrier.wait()
        assert is_lifted is True

    barrier = zk.recipes.Barrier(path)
    await barrier.create()

    worker = asyncio.ensure_future(start_worker(), loop=zk.loop)
    is_ok = await is_worker_started
    assert is_ok == 'ok'

    is_lifted = True
    await barrier.lift()
    await worker 

def _download_images(self, session, relative, image_urls):
        self.file_api.prepare_dir(relative)
        successed = 0

        for urls in [image_urls[i:i+self.parallel] for i in range(0, len(image_urls), self.parallel)]:
            done, pendings = await asyncio.wait([self.fetch_image(session, relative, u) for u in urls])
            for d in done:
                try:
                    successed += 1 if d.result() else 0
                except:
                    pass

            if successed >= self.limit:
                break 

def gather(self, wnid, relative="", include_subset=False):
        loop = asyncio.get_event_loop()
        session = self.create_session(loop)
        folders = []

        f = loop.run_until_complete(self.download_images(session, wnid, relative))
        folders.append(f)

        if include_subset:
            wnids = self._get_subsets(wnid)
            path = self.file_api.join_relative(relative, f)
            downloads = asyncio.wait([self.download_images(session, wnid, path) for wnid in wnids])
            done, pending = loop.run_until_complete(downloads)
            folders += [d.result() for d in done]

        session.close()

        return folders 

def listen_message_stream(self, id_blacklist=None):
        id_blacklist = set(id_blacklist or [self.me, ])

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        with aiohttp.ClientSession(loop=loop) as session:
            self.aioclient_session = session

            tasks = [
                asyncio.ensure_future(self.fetch(session, room, id_blacklist))
                for room in self.rooms
            ]
            done, _ = loop.run_until_complete(
                asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
            )
            for d in done:
                if d.exception():
                    raise d.exception() 

def wait_for_reaction_remove(bot, emoji=None, *, user=None,
                                   timeout=None, message=None, check=None):
    """Waits for a reaction to be removed by a user from a message within a time period.
    Made to act like other discord.py wait_for_* functions but is not fully implemented.

    Because of that, wait_for_reaction_remove(self, emoji: list, user, message, timeout=None)
    is a better representation of this function's def

    returns the actual event or None if timeout
    """
    if not (emoji and message) or isinstance(emoji, str):
        raise NotImplementedError("wait_for_reaction_remove(self, emoji, "
                                  "message, user=None, timeout=None, "
                                  "check=None) is a better representation "
                                  "of this function definition")
    remove_event = ReactionRemoveEvent(emoji, user, check=check)
    _reaction_remove_events[message.id] = remove_event
    done, pending = await asyncio.wait([remove_event.wait()],
                                       timeout=timeout)
    res = _reaction_remove_events.pop(message.id)
    try:
        return done.pop().result() and res
    except:
        return None 

def wait_for_first_response(tasks, converters):
    """given a list of unawaited tasks and non-coro result parsers to be called on the results,
    this function returns the 1st result that is returned and converted

    if it is possible for 2 tasks to complete at the same time,
    only the 1st result deteremined by asyncio.wait will be returned

    returns None if none successfully complete
    returns 1st error raised if any occur (probably)
    """
    primed = [wait_for_result(t, c) for t, c in zip(tasks, converters)]
    done, pending = await asyncio.wait(primed, return_when=asyncio.FIRST_COMPLETED)
    for p in pending:
        p.cancel()

    try:
        return done.pop().result()
    except NotImplementedError as e:
        raise e
    except:
        return None 

def async_run(tasks):
    """
    run a group of tasks async
    Requires the tasks arg to be a list of functools.partial()
    """
    if not tasks:
        return

    # start a new async event loop
    loop = asyncio.get_event_loop()
    # https://github.com/python/asyncio/issues/258
    executor = concurrent.futures.ThreadPoolExecutor(5)
    loop.set_default_executor(executor)

    async_tasks = [asyncio.ensure_future(async_task(task, loop)) for task in tasks]
    # run tasks in parallel
    loop.run_until_complete(asyncio.wait(async_tasks))
    # deal with errors (exceptions, etc)
    for task in async_tasks:
        error = task.exception()
        if error is not None:
            raise error

    executor.shutdown(wait=True) 

def fetch_bar2():
    day = datetime.datetime.strptime('20160114', '%Y%m%d').replace(tzinfo=pytz.FixedOffset(480))
    end = datetime.datetime.strptime('20160914', '%Y%m%d').replace(tzinfo=pytz.FixedOffset(480))
    while day <= end:
        day, trading = await is_trading_day(day)
        if trading:
            print('process ', day)
            tasks = [
                asyncio.ensure_future(update_from_shfe(day)),
                asyncio.ensure_future(update_from_dce(day)),
                asyncio.ensure_future(update_from_czce(day)),
                asyncio.ensure_future(update_from_cffex(day)),
            ]
            await asyncio.wait(tasks)
        day += datetime.timedelta(days=1)
    print('all done!')


# asyncio.get_event_loop().run_until_complete(fetch_bar2())
# create_main_all()
# fetch_from_quandl_all()
# clean_dailybar()
# load_kt_data() 

def loop(self: 'TelegramClient') -> asyncio.AbstractEventLoop:
        """
        Property with the ``asyncio`` event loop used by this client.

        Example
            .. code-block:: python

                # Download media in the background
                task = client.loop.create_task(message.download_media())

                # Do some work
                ...

                # Join the task (wait for it to complete)
                await task
        """
        return self._loop 

def test_operations(pool_or_redis, test_case, pool_size):
    repeat = 100
    redis = await pool_or_redis(pool_size)
    done, pending = await asyncio.wait(
        [asyncio.ensure_future(test_case(redis, i))
         for i in range(repeat)])

    assert not pending
    success = 0
    failures = []
    for fut in done:
        exc = fut.exception()
        if exc is None:
            success += 1
        else:
            failures.append(exc)
    assert repeat == success, failures
    assert not failures 

def publish(cls, service: Any, data: Any, topic: str, wait: bool = True, message_protocol: Any = MESSAGE_PROTOCOL_DEFAULT, topic_prefix: Optional[str] = MESSAGE_TOPIC_PREFIX, **kwargs: Any) -> None:
        message_protocol = getattr(service, 'message_protocol', None) if message_protocol == MESSAGE_PROTOCOL_DEFAULT else message_protocol

        payload = data
        if message_protocol:
            build_message_func = getattr(message_protocol, 'build_message', None)
            if build_message_func:
                payload = await build_message_func(service, topic, data, **kwargs)

        topic_arn = await cls.create_topic(cls, topic, service.context, topic_prefix)

        async def _publish_message() -> None:
            await cls.publish_message(cls, topic_arn, payload, service.context)

        if wait:
            await _publish_message()
        else:
            loop = asyncio.get_event_loop()  # type: Any
            loop.create_task(_publish_message()) 

def main(connections2observe4ip):
    logger = logging.getLogger('asyncio.main')
    logger.debug('starting jobs observing connections')
    # Starting the clients
    jobs_on_connections = []
    for _, connection_name, ping_ip in connections2observe4ip:
        # ------------------------------------------------------------------
        # This front-end code hides all details of connection.
        # We just use its name - such name should be meaningful for user.
        # like: "main_dns_server", "backup_ntp_server", ...
        # Another words, all we want here is stg like:
        # "give me connection to main_dns_server"
        # ------------------------------------------------------------------
        con_logger = logging.getLogger('tcp-async-io.{}'.format(connection_name))
        tcp_connection = get_connection(name=connection_name, variant='asyncio', logger=con_logger)

        # client_task= asyncio.ensure_future(ping_observing_task(tcp_connection, ping_ip))
        jobs_on_connections.append(ping_observing_task(tcp_connection, ping_ip))
    # await observers job to be done
    completed, pending = await asyncio.wait(jobs_on_connections)
    logger.debug('all jobs observing connections are done')


# ============================================================================== 

def launch_shards(self):
        if self.shard_count is None:
            self.shard_count, gateway = await self.http.get_bot_gateway()
        else:
            gateway = await self.http.get_gateway()

        self._connection.shard_count = self.shard_count

        shard_ids = self.shard_ids if self.shard_ids else range(self.shard_count)
        self._connection.shard_ids = shard_ids

        for shard_id in shard_ids:
            await self.launch_shard(gateway, shard_id)

        shards_to_wait_for = []
        for shard in self.shards.values():
            shard.complete_pending_reads()
            shards_to_wait_for.append(shard.wait())

        # wait for all pending tasks to finish
        await utils.sane_wait_for(shards_to_wait_for, timeout=300.0) 

def close(self):
        """|coro|

        Closes the connection to Discord.
        """
        if self.is_closed():
            return

        self._closed = True

        for vc in self.voice_clients:
            try:
                await vc.disconnect()
            except Exception:
                pass

        to_close = [asyncio.ensure_future(shard.ws.close(code=1000), loop=self.loop) for shard in self.shards.values()]
        if to_close:
            await asyncio.wait(to_close)

        await self.http.close() 

def test_stream_cancel(event_loop):
    async def cancel(task):
        await asyncio.sleep(0.001)
        task.cancel()

    async def test_stream_iterations(stream):
        async with async_timeout.timeout(0.5):
            while True:
                await _stream_iteration(stream)

    async with aiohttp.ClientSession() as session:
        client = peony.client.BasePeonyClient("", "", session=session)
        context = peony.stream.StreamResponse(method='GET',
                                              url="http://whatever.com",
                                              client=client)

        with context as stream:
            with patch.object(stream, '_connect',
                              side_effect=stream_content):
                coro = test_stream_iterations(stream)
                task = event_loop.create_task(coro)
                cancel_task = event_loop.create_task(cancel(task))

                with async_timeout.timeout(1):
                    await asyncio.wait([task, cancel_task]) 

def __aexit__(self, exc_type, exc_val, exc_tb):
            """Wait for executors to finish, then return."""
            logger.info(
                __("Waiting for executor count to drop to 0, now it is {}", self.value)
            )

            await self.condition.acquire()
            try:
                await self.condition.wait()
            finally:
                self.condition.release()
            logger.debug(
                __(
                    "Sync semaphore dropped to 0, tag sequence was {}.",
                    self.tag_sequence,
                )
            )

            self.active = False
            return False 

def execution_barrier(self):
        """Wait for executors to finish.

        At least one must finish after this point to avoid a deadlock.
        """

        async def _barrier():
            """Enter the sync block and exit the app afterwards."""
            async with self.sync_counter:
                pass
            await consumer.exit_consumer()

        self._ensure_counter()
        await asyncio.wait(
            [_barrier(), consumer.run_consumer(),]
        )
        self.sync_counter = self._SynchronizationManagerDummy() 

def _read(self, size=-1):
        remaining_size = size
        end_time = time() + self.read_timeout
        payload = []
        while remaining_size and (time() < end_time):
            remaining_time = end_time - time()
            done, pending = await asyncio.wait([self.reader.read(remaining_size)],
                                                timeout=remaining_time)
            if done:
                chunk = done.pop().result()
                payload.append(chunk)
                remaining_size -= len(chunk)
            if pending:
                pending.pop().cancel()
        if remaining_size:
            raise exc.UnfinishedRead
        return b''.join(payload) 

def test_double_barrier(zk, path):
    num_workers = 0
    workers = []

    async def start_worker(min_workers):
        barrier = zk.recipes.DoubleBarrier(path, min_workers)
        await barrier.enter()
        for i in range(5):
            assert num_workers >= min_workers
        await barrier.leave()

    target = 8
    for _ in range(target):
        num_workers += 1
        workers.append(
            asyncio.ensure_future(start_worker(target), loop=zk.loop))
    await asyncio.wait(workers, loop=zk.loop)
    await zk.delete(path) 

def test_data_watch(zk, path, data_watcher):
    data = []
    ready = asyncio.Event()
    test_data = b'test' * 1000

    async def data_callback(d):
        data.append(d)
        ready.set()

    data_watcher.add_callback(path, data_callback)
    assert data == []
    await zk.set_data(path, test_data)
    await asyncio.wait_for(ready.wait(), timeout=0.1)
    assert ready.is_set()
    assert data == [test_data]
    data_watcher.remove_callback(path, data_callback) 

def test_data_watch_delete(zk, path, data_watcher):
    data = []
    ready = asyncio.Event()
    test_data = b'test'

    async def data_callback(d):
        data.append(d)
        ready.set()

    await zk.set_data(path, test_data)

    data_watcher.add_callback(path, data_callback)
    await asyncio.sleep(0.2)
    assert data == [test_data]
    ready.clear()
    await zk.delete(path)

    await asyncio.wait_for(ready.wait(), timeout=1)
    assert ready.is_set()
    assert data == [test_data, NoNode]
    data_watcher.remove_callback(path, data_callback)

    await zk.create(path) 

def test_child_watch(child_watcher, path, zk, child1, child2):
    children = set()
    ready = asyncio.Event()

    async def children_callback(c):
        for child in c:
            children.add(child)
            ready.set()

    child_watcher.add_callback(path, children_callback)
    assert children == set()
    await zk.create(child1)
    await asyncio.wait([ready.wait()], timeout=0.1)
    assert children == {child1.split('/')[-1]}
    ready.clear()
    await zk.create(child2)
    await asyncio.wait([ready.wait()], timeout=0.1)
    assert ready.is_set()
    assert children == {child.split('/')[-1] for child in (child1, child2)}
    child_watcher.remove_callback(path, children_callback) 

def test_counter_multiple(zk, path):
    async def worker():
        c = zk.recipes.Counter(path)
        await c.start()
        await c.incr()

    workers = []
    for _i in range(5):
        workers.append(worker())

    done, _pending = await asyncio.wait(workers)
    assert len(done) == 5  # sanity check
    data, stat = await zk.get(path)
    await zk.delete(path)
    assert int(data) == 5
    assert stat.version == 5 

def run_tasks(self, coros, wait=True):
        async def executor(_coro):
            try:
                await _coro

            except Exception:
                pass

            finally:
                self.semaphore.release()

        tasks = []
        for coro in coros:
            await self.semaphore.acquire()
            tasks.append(self.bot.loop.create_task(executor(coro)))

        if wait and tasks:
            await asyncio.wait(tasks) 

def consumer(condition, n):
    with await condition:
        print('consumer {} is waiting'.format(n))
        await condition.wait()
        print('consumer {} triggered'.format(n))
    print('ending consumer {}'.format(n)) 

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 main():
  logging.debug("My Main Function Hit")
  await asyncio.wait([myWorker()]) 

def main(loop):
  lock = asyncio.Lock()
  await asyncio.wait([myWorker(lock), myWorker(lock)]), 

def coro1(event):
    print('coro1 waiting for event')
    await event.wait()
    print('coro1 triggered') 

def main(loop):
    # Create a shared event
    event = asyncio.Event()
    print('event start state: {}'.format(event.is_set()))

    loop.call_later(
        0.1, functools.partial(set_event, event)
    )

    await asyncio.wait([coro1(event), coro2(event)])
    print('event end state: {}'.format(event.is_set()))