def _process_msg(self, sid, subject, reply, data):
        """
        Dispatches the received message to the stored subscription.
        It first tries to detect whether the message should be
        dispatched to a passed callback.  In case there was not
        a callback, then it tries to set the message into a future.
        """
        payload_size = len(data)
        self.stats['in_msgs'] += 1
        self.stats['in_bytes'] += payload_size

        msg = Msg(subject=subject.decode(), reply=reply.decode(), data=data)

        # Don't process the message if the subscription has been removed
        sub = self._subs.get(sid)
        if sub is None:
            raise tornado.gen.Return()

        # Check if it is an old style request.
        if sub.future is not None:
            sub.future.set_result(msg)

            # Discard subscription since done
            self._subs.pop(sid, None)
            self._remove_subscription(sub)
            raise tornado.gen.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 self._error_cb(ErrSlowConsumer())
                raise tornado.gen.Return()
            sub.pending_queue.put_nowait(msg)
        except tornado.queues.QueueFull:
            if self._error_cb is not None:
                yield self._error_cb(ErrSlowConsumer())
Dispatches the received message to the stored subscription.
        It first tries to detect whether the message should be
        dispatched to a passed callback.  In case there was not
        a callback, then it tries to set the message into a future.