RaftClientImpl.java 文件源码

private CompletableFuture<RaftClientReply> sendAsync(Message message,
    boolean readOnly) {
  Objects.requireNonNull(message, "message == null");
  try {
    asyncRequestSemaphore.acquire();
  } catch (InterruptedException e) {
    throw new CompletionException(IOUtils.toInterruptedIOException(
        "Interrupted when sending " + message, e));
  }
  final long callId = nextCallId();
  final LongFunction<PendingAsyncRequest> constructor = seqNum -> new PendingAsyncRequest(seqNum,
      seq -> new RaftClientRequest(clientId, leaderId, groupId, callId, seq, message, readOnly));
  return slidingWindow.submitNewRequest(constructor, this::sendRequestWithRetryAsync
  ).getReplyFuture(
  ).thenApply(reply -> handleStateMachineException(reply, CompletionException::new)
  ).whenComplete((r, e) -> asyncRequestSemaphore.release());
}