code: def read_async(self):
          '''
          Asynchronously read messages and invoke a callback when they are ready.
  
          :param callback: A callback with the received data
          '''
          while not self.connected():
              try:
                  yield self.connect(timeout=5)
              except tornado.iostream.StreamClosedError:
                  log.trace('Subscriber closed stream on IPC %s before connect', self.socket_path)
                  yield tornado.gen.sleep(1)
              except Exception as exc:
                  log.error('Exception occurred while Subscriber connecting: %s', exc)
                  yield tornado.gen.sleep(1)
          yield self._read_async(self.__run_callbacksf read_async(self):
          '''
          Asynchronously read messages and invoke a callback when they are ready.
  
          :param callback: A callback with the received data
          '''
          while not self.connected():
              try:
                  yield self.connect(timeout=5)
              except tornado.iostream.StreamClosedError:
                  log.trace('Subscriber closed stream on IPC %s before connect', self.socket_path)
                  yield tornado.gen.sleep(1)
              except Exception as exc:
                  log.error('Exception occurred while Subscriber connecting: %s', exc)
                  yield tornado.gen.sleep(1)
          yield self._read_async(self.__run_callbacks)
docs: Asynchronously read messages and invoke a callback when they are ready.
  
          :param callback: A callback with the received data