/**
 * Schedules the next response chunk to be dispatched. If all input has been received and there
 * are no more chunks in the queue, the stream is closed.
 */
private void scheduleNextChunk() {
  synchronized (this) {
    if (scheduled) {
      // Dispatch task is already scheduled.
      return;
    }

    // Schedule the next response chunk if there is one.
    Chunk nextChunk = chunks.peek();
    if (nextChunk != null) {
      scheduled = true;
      // TODO(ejona): cancel future if RPC is cancelled
      Future<?> unused = executor.schedule(new LogExceptionRunnable(dispatchTask),
          nextChunk.delayMicroseconds, TimeUnit.MICROSECONDS);
      return;
    }
  }
}
 

@VisibleForTesting
void start(final ChannelHandlerContext ctx, final ScheduledExecutorService scheduler) {
  this.scheduler = scheduler;
  nextIdleMonitorTime = ticker.nanoTime() + maxConnectionIdleInNanos;

  shutdownTask = new LogExceptionRunnable(new Runnable() {
    @Override
    public void run() {
      if (shutdownDelayed) {
        if (!isActive) {
          // delay shutdown
          shutdownFuture = scheduler.schedule(
              shutdownTask, nextIdleMonitorTime - ticker.nanoTime(), TimeUnit.NANOSECONDS);
          shutdownDelayed = false;
        }
        // if isActive, exit. Will schedule a new shutdownFuture once onTransportIdle
      } else {
        close(ctx);
        shutdownFuture = null;
      }
    }
  });

  shutdownFuture =
      scheduler.schedule(shutdownTask, maxConnectionIdleInNanos, TimeUnit.NANOSECONDS);
}
 

/**
 * Schedules the next response chunk to be dispatched. If all input has been received and there
 * are no more chunks in the queue, the stream is closed.
 */
private void scheduleNextChunk() {
    synchronized (this) {
        if (scheduled) {
            // Dispatch task is already scheduled.
            return;
        }

        // Schedule the next response chunk if there is one.
        final Chunk nextChunk = chunks.peek();
        if (nextChunk != null) {
            scheduled = true;
            // TODO(ejona): cancel future if RPC is cancelled
            final Future<?> unused = executor.schedule(new LogExceptionRunnable(dispatchTask),
                                                       nextChunk.delayMicroseconds,
                                                       TimeUnit.MICROSECONDS);
            return;
        }
    }
}
 

/**
 * Schedules the next response chunk to be dispatched. If all input has been received and there
 * are no more chunks in the queue, the stream is closed.
 */
private void scheduleNextChunk() {
  synchronized (this) {
    if (scheduled) {
      // Dispatch task is already scheduled.
      return;
    }

    // Schedule the next response chunk if there is one.
    Chunk nextChunk = chunks.peek();
    if (nextChunk != null) {
      scheduled = true;
      // TODO(ejona): cancel future if RPC is cancelled
      Future<?> unused = executor.schedule(new LogExceptionRunnable(dispatchTask),
          nextChunk.delayMicroseconds, TimeUnit.MICROSECONDS);
      return;
    }
  }
}
 

@VisibleForTesting
void start(final ChannelHandlerContext ctx, final ScheduledExecutorService scheduler) {
  this.scheduler = scheduler;
  nextIdleMonitorTime = ticker.nanoTime() + maxConnectionIdleInNanos;

  shutdownTask = new LogExceptionRunnable(new Runnable() {
    @Override
    public void run() {
      if (shutdownDelayed) {
        if (!isActive) {
          // delay shutdown
          shutdownFuture = scheduler.schedule(
              shutdownTask, nextIdleMonitorTime - ticker.nanoTime(), TimeUnit.NANOSECONDS);
          shutdownDelayed = false;
        }
        // if isActive, exit. Will schedule a new shutdownFuture once onTransportIdle
      } else {
        close(ctx);
        shutdownFuture = null;
      }
    }
  });

  shutdownFuture =
      scheduler.schedule(shutdownTask, maxConnectionIdleInNanos, TimeUnit.NANOSECONDS);
}