Read a message from this stream and handle it.

    This method tries to read a message from this stream, blocking until a
    message is read.  Once read, it will handle it accordingly by calling
    self._handle_message().

    This is repeated as long as predicate() returns False.  There is some
    locking used internally here so that we don't end up with multiple threads
    blocked on a call to read_for_stream when another thread has read the
    message that caused predicate() to become True.

    Args:
      predicate: Callable, keep reading messages until it returns true.  Note
        that predicate() should not block, as doing so may cause this method to
        hang beyond its timeout.
      timeout: Timeout to use for this call.

    Raises:
      AdbStreamClosedError: If this stream is already closed.

def _read_messages_until_true(self, predicate, timeout):
    """Read a message from this stream and handle it.

    This method tries to read a message from this stream, blocking until a
    message is read.  Once read, it will handle it accordingly by calling
    self._handle_message().

    This is repeated as long as predicate() returns False.  There is some
    locking used internally here so that we don't end up with multiple threads
    blocked on a call to read_for_stream when another thread has read the
    message that caused predicate() to become True.

    Args:
      predicate: Callable, keep reading messages until it returns true.  Note
        that predicate() should not block, as doing so may cause this method to
        hang beyond its timeout.
      timeout: Timeout to use for this call.

    Raises:
      AdbStreamClosedError: If this stream is already closed.
    """
    while not predicate():
      # Hold the message_received Lock while we try to acquire the reader_lock
      # and waiting on the message_received condition, to prevent another reader
      # thread from notifying the condition between us failing to acquire the
      # reader_lock and waiting on the condition.
      self._message_received.acquire()
      if self._reader_lock.acquire(False):
        try:
          # Release the message_received Lock while we do the read so other
          # threads can wait() on the condition without having to block on
          # acquiring the message_received Lock (we may have a longer timeout
          # than them, so that would be bad).
          self._message_received.release()

          # We are now the thread responsible for reading a message.  Check
          # predicate() to make sure nobody else read a message between our last
          # check and acquiring the reader Lock.
          if predicate():
            return

          # Read and handle a message, using our timeout.
          self._handle_message(
              self.adb_connection.read_for_stream(self, timeout))

          # Notify anyone interested that we handled a message, causing them to
          # check their predicate again.
          with self._message_received:
            self._message_received.notify_all()
        finally:
          self._reader_lock.release()
      else:
        # There is some other thread reading a message.  Since we are already
        # holding the message_received Lock, we can immediately do the wait.
        try:
          self._message_received.wait(timeout.remaining)
          if timeout.has_expired():
            raise usb_exceptions.AdbTimeoutError(
                '%s timed out reading messages.', self)
        finally:
          # Make sure we release this even if an exception occurred.
          self._message_received.release()