nats/aio/client.py
def _process_msg(self, sid, subject, reply, data):
        """
        Process MSG sent by server.
        """
        payload_size = len(data)
        self.stats['in_msgs'] += 1
        self.stats['in_bytes'] += payload_size

        sub = self._subs.get(sid)
        if sub is None:
            # Skip in case no subscription present.
            return

        sub.received += 1
        if sub.max_msgs > 0 and sub.received >= sub.max_msgs:
            # Enough messages so can throwaway subscription now.
            self._subs.pop(sid, None)
        msg = self._build_message(subject, reply, data)

        # Check if it is an old style request.
        if sub.future is not None:
            if sub.future.cancelled():
                # Already gave up, nothing to do.
                return
            sub.future.set_result(msg)
            return

        # Let subscription wait_for_msgs coroutine process the messages,
        # but in case sending to the subscription task would block,
        # then consider it to be an slow consumer and drop the message.
        try:
            sub.pending_size += payload_size
            if sub.pending_size >= sub.pending_bytes_limit:
                # Substract again the bytes since throwing away
                # the message so would not be pending data.
                sub.pending_size -= payload_size

                if self._error_cb is not None:
                    yield from self._error_cb(
                        ErrSlowConsumer(subject=subject, sid=sid))
                return
            sub.pending_queue.put_nowait(msg)
        except asyncio.QueueFull:
            if self._error_cb is not None:
                yield from self._error_cb(
                    ErrSlowConsumer(subject=subject, sid=sid))
def _process_msg(self, sid, subject, reply, data):
        payload_size = len(data)
        self.stats['in_msgs'] += 1
        self.stats['in_bytes'] += payload_size

        sub = self._subs.get(sid)
        if sub is None:
            # Skip in case no subscription present.
            return

        sub.received += 1
        if sub.max_msgs > 0 and sub.received >= sub.max_msgs:
            # Enough messages so can throwaway subscription now.
            self._subs.pop(sid, None)
        msg = self._build_message(subject, reply, data)

        # Check if it is an old style request.
        if sub.future is not None:
            if sub.future.cancelled():
                # Already gave up, nothing to do.
                return
            sub.future.set_result(msg)
            return

        # Let subscription wait_for_msgs coroutine process the messages,
        # but in case sending to the subscription task would block,
        # then consider it to be an slow consumer and drop the message.
        try:
            sub.pending_size += payload_size
            if sub.pending_size >= sub.pending_bytes_limit:
                # Substract again the bytes since throwing away
                # the message so would not be pending data.
                sub.pending_size -= payload_size

                if self._error_cb is not None:
                    yield from self._error_cb(
                        ErrSlowConsumer(subject=subject, sid=sid))
                return
            sub.pending_queue.put_nowait(msg)
        except asyncio.QueueFull:
            if self._error_cb is not None:
                yield from self._error_cb(
                    ErrSlowConsumer(subject=subject, sid=sid))
def _process_msg ( self , sid , subject , reply , data ) : payload_size = len ( data ) self . stats [ 'in_msgs' ] += 1 self . stats [ 'in_bytes' ] += payload_size sub = self . _subs . get ( sid ) if sub is None : # Skip in case no subscription present. return sub . received += 1 if sub . max_msgs > 0 and sub . received >= sub . max_msgs : # Enough messages so can throwaway subscription now. self . _subs . pop ( sid , None ) msg = self . _build_message ( subject , reply , data ) # Check if it is an old style request. if sub . future is not None : if sub . future . cancelled ( ) : # Already gave up, nothing to do. return sub . future . set_result ( msg ) return # Let subscription wait_for_msgs coroutine process the messages, # but in case sending to the subscription task would block, # then consider it to be an slow consumer and drop the message. try : sub . pending_size += payload_size if sub . pending_size >= sub . pending_bytes_limit : # Substract again the bytes since throwing away # the message so would not be pending data. sub . pending_size -= payload_size if self . _error_cb is not None : yield from self . _error_cb ( ErrSlowConsumer ( subject = subject , sid = sid ) ) return sub . pending_queue . put_nowait ( msg ) except asyncio . QueueFull : if self . _error_cb is not None : yield from self . _error_cb ( ErrSlowConsumer ( subject = subject , sid = sid ) )
Process MSG sent by server.
Process MSG sent by server .