nats/aio/client.py
def request(self, subject, payload, timeout=0.5, expected=1, cb=None):
        """
        Implements the request/response pattern via pub/sub
        using a single wildcard subscription that handles
        the responses.

        """
        if self.is_draining_pubs:
            raise ErrConnectionDraining

        # If callback given then continue to use old style.
        if cb is not None:
            next_inbox = INBOX_PREFIX[:]
            next_inbox.extend(self._nuid.next())
            inbox = next_inbox.decode()

            sid = yield from self.subscribe(inbox, cb=cb)
            yield from self.auto_unsubscribe(sid, expected)
            yield from self.publish_request(subject, inbox, payload)
            return sid

        if self._resp_sub_prefix is None:
            self._resp_map = {}

            # Create a prefix and single wildcard subscription once.
            self._resp_sub_prefix = INBOX_PREFIX[:]
            self._resp_sub_prefix.extend(self._nuid.next())
            self._resp_sub_prefix.extend(b'.')
            resp_mux_subject = self._resp_sub_prefix[:]
            resp_mux_subject.extend(b'*')
            sub = Subscription(subject=resp_mux_subject.decode())

            # FIXME: Allow setting pending limits for responses mux subscription.
            sub.pending_msgs_limit = DEFAULT_SUB_PENDING_MSGS_LIMIT
            sub.pending_bytes_limit = DEFAULT_SUB_PENDING_BYTES_LIMIT
            sub.pending_queue = asyncio.Queue(
                maxsize=sub.pending_msgs_limit,
                loop=self._loop,
                )

            # Single task for handling the requests
            @asyncio.coroutine
            def wait_for_msgs():
                nonlocal sub
                while True:
                    try:
                        msg = yield from sub.pending_queue.get()
                        token = msg.subject[INBOX_PREFIX_LEN:]

                        try:
                            fut = self._resp_map[token]
                            fut.set_result(msg)
                            del self._resp_map[token]
                        except (asyncio.CancelledError, asyncio.InvalidStateError):
                            # Request may have timed out already so remove entry.
                            del self._resp_map[token]
                            continue
                        except KeyError:
                            # Future already handled so drop any extra
                            # responses which may have made it.
                            continue

                    except asyncio.CancelledError:
                        break

            sub.wait_for_msgs_task = self._loop.create_task(
                wait_for_msgs())

            # Store the subscription in the subscriptions map,
            # then send the protocol commands to the server.
            self._ssid += 1
            ssid = self._ssid
            self._subs[ssid] = sub
            yield from self._subscribe(sub, ssid)

        # Use a new NUID for the token inbox and then use the future.
        token = self._nuid.next()
        inbox = self._resp_sub_prefix[:]
        inbox.extend(token)
        future = asyncio.Future(loop=self._loop)
        self._resp_map[token.decode()] = future
        yield from self.publish_request(subject, inbox.decode(), payload)

        # Wait for the response or give up on timeout.
        try:
            msg = yield from asyncio.wait_for(future, timeout, loop=self._loop)
            return msg
        except asyncio.TimeoutError:
            future.cancel()
            raise ErrTimeout
def request(self, subject, payload, timeout=0.5, expected=1, cb=None):
        """
        Implements the request/response pattern via pub/sub
        using a single wildcard subscription that handles
        the responses.

        """
        if self.is_draining_pubs:
            raise ErrConnectionDraining

        # If callback given then continue to use old style.
        if cb is not None:
            next_inbox = INBOX_PREFIX[:]
            next_inbox.extend(self._nuid.next())
            inbox = next_inbox.decode()

            sid = yield from self.subscribe(inbox, cb=cb)
            yield from self.auto_unsubscribe(sid, expected)
            yield from self.publish_request(subject, inbox, payload)
            return sid

        if self._resp_sub_prefix is None:
            self._resp_map = {}

            # Create a prefix and single wildcard subscription once.
            self._resp_sub_prefix = INBOX_PREFIX[:]
            self._resp_sub_prefix.extend(self._nuid.next())
            self._resp_sub_prefix.extend(b'.')
            resp_mux_subject = self._resp_sub_prefix[:]
            resp_mux_subject.extend(b'*')
            sub = Subscription(subject=resp_mux_subject.decode())

            # FIXME: Allow setting pending limits for responses mux subscription.
            sub.pending_msgs_limit = DEFAULT_SUB_PENDING_MSGS_LIMIT
            sub.pending_bytes_limit = DEFAULT_SUB_PENDING_BYTES_LIMIT
            sub.pending_queue = asyncio.Queue(
                maxsize=sub.pending_msgs_limit,
                loop=self._loop,
                )

            # Single task for handling the requests
            @asyncio.coroutine
            def wait_for_msgs():
                nonlocal sub
                while True:
                    try:
                        msg = yield from sub.pending_queue.get()
                        token = msg.subject[INBOX_PREFIX_LEN:]

                        try:
                            fut = self._resp_map[token]
                            fut.set_result(msg)
                            del self._resp_map[token]
                        except (asyncio.CancelledError, asyncio.InvalidStateError):
                            # Request may have timed out already so remove entry.
                            del self._resp_map[token]
                            continue
                        except KeyError:
                            # Future already handled so drop any extra
                            # responses which may have made it.
                            continue

                    except asyncio.CancelledError:
                        break

            sub.wait_for_msgs_task = self._loop.create_task(
                wait_for_msgs())

            # Store the subscription in the subscriptions map,
            # then send the protocol commands to the server.
            self._ssid += 1
            ssid = self._ssid
            self._subs[ssid] = sub
            yield from self._subscribe(sub, ssid)

        # Use a new NUID for the token inbox and then use the future.
        token = self._nuid.next()
        inbox = self._resp_sub_prefix[:]
        inbox.extend(token)
        future = asyncio.Future(loop=self._loop)
        self._resp_map[token.decode()] = future
        yield from self.publish_request(subject, inbox.decode(), payload)

        # Wait for the response or give up on timeout.
        try:
            msg = yield from asyncio.wait_for(future, timeout, loop=self._loop)
            return msg
        except asyncio.TimeoutError:
            future.cancel()
            raise ErrTimeout
Implements the request/response pattern via pub/sub
        using a single wildcard subscription that handles
        the responses.