code: async def _process_msg(self, sub):
          
          while True:
              try:
                  raw_msg = await sub._msgs_queue.get()
                  msg = Msg()
                  msg_proto = protocol.MsgProto()
                  msg_proto.ParseFromString(raw_msg.data)
                  msg.proto = msg_proto
                  msg.sub = sub
  
                  # Yield the message to the subscription callback.
                  await sub.cb(msg)
  
                  if not sub.manual_acks:
                      # Process auto-ack if not done manually in the callback,
                      # by publishing into the ack inbox from the subscription.
                      msg_ack = protocol.Ack()
                      msg_ack.subject = msg.proto.subject
                      msg_ack.sequence = msg.proto.sequence
                      await self._nc.publish(sub.ack_inbox, msg_ack.SerializeToString())
              except asyncio.CancelledError:
                  break
              except Exception as ex:
                  if sub.error_cb:
                      try:
                          await sub.error_cb(ex)
                      except:
                          logger.exception(
                              "Exception in error callback for subscription to '%s'",
                              sub.subject
                          )
                  continue
docs: Receives the msgs from the STAN subscriptions and replies.
          By default it will reply back with an ack unless manual acking
          was specified in one of the subscription options.