Java 类io.grpc.internal.LogExceptionRunnable 实例源码

/**
 * 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;
        }
    }
}

/**
 * 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);
}

private synchronized void run() {
    if (resolutionTask != null) {
        resolutionTask.cancel(false);
        resolutionTask = null;
    }

    if (shutdown) {
        return;
    }

    checkNotNull(listener, "resolver not started");
    checkNotNull(timerService, "resolver not started");
    checkNotNull(executor, "resolver not started");

    resolving = true;
    try {
        final Response<List<CatalogService>> response = tag
                .map(tag -> catalogClient.getCatalogService(serviceName, tag, QueryParams.DEFAULT))
                .orElseGet(() -> catalogClient.getCatalogService(serviceName, QueryParams.DEFAULT));

        final List<EquivalentAddressGroup> servers = response.getValue().stream()
                .map((service) -> {
                    // use service address then fall back to address
                    String host = service.getServiceAddress();
                    if (Strings.isNullOrEmpty(host)) {
                        host = service.getAddress();
                    }

                    final int port = service.getServicePort();

                    final SocketAddress address = new InetSocketAddress(host, port);

                    return new EquivalentAddressGroup(address);
                })
                .collect(Collectors.toList());

        listener.onAddresses(servers, Attributes.EMPTY);
    } catch (final Exception e) {
        if (shutdown) {
            return;
        }

        resolutionTask = timerService.schedule(
                new LogExceptionRunnable(this::run),
                retryInterval, retryIntervalTimeUnit
        );

        listener.onError(Status.UNAVAILABLE.withCause(e));
    } finally {
        resolving = false;
    }
}