@asyncio.coroutine<EOL><INDENT>def request(self, subject, payload, timeout=<NUM_LIT:0.5>, expected=<NUM_LIT:1>, cb=None):<DEDENT>
if self.is_draining_pubs:<EOL><INDENT>raise ErrConnectionDraining<EOL><DEDENT>if cb is not None:<EOL><INDENT>next_inbox = INBOX_PREFIX[:]<EOL>next_inbox.extend(self._nuid.next())<EOL>inbox = next_inbox.decode()<EOL>sid = yield from self.subscribe(inbox, cb=cb)<EOL>yield from self.auto_unsubscribe(sid, expected)<EOL>yield from self.publish_request(subject, inbox, payload)<EOL>return sid<EOL><DEDENT>if self._resp_sub_prefix is None:<EOL><INDENT>self._resp_map = {}<EOL>self._resp_sub_prefix = INBOX_PREFIX[:]<EOL>self._resp_sub_prefix.extend(self._nuid.next())<EOL>self._resp_sub_prefix.extend(b'<STR_LIT:.>')<EOL>resp_mux_subject = self._resp_sub_prefix[:]<EOL>resp_mux_subject.extend(b'<STR_LIT:*>')<EOL>sub = Subscription(subject=resp_mux_subject.decode())<EOL>sub.pending_msgs_limit = DEFAULT_SUB_PENDING_MSGS_LIMIT<EOL>sub.pending_bytes_limit = DEFAULT_SUB_PENDING_BYTES_LIMIT<EOL>sub.pending_queue = asyncio.Queue(<EOL>maxsize=sub.pending_msgs_limit,<EOL>loop=self._loop,<EOL>)<EOL>@asyncio.coroutine<EOL>def wait_for_msgs():<EOL><INDENT>nonlocal sub<EOL>while True:<EOL><INDENT>try:<EOL><INDENT>msg = yield from sub.pending_queue.get()<EOL>token = msg.subject[INBOX_PREFIX_LEN:]<EOL>try:<EOL><INDENT>fut = self._resp_map[token]<EOL>fut.set_result(msg)<EOL>del self._resp_map[token]<EOL><DEDENT>except (asyncio.CancelledError, asyncio.InvalidStateError):<EOL><INDENT>del self._resp_map[token]<EOL>continue<EOL><DEDENT>except KeyError:<EOL><INDENT>continue<EOL><DEDENT><DEDENT>except asyncio.CancelledError:<EOL><INDENT>break<EOL><DEDENT><DEDENT><DEDENT>sub.wait_for_msgs_task = self._loop.create_task(<EOL>wait_for_msgs())<EOL>self._ssid += <NUM_LIT:1><EOL>ssid = self._ssid<EOL>self._subs[ssid] = sub<EOL>yield from self._subscribe(sub, ssid)<EOL><DEDENT>token = self._nuid.next()<EOL>inbox = self._resp_sub_prefix[:]<EOL>inbox.extend(token)<EOL>future = asyncio.Future(loop=self._loop)<EOL>self._resp_map[token.decode()] = future<EOL>yield from self.publish_request(subject, inbox.decode(), payload)<EOL>try:<EOL><INDENT>msg = yield from asyncio.wait_for(future, timeout, loop=self._loop)<EOL>return msg<EOL><DEDENT>except asyncio.TimeoutError:<EOL><INDENT>future.cancel()<EOL>raise ErrTimeout<EOL><DEDENT>
Implements the request/response pattern via pub/sub
using a single wildcard subscription that handles
the responses.