@Override
  public void onNext(T value) {
    synchronized (lock) {
      if (++numMessages >= maxMessagesBeforeCheck) {
        numMessages = 0;
        int waitSeconds = 1;
        int totalSecondsWaited = 0;
        int phase = phaser.getPhase();
        // Record the initial phase in case we are in the inbound gRPC thread where the phase won't
        // advance.
        int initialPhase = phase;
        // A negative phase indicates that the phaser is terminated.
        while (phase >= 0 && !outboundObserver.isReady()) {
          try {
            phase = phaser.awaitAdvanceInterruptibly(phase, waitSeconds, TimeUnit.SECONDS);
          } catch (TimeoutException e) {
            totalSecondsWaited += waitSeconds;
            // Double the backoff for re-evaluating the isReady bit up to a maximum of once per
            // minute. This bounds the waiting if the onReady callback is not called as expected.
            waitSeconds = Math.min(waitSeconds * 2, 60);
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
          }
        }
        if (totalSecondsWaited > 0) {
          // If the phase didn't change, this means that the installed onReady callback had not
          // been invoked.
          if (initialPhase == phase) {
            LOG.info(
                "Output channel stalled for {}s, outbound thread {}. OnReady notification was "
                    + "not invoked, ensure the inbound gRPC thread is not used for output.",
                totalSecondsWaited,
                Thread.currentThread().getName());
          } else if (totalSecondsWaited > 60) {
            LOG.warn(
                "Output channel stalled for {}s, outbound thread {}.",
                totalSecondsWaited,
                Thread.currentThread().getName());
          } else {
            LOG.debug(
                "Output channel stalled for {}s, outbound thread {}.",
                totalSecondsWaited,
                Thread.currentThread().getName());
          }
        }
      }
      outboundObserver.onNext(value);
    }
  }
@Test
  public void testMessageCheckInterval() throws Exception {
    final AtomicInteger index = new AtomicInteger();
    ArrayListMultimap<Integer, String> values = ArrayListMultimap.create();
    final DirectStreamObserver<String> streamObserver =
        new DirectStreamObserver<>(
            new AdvancingPhaser(1),
            TestStreams.withOnNext((String t) -> assertTrue(values.put(index.get(), t)))
                .withIsReady(
                    () -> {
                      index.incrementAndGet();
                      return true;
                    })
                .build(),
            10);

    List<String> prefixes = ImmutableList.of("0", "1", "2", "3", "4");
    List<Future<String>> results = new ArrayList<>();
    for (final String prefix : prefixes) {
      results.add(
          executor.submit(
              () -> {
                for (int i = 0; i < 10; i++) {
                  streamObserver.onNext(prefix + i);
                }
                return prefix;
              }));
    }
    for (Future<?> result : results) {
      result.get();
    }
    assertEquals(50, values.size());
    for (Collection<String> valuesPerMessageCheck : values.asMap().values()) {
      assertThat(valuesPerMessageCheck, hasSize(10));
    }

    // Check that order was maintained per writer.
    int[] prefixesIndex = new int[prefixes.size()];
    for (String onNextValue : values.values()) {
      int prefix = Integer.parseInt(onNextValue.substring(0, 1));
      int suffix = Integer.parseInt(onNextValue.substring(1, 2));
      assertEquals(prefixesIndex[prefix], suffix);
      prefixesIndex[prefix] += 1;
    }
  }