docs: def listen(self, message_consumer):
          """Blocking call to listen for messages on the rfile.
  
          Args:
              message_consumer (fn): function that is passed each message as it is read off the socket.
          """
          while not self._rfile.closed:
              request_str = self._read_message()
  
              if request_str is None:
                  break
  
              try:
                  message_consumer(json.loads(request_str.decode('utf-8')))
              except ValueError:
                  log.exception("Failed to parse JSON message %s", request_str)
                  continue