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
def request ( self , subject , payload , timeout = 0.5 , expected = 1 , cb = None ) : 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.
Implements the request / response pattern via pub / sub using a single wildcard subscription that handles the responses .