async def _process_msg(self, sub):
while True:<EOL><INDENT>try:<EOL><INDENT>raw_msg = await sub._msgs_queue.get()<EOL>msg = Msg()<EOL>msg_proto = protocol.MsgProto()<EOL>msg_proto.ParseFromString(raw_msg.data)<EOL>msg.proto = msg_proto<EOL>msg.sub = sub<EOL>await sub.cb(msg)<EOL>if not sub.manual_acks:<EOL><INDENT>msg_ack = protocol.Ack()<EOL>msg_ack.subject = msg.proto.subject<EOL>msg_ack.sequence = msg.proto.sequence<EOL>await self._nc.publish(sub.ack_inbox, msg_ack.SerializeToString())<EOL><DEDENT><DEDENT>except asyncio.CancelledError:<EOL><INDENT>break<EOL><DEDENT>except Exception as ex:<EOL><INDENT>if sub.error_cb:<EOL><INDENT>try:<EOL><INDENT>await sub.error_cb(ex)<EOL><DEDENT>except:<EOL><INDENT>logger.exception(<EOL>"<STR_LIT>",<EOL>sub.subject<EOL>)<EOL><DEDENT><DEDENT>continue<EOL><DEDENT><DEDENT>
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.