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):
        """
        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))
Process MSG sent by server.