code: async def read_loop(self):
          
          while not self.closing:
              try:
                  xid, zxid, response = await self.read_response()
              except (ConnectionAbortedError, asyncio.CancelledError):
                  return
              except Exception as e:
                  log.exception("Error reading response.")
                  self.abort()
                  return
  
              payload_log.debug("[RECV] (xid: %s) %s", xid, response)
  
              if xid == protocol.WATCH_XID:
                  self.watch_handler(response)
                  continue
              elif xid in protocol.SPECIAL_XIDS:
                  f = self.pending_specials[xid].pop()
              else:
                  f = self.pending.pop(xid)
  
              if isinstance(response, Exception):
                  f.set_exception(response)
              elif not f.cancelled():
                  f.set_result((zxid, response))
docs: Infinite loop that reads messages off of the socket while not closed.
  
          When a message is received its corresponding pending Future is set
          to have the message as its result.
  
          This is never used directly and is fired as a separate callback on the
          I/O loop via the `connect()` method.