code: def _process_msg(self, sid, subject, reply, data):
          
          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())
docs: 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.