nats/aio/client.py
def subscribe(self, subject,
                  queue="",
                  cb=None,
                  future=None,
                  max_msgs=0,
                  is_async=False,
                  pending_msgs_limit=DEFAULT_SUB_PENDING_MSGS_LIMIT,
                  pending_bytes_limit=DEFAULT_SUB_PENDING_BYTES_LIMIT,
                  ):
        """
        Takes a subject string and optional queue string to send a SUB cmd,
        and a callback which to which messages (Msg) will be dispatched to
        be processed sequentially by default.
        """
        if subject == "":
            raise ErrBadSubject

        if self.is_closed:
            raise ErrConnectionClosed

        if self.is_draining:
            raise ErrConnectionDraining

        sub = Subscription(subject=subject,
                           queue=queue,
                           max_msgs=max_msgs,
                           is_async=is_async,
                           )
        if cb is not None:
            if asyncio.iscoroutinefunction(cb):
                sub.coro = cb
            elif sub.is_async:
                raise NatsError(
                    "nats: must use coroutine for async subscriptions")
            else:
                # NOTE: Consider to deprecate this eventually, it should always
                # be coroutines otherwise they could affect the single thread,
                # for now still allow to be flexible.
                sub.cb = cb

            sub.pending_msgs_limit = pending_msgs_limit
            sub.pending_bytes_limit = pending_bytes_limit
            sub.pending_queue = asyncio.Queue(
                maxsize=pending_msgs_limit,
                loop=self._loop,
                )

            # Close the delivery coroutine over the sub and error handler
            # instead of having subscription type hold over state of the conn.
            err_cb = self._error_cb

            @asyncio.coroutine
            def wait_for_msgs():
                nonlocal sub
                nonlocal err_cb

                while True:
                    try:
                        msg = yield from sub.pending_queue.get()
                        sub.pending_size -= len(msg.data)

                        try:
                            # Invoke depending of type of handler.
                            if sub.coro is not None:
                                if sub.is_async:
                                    # NOTE: Deprecate this usage in a next release,
                                    # the handler implementation ought to decide
                                    # the concurrency level at which the messages
                                    # should be processed.
                                    self._loop.create_task(sub.coro(msg))
                                else:
                                    yield from sub.coro(msg)
                            elif sub.cb is not None:
                                if sub.is_async:
                                    raise NatsError(
                                        "nats: must use coroutine for async subscriptions")
                                else:
                                    # Schedule regular callbacks to be processed sequentially.
                                    self._loop.call_soon(sub.cb, msg)
                        except asyncio.CancelledError:
                            # In case the coroutine handler gets cancelled
                            # then stop task loop and return.
                            break
                        except Exception as e:
                            # All errors from calling a handler
                            # are async errors.
                            if err_cb is not None:
                                yield from err_cb(e)

                    except asyncio.CancelledError:
                        break

            # Start task for each subscription, it should be cancelled
            # on both unsubscribe and closing as well.
            sub.wait_for_msgs_task = self._loop.create_task(
                wait_for_msgs())

        elif future is not None:
            # Used to handle the single response from a request.
            sub.future = future
        else:
            raise NatsError("nats: invalid subscription type")

        self._ssid += 1
        ssid = self._ssid
        self._subs[ssid] = sub
        yield from self._subscribe(sub, ssid)
        return ssid
def subscribe(self, subject,
                  queue="",
                  cb=None,
                  future=None,
                  max_msgs=0,
                  is_async=False,
                  pending_msgs_limit=DEFAULT_SUB_PENDING_MSGS_LIMIT,
                  pending_bytes_limit=DEFAULT_SUB_PENDING_BYTES_LIMIT,
                  ):
        """
        Takes a subject string and optional queue string to send a SUB cmd,
        and a callback which to which messages (Msg) will be dispatched to
        be processed sequentially by default.
        """
        if subject == "":
            raise ErrBadSubject

        if self.is_closed:
            raise ErrConnectionClosed

        if self.is_draining:
            raise ErrConnectionDraining

        sub = Subscription(subject=subject,
                           queue=queue,
                           max_msgs=max_msgs,
                           is_async=is_async,
                           )
        if cb is not None:
            if asyncio.iscoroutinefunction(cb):
                sub.coro = cb
            elif sub.is_async:
                raise NatsError(
                    "nats: must use coroutine for async subscriptions")
            else:
                # NOTE: Consider to deprecate this eventually, it should always
                # be coroutines otherwise they could affect the single thread,
                # for now still allow to be flexible.
                sub.cb = cb

            sub.pending_msgs_limit = pending_msgs_limit
            sub.pending_bytes_limit = pending_bytes_limit
            sub.pending_queue = asyncio.Queue(
                maxsize=pending_msgs_limit,
                loop=self._loop,
                )

            # Close the delivery coroutine over the sub and error handler
            # instead of having subscription type hold over state of the conn.
            err_cb = self._error_cb

            @asyncio.coroutine
            def wait_for_msgs():
                nonlocal sub
                nonlocal err_cb

                while True:
                    try:
                        msg = yield from sub.pending_queue.get()
                        sub.pending_size -= len(msg.data)

                        try:
                            # Invoke depending of type of handler.
                            if sub.coro is not None:
                                if sub.is_async:
                                    # NOTE: Deprecate this usage in a next release,
                                    # the handler implementation ought to decide
                                    # the concurrency level at which the messages
                                    # should be processed.
                                    self._loop.create_task(sub.coro(msg))
                                else:
                                    yield from sub.coro(msg)
                            elif sub.cb is not None:
                                if sub.is_async:
                                    raise NatsError(
                                        "nats: must use coroutine for async subscriptions")
                                else:
                                    # Schedule regular callbacks to be processed sequentially.
                                    self._loop.call_soon(sub.cb, msg)
                        except asyncio.CancelledError:
                            # In case the coroutine handler gets cancelled
                            # then stop task loop and return.
                            break
                        except Exception as e:
                            # All errors from calling a handler
                            # are async errors.
                            if err_cb is not None:
                                yield from err_cb(e)

                    except asyncio.CancelledError:
                        break

            # Start task for each subscription, it should be cancelled
            # on both unsubscribe and closing as well.
            sub.wait_for_msgs_task = self._loop.create_task(
                wait_for_msgs())

        elif future is not None:
            # Used to handle the single response from a request.
            sub.future = future
        else:
            raise NatsError("nats: invalid subscription type")

        self._ssid += 1
        ssid = self._ssid
        self._subs[ssid] = sub
        yield from self._subscribe(sub, ssid)
        return ssid
def subscribe ( self , subject , queue = "" , cb = None , future = None , max_msgs = 0 , is_async = False , pending_msgs_limit = DEFAULT_SUB_PENDING_MSGS_LIMIT , pending_bytes_limit = DEFAULT_SUB_PENDING_BYTES_LIMIT , ) : if subject == "" : raise ErrBadSubject if self . is_closed : raise ErrConnectionClosed if self . is_draining : raise ErrConnectionDraining sub = Subscription ( subject = subject , queue = queue , max_msgs = max_msgs , is_async = is_async , ) if cb is not None : if asyncio . iscoroutinefunction ( cb ) : sub . coro = cb elif sub . is_async : raise NatsError ( "nats: must use coroutine for async subscriptions" ) else : # NOTE: Consider to deprecate this eventually, it should always # be coroutines otherwise they could affect the single thread, # for now still allow to be flexible. sub . cb = cb sub . pending_msgs_limit = pending_msgs_limit sub . pending_bytes_limit = pending_bytes_limit sub . pending_queue = asyncio . Queue ( maxsize = pending_msgs_limit , loop = self . _loop , ) # Close the delivery coroutine over the sub and error handler # instead of having subscription type hold over state of the conn. err_cb = self . _error_cb @ asyncio . coroutine def wait_for_msgs ( ) : nonlocal sub nonlocal err_cb while True : try : msg = yield from sub . pending_queue . get ( ) sub . pending_size -= len ( msg . data ) try : # Invoke depending of type of handler. if sub . coro is not None : if sub . is_async : # NOTE: Deprecate this usage in a next release, # the handler implementation ought to decide # the concurrency level at which the messages # should be processed. self . _loop . create_task ( sub . coro ( msg ) ) else : yield from sub . coro ( msg ) elif sub . cb is not None : if sub . is_async : raise NatsError ( "nats: must use coroutine for async subscriptions" ) else : # Schedule regular callbacks to be processed sequentially. self . _loop . call_soon ( sub . cb , msg ) except asyncio . CancelledError : # In case the coroutine handler gets cancelled # then stop task loop and return. break except Exception as e : # All errors from calling a handler # are async errors. if err_cb is not None : yield from err_cb ( e ) except asyncio . CancelledError : break # Start task for each subscription, it should be cancelled # on both unsubscribe and closing as well. sub . wait_for_msgs_task = self . _loop . create_task ( wait_for_msgs ( ) ) elif future is not None : # Used to handle the single response from a request. sub . future = future else : raise NatsError ( "nats: invalid subscription type" ) self . _ssid += 1 ssid = self . _ssid self . _subs [ ssid ] = sub yield from self . _subscribe ( sub , ssid ) return ssid
Takes a subject string and optional queue string to send a SUB cmd,
        and a callback which to which messages (Msg) will be dispatched to
        be processed sequentially by default.
Takes a subject string and optional queue string to send a SUB cmd and a callback which to which messages ( Msg ) will be dispatched to be processed sequentially by default .