Python asyncio 模块，wait_for() 实例源码

def _get_connection(target,
                          port,
                          ssl,
                          timeout,
                          loop):
    con = asyncio.open_connection(host=target,
                                  port=port,
                                  ssl=ssl)

    try:
        reader, writer = await asyncio.wait_for(con,
                                                int(timeout),
                                                loop=loop)

        return reader, writer
    except (asyncio.TimeoutError, ConnectionRefusedError):
        # If this is reach -> port closed
        return None, None

def execute(self):
        """
        Execute call.

        :return: Future with results.
        :rtype: Future<any>
        """
        # Create new future to be returned. This future will be the answered inside a script callback.
        future = None
        if self.response_id:
            self._client.script_handlers[self.response_id] = future = asyncio.Future()

        # Execute the call itself and register the callback script handler.
        gbx_res = await self._client.execute(self.method, *self.args)

        if self.response_id:
            return await asyncio.wait_for(future, self.timeout) # Timeout after 15 seconds!
        return gbx_res

def test_scp(event_loop):
    async with base.CleanModel() as model:
        await model.add_machine()
        await asyncio.wait_for(
            model.block_until(lambda: model.machines),
            timeout=240)
        machine = model.machines['0']
        await asyncio.wait_for(
            model.block_until(lambda: (machine.status == 'running' and
                                       machine.agent_status == 'started')),
            timeout=480)

        with NamedTemporaryFile() as f:
            f.write(b'testcontents')
            f.flush()
            await machine.scp_to(f.name, 'testfile')

        with NamedTemporaryFile() as f:
            await machine.scp_from('testfile', f.name)
            assert f.read() == b'testcontents'

def test_scp(event_loop):
    async with base.CleanModel() as model:
        app = await model.deploy('ubuntu')

        await asyncio.wait_for(
            model.block_until(lambda: app.units),
            timeout=60)
        unit = app.units[0]
        await asyncio.wait_for(
            model.block_until(lambda: unit.machine),
            timeout=60)
        machine = unit.machine
        await asyncio.wait_for(
            model.block_until(lambda: (machine.status == 'running' and
                                       machine.agent_status == 'started')),
            timeout=480)

        with NamedTemporaryFile() as f:
            f.write(b'testcontents')
            f.flush()
            await unit.scp_to(f.name, 'testfile')

        with NamedTemporaryFile() as f:
            await unit.scp_from('testfile', f.name)
            assert f.read() == b'testcontents'

def test_upgrade_charm_resource(event_loop):
    async with base.CleanModel() as model:
        app = await model.deploy('cs:~cynerva/upgrade-charm-resource-test-1')

        def units_ready():
            if not app.units:
                return False
            unit = app.units[0]
            return unit.workload_status == 'active' and \
                unit.agent_status == 'idle'

        await asyncio.wait_for(model.block_until(units_ready), timeout=480)
        unit = app.units[0]
        expected_message = 'I have no resource.'
        assert unit.workload_status_message == expected_message

        await app.upgrade_charm(revision=2)
        await asyncio.wait_for(
            model.block_until(
                lambda: unit.workload_status_message != 'I have no resource.'
            ),
            timeout=60
        )
        expected_message = 'My resource: I am the resource.'
        assert app.units[0].workload_status_message == expected_message

def calculate(self, *, equation : str):
        '''Calculator'''
        #_equation = re.sub("[^[0-9]+-/*^%\.]", "", equation).replace('^', "**") #words
        replacements = {"pi" : "math.pi", 'e' : "math.e", "sin" : "math.sin", "cos" : "math.cos", "tan" : "math.tan", '^' : "**"}
        allowed = set("0123456789.+-*/^%()")
        for key, value in replacements.items():
            equation = equation.replace(key, value)
        equation = "".join(character for character in equation if character in allowed)
        print("Calculated " + equation)
        with multiprocessing.Pool(1) as pool:
            async_result = pool.apply_async(eval, (equation,))
            future = self.bot.loop.run_in_executor(None, async_result.get, 10.0)
            try:
                result = await asyncio.wait_for(future, 10.0, loop = self.bot.loop)
                await self.bot.embed_reply("{} = {}".format(equation, result))
            except discord.errors.HTTPException:
                await self.bot.embed_reply(":no_entry: Output too long")
            except SyntaxError:
                await self.bot.embed_reply(":no_entry: Syntax error")
            except ZeroDivisionError:
                await self.bot.embed_reply(":no_entry: Error: Division by zero")
            except (concurrent.futures.TimeoutError, multiprocessing.context.TimeoutError):
                await self.bot.embed_reply(":no_entry: Execution exceeded time limit")

def test_limit(self):
        await self.plugin.before_request(self.endpoint_desc, self.session,
                                         self.request_params)

        fut = asyncio.ensure_future(self.plugin.before_request(self.endpoint_desc, self.session,
                                                               self.request_params))

        with self.assertRaises(TimeoutError):
            await asyncio.wait_for(shield(fut), 0.1)

        await self.plugin.on_response(self.endpoint_desc, self.session,
                                      self.request_params, None)

        await asyncio.sleep(0.2)

        await asyncio.wait_for(fut, 0.5)

def __aenter__(self):
        # Timeout != 0 -> apply timeout
        try:
            if self.timeout:
                return await asyncio.wait_for(self.fn(*self.args,
                                                      **self.kwargs),
                                              timeout=self.timeout,
                                              loop=self.loop)
            # Timeout == 0 -> infinite --> Apply very long timeout
            else:
                return await asyncio.wait_for(self.fn(*self.args,
                                                      **self.kwargs),
                                              timeout=self.infinite_timeout,
                                              loop=self.loop)

        except concurrent.futures.TimeoutError as e:
            log.error(
                '{function}: {error_message}'.format(function=self.fn.__name__,
                                                     error_message=e))
            raise AioTasksTimeout(e) from e

def testing(label, cache, loop):

    def w(g):
        return asyncio.wait_for(g, args.timeout, loop=loop)

    key = 'foo-%s' % label
    while True:
        logging.info('%s %s', label, '-'*20)
        try:
            ret = yield from w(cache.set(key, 'hello-%s-world' % label))
            logging.info('%s set %s', label, ret)
            ret = yield from w(cache.get(key))
            logging.info('%s get %s', label, ret)
            ret = yield from w(cache.delete(key))
            logging.info('%s del %s', label, ret)
            ret = yield from w(cache.get(key))
            logging.info('%s get2 %s', label, ret)
        except asyncio.TimeoutError:
            logging.warn('%s Timeout', label)
        except Exception as exc:
            logging.exception('%s Client exception: %r', label, exc)
            break

def set_up_relay(self, addr, port, **kwargs):
        if not self.client:
            assert (addr is not None and port is not None)
            try:
                client = self.get_relay_protocal()
                fut = self.loop.create_connection(
                    lambda: client,
                    addr,
                    port,
                    **kwargs)

                _, self.client = yield from asyncio.wait_for(fut, constants.RELAY_CONNECT_TIMEOUT, loop=self.loop)
            except (ConnectionError, concurrent.futures.TimeoutError):
                PROTO_LOG.exception('Fail to set up connection to %s:%d', addr, port)
                return False
            else:
                PROTO_LOG.info('Connection to {}'.format(self.client.transport.get_extra_info('peername')))
                return True

        else:
            PROTO_LOG.warn('client(%s) alreader exist!', self.client.__repr__)
            return True

def expect_async(expecter, timeout=None):
    # First process data that was previously read - if it maches, we don't need
    # async stuff.
    previously_read = expecter.spawn.buffer
    expecter.spawn.buffer = expecter.spawn.string_type()
    idx = expecter.new_data(previously_read)
    if idx is not None:
        return idx

    transport, pw = yield from asyncio.get_event_loop()\
        .connect_read_pipe(lambda: PatternWaiter(expecter), expecter.spawn)

    try:
        return (yield from asyncio.wait_for(pw.fut, timeout))
    except asyncio.TimeoutError as e:
        transport.pause_reading()
        return expecter.timeout(e)

def wait_cgroup(sock, execute_task, time_limit_ns, memory_limit_bytes, process_limit):
    cgroup = CGroup()
    try:
        cgroup.memory_limit_bytes = memory_limit_bytes
        cgroup.pids_max = process_limit
        await cgroup.accept(sock)
        start_idle = _get_idle()

        while True:
            cpu_usage_ns = cgroup.cpu_usage_ns
            idle_usage_ns = int((_get_idle() - start_idle) / cpu_count() * 1e9)
            time_usage_ns = max(cpu_usage_ns, idle_usage_ns)
            time_remain_ns = time_limit_ns - time_usage_ns
            if time_remain_ns <= 0:
                return time_usage_ns, cgroup.memory_usage_bytes
            try:
                await wait_for(shield(execute_task), (time_remain_ns + WAIT_JITTER_NS) / 1e9)
                return cgroup.cpu_usage_ns, cgroup.memory_usage_bytes
            except TimeoutError:
                pass
    finally:
        while cgroup.kill():
            await sleep(.001)
        cgroup.close()

def run_in_event_loop(self):
        # starting up, set the enabled signals for the Agent and background tasks
        self.run_enabled.set()
        self.oid_updaters_enabled.set()
        self.stopped.clear()

        # run while
        while self.run_enabled.is_set():
            # start the MIB updater(s) and remember the future obj.
            background_task = self.mib_table.start_background_tasks(self.oid_updaters_enabled)
            # wait for the socket manager to close
            await self.socket_mgr.connection_loop()

            #
            # Main thread will block here until the connection closes.
            # When this await is passed, we enter the shutdown phase.
            #

            # signal background tasks to halt
            self.oid_updaters_enabled.clear()
            # wait for handlers to come back
            await asyncio.wait_for(background_task, BACKGROUND_WAIT_TIMEOUT, loop=self.loop)

        # signal that we're done!
        self.stopped.set()

def run(self):
        """
        Runs the async partion reciever event loop to retrive messages from the event queue
        """
        # Implement pull max batch from queue instead of one message at a time
        while (not self.eh_partition_pump.is_closing()) \
              or self.eh_partition_pump.pump_status == "Errored":
            try:
                if self.eh_partition_pump.partition_receive_handler:
                    msgs = await asyncio.wait_for(self.eh_partition_pump.\
                                                    partition_receive_handler. \
                                                    receive(self.max_batch_size),
                                                self.recieve_timeout,
                                                loop=self.eh_partition_pump.loop)
                    await self.process_events_async(msgs)
            except asyncio.TimeoutError as err:
                if self.eh_partition_pump.partition_receive_handler:
                    logging.info("No events received, queue size %d, delivered %d",
                                self.eh_partition_pump.partition_receive_handler.messages.qsize(),
                                self.eh_partition_pump.partition_receive_handler.delivered)
                if self.eh_partition_pump.host.eph_options.release_pump_on_timeout:
                    await self.process_error_async(err)

def pump(_pid, _recv, _dl):
    total = 0
    iteration = 0
    while time.time() < _dl:
        try:
            batch = await asyncio.wait_for(_recv.receive(100), 60.0)
            size = len(batch)
            total += size
            iteration += size
            if iteration >= 80:
                iteration = 0
                logger.info("%s: total received %d, last sn=%d, last offset=%s",
                            _pid,
                            total,
                            batch[-1].sequence_number,
                            batch[-1].offset)
        except asyncio.TimeoutError:
            logger.info("%s: No events received, queue size %d, delivered %d",
                        _pid,
                        _recv.messages.qsize(),
                        _recv.delivered)

def read(self, size=-1):
        log.debug('[Connection.read]: {}: size = {}'.format(self.key, size))
        self._watch()
        # assert self.closed() is not True, 'connection is closed'
        # assert self.stale() is not True, 'connection is stale'

        if self.stale():
            log.debug('[Connection.read] [Error] '
                      '[ConnectionIsStale]: {}'.format(self.key))
            raise ConnectionIsStale('{}'.format(self.key))

        if size < 0:
            chuck = yield from asyncio.wait_for(self.reader.read(size),
                                                timeout=MAX_CONNECTION_TIMEOUT)
            return chuck
        else:
            chucks = b''
            while size:
                chuck = yield from asyncio.wait_for(self.reader.read(size),
                                                    timeout=MAX_CONNECTION_TIMEOUT)
                size -= len(chuck)
                chucks += chuck
            return chucks

def readline(self):
        # assert self.closed() is False, 'connection is closed'
        # assert self.stale() is not True, 'connection is stale'

        if self.stale():
            log.debug('[Connection.readline] [Error] '
                      '[ConnectionIsStale]: {}'.format(self.key))
            raise ConnectionIsStale('{}'.format(self.key))

        chuck = yield from asyncio.wait_for(self.reader.readline(),
                                            timeout=MAX_CONNECTION_TIMEOUT)

        log.debug('[Connection.readline]: '
                  '{}: size = {}'.format(self.key, len(chuck)))

        return chuck

def finish_connections(self, timeout=None):
        # try to close connections in 90% of graceful timeout
        timeout90 = None
        if timeout:
            timeout90 = timeout / 100 * 90

        for handler in self._connections.keys():
            handler.closing(timeout=timeout90)

        if timeout:
            try:
                await wait_for(
                    self._connections_cleanup(), timeout, loop=self._loop)
            except TimeoutError:
                self._server_logger.warning(
                    "Not all connections are closed (pending: %d)",
                    len(self._connections))

        for transport in self._connections.values():
            transport.close()

        self._connections.clear()

def wait_for(self, word, try_scanning=False):
        self.root.info("WAITING FOR {} BUTTON TO APPEAR".format(word))
        ok = ''
        while ok != word and not self.run_time.stop:
            # root.debug("waiting for {}".format(word))
            img = self.get_img_from_screen_shot()
            img = img[745:770, 210:270]
            try:
                if try_scanning:
                    self.scan_for_word('ok', LOW_CORR)
                ok = self.img_to_string(img,
                                        "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789")
            except:
                self.wait_for_ui(1)
                continue
            if ok == word:
                break
            self.wait_for_ui(2)

def _start_loop(self, websocket, event_handler):
        """
        We will listen for websockets events, sending a heartbeat/pong everytime
        we react a TimeoutError. If we don't the webserver would close the idle connection,
        forcing us to reconnect.
        """
        log.debug('Starting websocket loop')
        while True:
            try:
                yield from asyncio.wait_for(
                    self._wait_for_message(websocket, event_handler),
                    timeout=self.options['timeout']
                )
            except asyncio.TimeoutError:
                yield from websocket.pong()
                log.debug("Sending heartbeat...")
                continue

def poll(self):
        """Wait for packets to send to the client."""
        try:
            packets = [await asyncio.wait_for(self.queue.get(),
                                              self.server.ping_timeout)]
            self.queue.task_done()
        except (asyncio.TimeoutError, asyncio.CancelledError):
            raise IOError()
        if packets == [None]:
            return []
        try:
            packets.append(self.queue.get_nowait())
            self.queue.task_done()
        except asyncio.QueueEmpty:
            pass
        return packets

def trigger(self, data, *args, **kwargs):
        """
        Triggers all activities connected to his port.

        :param data: The data being transmitted.
        :param args:  Random stuff
        :param kwargs: More random stuff
        """
        # ctlog.debug("InputPort.trigger({})".format(data))
        # Only transmit the data if there are activities connected to this port.
        if len(self._activities):
            futures = [None] * len(self._activities)
            for i, activity in enumerate(self._activities):
                futures[i] = activity.trigger([self], {self.attribute_name: data}, self.parent_object, *args, **kwargs)

            try:
                # TODO: This will stop calling modules as soon as one raises an exception. Figure out a way to handle
                #       exceptions individually for each future.
                await wait_for(shield(wait(futures)), self.time_out)
            except Exception as e:
                print(self.channel_name, self.time_out)
                self.parent_object.root.handle_exception(sys.exc_info())

def wait_for_entrance(self, timeout=None, nick=None, build=None, addr=None, check=None):
        """
        Waits for entrance.

        :param timeout:  Time to wait for entrance event, if exceeded, returns None.
        :param nick: Player's nick.
        :param build: Player's build.
        :param addr: Player's address (IP:PORT)
        :return:
        """
        future = asyncio.Future(loop=self.loop)
        margs = (nick, build, addr)
        predicate = self.__get_predicate(margs, check)
        self.__listeners[ServerEvent.ENTRANCE].append((predicate, future))
        try:
            data = yield from asyncio.wait_for(future, timeout,
                                               loop=self.loop)
        except asyncio.TimeoutError:
            data = None
        return data

def wait_for_respawn(self, timeout=None, team=None, nick=None, check=None):
        """
        Waits for respawn event.

        :param timeout: Time to wait for respawn event, if exceeded, returns None.
        :param team: Player's team.
        :param nick: Player's nick.
        :param check: Check function, ignored if none.

        :return: Returns message info dict keys: ('team', 'nick').
        :rtype: dict
        """
        future = asyncio.Future(loop=self.loop)
        margs = (team, nick)
        predicate = self.__get_predicate(margs, check)
        self.__listeners[ServerEvent.RESPAWN].append((predicate, future))
        try:
            data = yield from asyncio.wait_for(future, timeout,
                                               loop=self.loop)
        except asyncio.TimeoutError:
            data = None
        return data

def wait_for_elim_teams_flag(self, timeout=None, team=None, nick=None, points=None, check=None):
        """
        Waits for elim teams flag event.

        :param timeout: Time to wait for event, if exceeded, returns None.
        :param team: Player's team.
        :param nick: Player's nick.
        :param points: Points scored.
        :type points: int
        :param check: Check function, ignored if none.

        :return: Returns message info dict keys: ('team', 'nick', 'points').
        :rtype: dict
        """
        future = asyncio.Future(loop=self.loop)
        margs = (team, nick, points)
        predicate = self.__get_predicate(margs, check)
        self.__listeners[ServerEvent.ELIM_TEAMS_FLAG].append((predicate, future))
        try:
            data = yield from asyncio.wait_for(future, timeout,
                                               loop=self.loop)
        except asyncio.TimeoutError:
            data = None
        return data

def wait_for_team_switched(self, timeout=None, nick=None, old_team=None, new_team=None, check=None):
        """
        Waits for team switch event.

        :param timeout: Time to wait for event, if exceeded, returns None.
        :param old_team: Player's old team.
        :param new_team: Player's new team.
        :param nick: Player's nick.
        :param check: Check function, ignored if none.

        :return: Returns message info dict keys: ('nick', 'old_team', 'new_nick').
        :rtype: dict
        """
        future = asyncio.Future(loop=self.loop)
        margs = (nick, old_team, new_team)
        predicate = self.__get_predicate(margs, check)
        self.__listeners[ServerEvent.TEAM_SWITCHED].append((predicate, future))
        try:
            data = yield from asyncio.wait_for(future, timeout,
                                               loop=self.loop)
        except asyncio.TimeoutError:
            data = None
        return data

def wait_for_round_started(self, timeout=None, check=None):
        """
        Waits for round start.

        :param timeout: Time to wait for event, if exceeded, returns None.
        :param check: Check function, ignored if none.

        :return: Returns an empty dict.
        :rtype: dict
        """
        future = asyncio.Future(loop=self.loop)
        margs = tuple()
        predicate = self.__get_predicate(margs, check)
        self.__listeners[ServerEvent.ROUND_STARTED].append((predicate, future))
        try:
            data = yield from asyncio.wait_for(future, timeout,
                                               loop=self.loop)
        except asyncio.TimeoutError:
            data = None
        return data

def wait_for_game_end(self, timeout=None, score_blue=None, score_red=None, score_yellow=None, score_purple=None, check=None):
        """
        Waits for game end.

        :param timeout: Time to wait for event, if exceeded, returns None.
        :param score_blue: Blue score
        :param score_red: Red score.
        :param score_yellow: Yellow score.
        :param score_purple: Purple score.
        :param check: Check function, ignored if none.

        :return: Returns an empty dict.
        :rtype: dict
        """
        future = asyncio.Future(loop=self.loop)
        margs = (score_blue, score_red, score_yellow, score_purple)
        predicate = self.__get_predicate(margs, check)
        self.__listeners[ServerEvent.GAME_END].append((predicate, future))
        try:
            data = yield from asyncio.wait_for(future, timeout,
                                               loop=self.loop)
        except asyncio.TimeoutError:
            data = None
        return data

def wait_for_elim(self, timeout=None, killer_nick=None, killer_weapon=None, victim_nick=None, victim_weapon=None,
                      check=None):
        """
        Waits for elimination event.

        :param timeout: Time to wait for elimination event, if exceeded, returns None.
        :param killer_nick: Killer's nick to match, ignored if None.
        :param killer_weapon: Killer's weapon to match, ignored if None.
        :param victim_nick:  Victim's nick to match, ignored if None.
        :param victim_weapon: Victim's weapon to match, ignored if None.
        :param check: Check function, ignored if None.

        :return: Returns message info dict keys: ('killer_nick', 'killer_weapon', 'victim_nick', 'victim_weapon')
        :rtype: dict
        """
        future = asyncio.Future(loop=self.loop)
        margs = (killer_nick, killer_weapon, victim_nick, victim_weapon)
        predicate = self.__get_predicate(margs, check)
        self.__listeners[ServerEvent.ELIM].append((predicate, future))
        try:
            elim_info = yield from asyncio.wait_for(future, timeout, loop=self.loop)
        except asyncio.TimeoutError:
            elim_info = None
        return elim_info

def expect_async(expecter, timeout=None):
    # First process data that was previously read - if it maches, we don't need
    # async stuff.
    previously_read = expecter.spawn.buffer
    expecter.spawn.buffer = expecter.spawn.string_type()
    idx = expecter.new_data(previously_read)
    if idx is not None:
        return idx

    transport, pw = yield from asyncio.get_event_loop()\
        .connect_read_pipe(lambda: PatternWaiter(expecter), expecter.spawn)

    try:
        return (yield from asyncio.wait_for(pw.fut, timeout))
    except asyncio.TimeoutError as e:
        transport.pause_reading()
        return expecter.timeout(e)

def connect(self, remote: kademlia.Node) -> LESPeer:
        """
        Connect to the given remote and return a Peer instance when successful.
        Returns None if the remote is unreachable, times out or is useless.
        """
        if remote in self.connected_nodes:
            self.logger.debug("Skipping %s; already connected to it", remote)
            return None
        expected_exceptions = (
            UnreachablePeer, asyncio.TimeoutError, PeerConnectionLost,
            UselessPeer, PeerDisconnected)
        try:
            self.logger.info("Connecting to %s...", remote)
            peer = await asyncio.wait_for(
                handshake(remote, self.privkey, self.peer_class, self.chaindb, self.network_id,
                          self.msg_handler),
                HANDSHAKE_TIMEOUT)
            return cast(LESPeer, peer)
        except expected_exceptions as e:
            self.logger.info("Could not complete handshake with %s: %s", remote, repr(e))
        except Exception:
            self.logger.warn("Unexpected error during auth/p2p handhsake with %s: %s",
                             remote, traceback.format_exc())
        return None

def wait_ping(self, remote: Node) -> bool:
        """Wait for a ping from the given remote.

        This coroutine adds a callback to ping_callbacks and yields control until that callback is
        called or a timeout (k_request_timeout) occurs. At that point it returns whether or not
        a ping was received from the given node.
        """
        if remote in self.ping_callbacks:
            raise AlreadyWaiting(
                "There's another coroutine waiting for a ping packet from {}".format(remote))

        event = asyncio.Event()
        self.ping_callbacks[remote] = event.set
        got_ping = False
        try:
            got_ping = await asyncio.wait_for(event.wait(), k_request_timeout)
            self.logger.debug('got expected ping from {}'.format(remote))
        except asyncio.futures.TimeoutError:
            self.logger.debug('timed out waiting for ping from {}'.format(remote))
        # TODO: Use a contextmanager to ensure we always delete the callback from the list.
        del self.ping_callbacks[remote]
        return got_ping

def open(self):
        host, port = await self._lookup_service()

        conn_fut = self.loop.create_connection(
            ThriftClientProtocolFactory(self._client_class, timeouts=self._timeouts),
            host=host, port=port)
        (transport, protocol) = await asyncio.wait_for(conn_fut,
                                                       self._open_timeout,
                                                       loop=self.loop)
        self._inc_counter('connected')
        self._protocol = protocol
        self._transport = transport

        self._client = protocol.client
        # hookup the close method to the client
        self._client.close = self.close

        self._connected = True
        return self._client

def wait_on_sibling(self, sibling, time_limit=None):
        log.debug("Waiting on sibling %s", sibling)

        path = self.sibling_path(sibling)

        unblocked = self.client.wait_for_event(WatchEvent.DELETED, path)

        exists = await self.client.exists(path=path, watch=True)
        if not exists:
            unblocked.set_result(None)

        try:
            if time_limit:
                await asyncio.wait_for(unblocked, time_limit, loop=self.client.loop.create_future())
            else:
                await unblocked
        except asyncio.TimeoutError:
            raise exc.TimeoutError

def wait(self, timeout=None):
        time_limit = None
        if timeout is not None:
            time_limit = time.time() + timeout

        barrier_lifted = self.client.wait_for_event(
            WatchEvent.DELETED, self.path
        )

        exists = await self.client.exists(path=self.path, watch=True)
        if not exists:
            return

        try:
            if time_limit:
                await asyncio.wait_for(barrier_lifted, time_limit, loop=self.client.loop)
            else:
                await barrier_lifted
        except asyncio.TimeoutError:
            raise exc.TimeoutError

def heartbeat(self):
        if self.closing:
            return

        await self.ensure_safe_state()

        try:
            timeout = self.timeout - self.timeout/HEARTBEAT_FREQUENCY
            zxid, _ = await asyncio.wait_for(self.conn.send(protocol.PingRequest()), timeout, loop=self.loop)
            self.last_zxid = zxid
        except (exc.ConnectError, asyncio.TimeoutError):
            self.state.transition_to(States.SUSPENDED)
        except Exception as e:
            log.exception('in heartbeat: {}'.format(e))
            raise e
        finally:
            self.set_heartbeat()

def on_run(self):
        try:
            while True:
                await asyncio.wait_for(
                    self.revive_event.wait(),
                    timeout=self.timeout,
                    loop=self.loop,
                )
                self.revive_event.clear()
        except asyncio.TimeoutError:
            try:
                if asyncio.iscoroutinefunction(self.callback):
                    await self.callback()
                else:
                    self.callback()
            except Exception:
                logger.exception("Error in timeout callback execution.")

def on_run(self):
        while True:
            try:
                await asyncio.wait_for(
                    self.reset_event.wait(),
                    timeout=self.period,
                    loop=self.loop,
                )
            except asyncio.TimeoutError:
                try:
                    if asyncio.iscoroutinefunction(self.callback):
                        await self.callback()
                    else:
                        self.callback()
                except Exception:
                    logger.exception("Error in timer callback execution.")
            else:
                self.reset_event.clear()

def test_plain_unknown_username(event_loop):
    async with azmq.Context() as context:
        authenticator = ZAPAuthenticator(context)
        authenticator.add_user(username='user', password='pwd')
        context.set_zap_authenticator(authenticator)
        req_socket = context.socket(
            azmq.REQ,
            mechanism=PlainClient(username='user2', password='pwd'),
        )
        rep_socket = context.socket(azmq.REP, mechanism=PlainServer())

        try:
            req_socket.connect(ENDPOINT)
            rep_socket.bind(ENDPOINT)

            await req_socket.send_multipart([b'my', b'request'])

            with pytest.raises(asyncio.TimeoutError):
                await asyncio.wait_for(rep_socket.recv_multipart(), 0.25)

        finally:
            req_socket.close()
            rep_socket.close()

def test_plain_invalid_password(event_loop):
    async with azmq.Context() as context:
        authenticator = ZAPAuthenticator(context)
        authenticator.add_user(username='user', password='pwd')
        context.set_zap_authenticator(authenticator)
        req_socket = context.socket(
            azmq.REQ,
            mechanism=PlainClient(username='user', password='pwd2'),
        )
        rep_socket = context.socket(azmq.REP, mechanism=PlainServer())

        try:
            req_socket.connect(ENDPOINT)
            rep_socket.bind(ENDPOINT)

            await req_socket.send_multipart([b'my', b'request'])

            with pytest.raises(asyncio.TimeoutError):
                await asyncio.wait_for(rep_socket.recv_multipart(), 0.25)

        finally:
            req_socket.close()
            rep_socket.close()

def test_zap_successful_authentication(event_loop):
    class MyZAPAuthenticator(BaseZAPAuthenticator):
        async def on_request(self, *args, **kwargs):
            return 'bob', {b'foo': b'bar'}

    async with azmq.Context() as context:
        async with MyZAPAuthenticator(context=context):
            async with ZAPClient(context=context) as zap_client:
                username, metadata = await asyncio.wait_for(
                    zap_client.authenticate(
                        domain='domain',
                        address='127.0.0.1',
                        identity=b'bob',
                        mechanism=b'CURVE',
                        credentials=[b'mycred', b'value'],
                    ),
                    1,
                )

    assert username == 'bob'
    assert metadata == {b'foo': b'bar'}

def test_zap_authentication_failure(event_loop):
    class MyZAPAuthenticator(BaseZAPAuthenticator):
        async def on_request(self, *args, **kwargs):
            raise ZAPAuthenticationFailure("Some error")

    async with azmq.Context() as context:
        async with MyZAPAuthenticator(context=context):
            async with ZAPClient(context=context) as zap_client:
                with pytest.raises(ZAPAuthenticationFailure) as error:
                    await asyncio.wait_for(
                        zap_client.authenticate(
                            domain='domain',
                            address='127.0.0.1',
                            identity=b'bob',
                            mechanism=b'CURVE',
                            credentials=[b'mycred', b'value'],
                        ),
                        1,
                    )

    assert error.value.code == 400

def test_zap_internal_error(event_loop):
    class MyZAPAuthenticator(BaseZAPAuthenticator):
        async def on_request(self, *args, **kwargs):
            raise RuntimeError

    async with azmq.Context() as context:
        async with MyZAPAuthenticator(context=context):
            async with ZAPClient(context=context) as zap_client:
                with pytest.raises(ZAPInternalError) as error:
                    await asyncio.wait_for(
                        zap_client.authenticate(
                            domain='domain',
                            address='127.0.0.1',
                            identity=b'bob',
                            mechanism=b'CURVE',
                            credentials=[b'mycred', b'value'],
                        ),
                        1,
                    )

    assert error.value.code == 500

def test_zap_custom_error(event_loop):
    class MyZAPAuthenticator(BaseZAPAuthenticator):
        async def on_request(self, *args, **kwargs):
            raise ZAPError('some error', 408)

    async with azmq.Context() as context:
        async with MyZAPAuthenticator(context=context):
            async with ZAPClient(context=context) as zap_client:
                with pytest.raises(ZAPError) as error:
                    await asyncio.wait_for(
                        zap_client.authenticate(
                            domain='domain',
                            address='127.0.0.1',
                            identity=b'bob',
                            mechanism=b'CURVE',
                            credentials=[b'mycred', b'value'],
                        ),
                        1,
                    )

    assert error.value.text == 'some error'
    assert error.value.code == 408

def test_zap_successful_authentication_after_invalid_request(event_loop):
    class MyZAPAuthenticator(BaseZAPAuthenticator):
        async def on_request(self, *args, **kwargs):
            return 'bob', {b'foo': b'bar'}

    async with azmq.Context() as context:
        async with MyZAPAuthenticator(context=context):
            async with ZAPClient(context=context) as zap_client:
                async with context.socket(azmq.DEALER) as socket:
                    socket.connect(ZAP_INPROC_ENDPOINT)
                    await socket.send_multipart([b'invalid', b'data'])

                    username, metadata = await asyncio.wait_for(
                        zap_client.authenticate(
                            domain='domain',
                            address='127.0.0.1',
                            identity=b'bob',
                            mechanism=b'CURVE',
                            credentials=[b'mycred', b'value'],
                        ),
                        1,
                    )

    assert username == 'bob'
    assert metadata == {b'foo': b'bar'}

def test_zap_default_authenticator_authentication_failure(event_loop):
    async with azmq.Context() as context:
        async with ZAPAuthenticator(context=context) as authenticator:
            authenticator.allow('192.168.0.1')

            async with ZAPClient(context=context) as zap_client:
                with pytest.raises(ZAPAuthenticationFailure) as error:
                    await asyncio.wait_for(
                        zap_client.authenticate(
                            domain='domain',
                            address='127.0.0.1',
                            identity=b'bob',
                            mechanism=b'CURVE',
                            credentials=[b'mycred', b'value'],
                        ),
                        1,
                    )

    assert error.value.code == 400

def test_zap_default_authenticator_authentication_success(event_loop):
    async with azmq.Context() as context:
        async with ZAPAuthenticator(context=context) as authenticator:
            authenticator.deny('192.168.0.1')

            async with ZAPClient(context=context) as zap_client:
                username, metadata = await asyncio.wait_for(
                    zap_client.authenticate(
                        domain='domain',
                        address='127.0.0.1',
                        identity=b'bob',
                        mechanism=b'NULL',
                        credentials=[],
                    ),
                    1,
                )

    assert username == ''
    assert metadata == {}

def test_tcp_req_socket(event_loop, socket_factory, connect_or_bind):
    rep_socket = socket_factory.create(zmq.REP)
    connect_or_bind(rep_socket, 'tcp://127.0.0.1:3333', reverse=True)

    def run():
        frames = rep_socket.recv_multipart()
        assert frames == [b'my', b'question']
        rep_socket.send_multipart([b'your', b'answer'])

    with run_in_background(run):
        async with azmq.Context(loop=event_loop) as context:
            socket = context.socket(azmq.REQ)
            connect_or_bind(socket, 'tcp://127.0.0.1:3333')
            await asyncio.wait_for(
                socket.send_multipart([b'my', b'question']),
                1,
            )
            frames = await asyncio.wait_for(socket.recv_multipart(), 1)
            assert frames == [b'your', b'answer']

def test_tcp_rep_socket(event_loop, socket_factory, connect_or_bind):
    req_socket = socket_factory.create(zmq.REQ)
    connect_or_bind(req_socket, 'tcp://127.0.0.1:3333', reverse=True)

    def run():
        req_socket.send_multipart([b'my', b'question'])
        frames = req_socket.recv_multipart()
        assert frames == [b'your', b'answer']

    with run_in_background(run):
        async with azmq.Context(loop=event_loop) as context:
            socket = context.socket(azmq.REP)
            connect_or_bind(socket, 'tcp://127.0.0.1:3333')
            frames = await asyncio.wait_for(socket.recv_multipart(), 1)
            assert frames == [b'my', b'question']
            await asyncio.wait_for(
                socket.send_multipart([b'your', b'answer']),
                1,
            )

def test_tcp_dealer_socket(event_loop, socket_factory, connect_or_bind):
    rep_socket = socket_factory.create(zmq.REP)
    connect_or_bind(rep_socket, 'tcp://127.0.0.1:3333', reverse=True)

    def run():
        frames = rep_socket.recv_multipart()
        assert frames == [b'my', b'question']
        rep_socket.send_multipart([b'your', b'answer'])

    with run_in_background(run):
        async with azmq.Context(loop=event_loop) as context:
            socket = context.socket(azmq.DEALER)
            connect_or_bind(socket, 'tcp://127.0.0.1:3333')
            await asyncio.wait_for(
                socket.send_multipart([b'', b'my', b'question']),
                1,
            )
            frames = await asyncio.wait_for(socket.recv_multipart(), 1)
            assert frames == [b'', b'your', b'answer']