public void send(ReqT request, long timeoutMs) throws IOException {
    if (mClosed || mCanceled || mClosedFromRemote) {
      throw new CancelledException(formatErrorMessage(
          "Failed to send request %s: stream is already closed or cancelled. clientClosed: %s "
              + "clientCancelled: %s serverClosed: %s",
          LogUtils.truncateMessageLineLength(request), mClosed, mCanceled, mClosedFromRemote));
    }
    try (LockResource ignored = new LockResource(mLock)) {
      long startMs = System.currentTimeMillis();
      while (true) {
        checkError();
        if (mRequestObserver.isReady()) {
          break;
        }

        long waitedForMs = System.currentTimeMillis() - startMs;
        if (waitedForMs >= timeoutMs) {
          throw new DeadlineExceededException(formatErrorMessage(
              "Timeout sending request %s after %dms. clientClosed: %s clientCancelled: %s "
                  + "serverClosed: %s",
              LogUtils.truncateMessageLineLength(request), timeoutMs, mClosed, mCanceled,
              mClosedFromRemote));
        }

        try {
          // Wait for a minute max
          long awaitMs = Math.min(timeoutMs - waitedForMs, Constants.MINUTE_MS);
          if (!mReadyOrFailed.await(awaitMs, TimeUnit.MILLISECONDS)) {
            // Log a warning before looping again
            LOG.warn(
                "Stream is not ready for client to send request, will wait again. totalWaitMs: {} "
                    + "clientClosed: {} clientCancelled: {} serverClosed: {} description: {}",
                System.currentTimeMillis() - startMs, mClosed, mCanceled, mClosedFromRemote,
                mDescription);
          }
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          throw new CancelledException(
              formatErrorMessage("Failed to send request %s: interrupted while waiting for server.",
                  LogUtils.truncateMessageLineLength(request)), e);
        }
      }
    }
    mRequestObserver.onNext(request);
  }
@Test
  public void send() throws Exception {
    WriteRequest request = WriteRequest.newBuilder().build();

    mStream.send(request, TIMEOUT);

    verify(mRequestObserver).onNext(request);
  }