/**
 * Registers server handlers on the configured protocol.
 */
private void registerHandlers(RaftServerProtocol protocol) {
  protocol.registerOpenSessionHandler(request -> runOnContextIfReady(() -> role.onOpenSession(request), OpenSessionResponse::builder));
  protocol.registerCloseSessionHandler(request -> runOnContextIfReady(() -> role.onCloseSession(request), CloseSessionResponse::builder));
  protocol.registerKeepAliveHandler(request -> runOnContextIfReady(() -> role.onKeepAlive(request), KeepAliveResponse::builder));
  protocol.registerMetadataHandler(request -> runOnContextIfReady(() -> role.onMetadata(request), MetadataResponse::builder));
  protocol.registerConfigureHandler(request -> runOnContext(() -> role.onConfigure(request)));
  protocol.registerInstallHandler(request -> runOnContext(() -> role.onInstall(request)));
  protocol.registerJoinHandler(request -> runOnContext(() -> role.onJoin(request)));
  protocol.registerReconfigureHandler(request -> runOnContext(() -> role.onReconfigure(request)));
  protocol.registerLeaveHandler(request -> runOnContext(() -> role.onLeave(request)));
  protocol.registerTransferHandler(request -> runOnContext(() -> role.onTransfer(request)));
  protocol.registerAppendHandler(request -> runOnContext(() -> role.onAppend(request)));
  protocol.registerPollHandler(request -> runOnContext(() -> role.onPoll(request)));
  protocol.registerVoteHandler(request -> runOnContext(() -> role.onVote(request)));
  protocol.registerCommandHandler(request -> runOnContextIfReady(() -> role.onCommand(request), CommandResponse::builder));
  protocol.registerQueryHandler(request -> runOnContextIfReady(() -> role.onQuery(request), QueryResponse::builder));
}
 

@Override
public CompletableFuture<KeepAliveResponse> onKeepAlive(KeepAliveRequest request) {
  raft.checkThread();
  logRequest(request);

  if (raft.getLeader() == null) {
    return CompletableFuture.completedFuture(logResponse(KeepAliveResponse.builder()
        .withStatus(RaftResponse.Status.ERROR)
        .withError(RaftError.Type.NO_LEADER)
        .build()));
  } else {
    return forward(request, raft.getProtocol()::keepAlive)
        .exceptionally(error -> KeepAliveResponse.builder()
            .withStatus(RaftResponse.Status.ERROR)
            .withError(RaftError.Type.NO_LEADER)
            .build())
        .thenApply(this::logResponse);
  }
}
 

/**
 * Sends a keep alive request to the given node.
 *
 * @param request the request to send
 * @return a future to be completed with the response
 */
public CompletableFuture<KeepAliveResponse> keepAlive(KeepAliveRequest request) {
  CompletableFuture<KeepAliveResponse> future = new CompletableFuture<>();
  if (context.isCurrentContext()) {
    sendRequest(request, protocol::keepAlive, future);
  } else {
    context.execute(() -> sendRequest(request, protocol::keepAlive, future));
  }
  return future;
}
 

@Override
public CompletableFuture<KeepAliveResponse> onKeepAlive(KeepAliveRequest request) {
  logRequest(request);
  return Futures.completedFuture(logResponse(KeepAliveResponse.builder()
      .withStatus(Status.ERROR)
      .withError(RaftError.Type.UNAVAILABLE)
      .build()));
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId,
                                                      KeepAliveRequest request) {
  return sendAndReceive(memberId, "keep-alive", request);
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId,
                                                      KeepAliveRequest request) {
  return getServer(memberId).thenCompose(protocol ->
      protocol.keepAlive(encode(request))).thenApply(this::decode);
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId,
                                                      KeepAliveRequest request) {
  return sendAndReceive(memberId, "keep-alive", request);
}
 

@Override
public void registerKeepAliveHandler(Function<KeepAliveRequest,
    CompletableFuture<KeepAliveResponse>> handler) {
  registerHandler("keep-alive", handler);
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId,
                                                      KeepAliveRequest request) {
  return getServer(memberId).thenCompose(listener ->
      listener.keepAlive(encode(request))).thenApply(this::decode);
}
 

@Override
public void registerKeepAliveHandler(Function<KeepAliveRequest,
    CompletableFuture<KeepAliveResponse>> handler) {
  this.keepAliveHandler = handler;
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId,
                                                      KeepAliveRequest request) {
  return sendAndReceive(memberId, "keep-alive", request);
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId,
                                                      KeepAliveRequest request) {
  return getServer(memberId).thenCompose(protocol ->
      protocol.keepAlive(encode(request))).thenApply(this::decode);
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId,
                                                      KeepAliveRequest request) {
  return sendAndReceive(memberId, "keep-alive", request);
}
 

@Override
public void registerKeepAliveHandler(Function<KeepAliveRequest,
    CompletableFuture<KeepAliveResponse>> handler) {
  registerHandler("keep-alive", handler);
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId,
                                                      KeepAliveRequest request) {
  return getServer(memberId).thenCompose(listener ->
      listener.keepAlive(encode(request))).thenApply(this::decode);
}
 

@Override
public void registerKeepAliveHandler(Function<KeepAliveRequest,
    CompletableFuture<KeepAliveResponse>> handler) {
  this.keepAliveHandler = handler;
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId, KeepAliveRequest request) {
  return sendAndReceive(context.keepAliveSubject, request, memberId);
}
 

@Override
public void registerKeepAliveHandler(Function<KeepAliveRequest, CompletableFuture<KeepAliveResponse>> handler) {
  clusterCommunicator.subscribe(context.keepAliveSubject, serializer::decode, handler, serializer::encode);
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId, KeepAliveRequest request) {
  return sendAndReceive(context.keepAliveSubject, request, memberId);
}
 

@Override
public CompletableFuture<KeepAliveResponse> onKeepAlive(KeepAliveRequest request) {
  final long term = raft.getTerm();
  final long timestamp = System.currentTimeMillis();

  raft.checkThread();
  logRequest(request);

  CompletableFuture<KeepAliveResponse> future = new CompletableFuture<>();
  appendAndCompact(new KeepAliveEntry(term, timestamp, request.sessionIds(), request.commandSequenceNumbers(), request.eventIndexes()))
      .whenCompleteAsync((entry, error) -> {
        if (error != null) {
          future.complete(logResponse(KeepAliveResponse.builder()
              .withStatus(RaftResponse.Status.ERROR)
              .withLeader(raft.getCluster().getMember().memberId())
              .withError(RaftError.Type.PROTOCOL_ERROR)
              .build()));
          return;
        }

        appender.appendEntries(entry.index()).whenComplete((commitIndex, commitError) -> {
          raft.checkThread();
          if (isRunning()) {
            if (commitError == null) {
              raft.getServiceManager().<long[]>apply(entry.index()).whenCompleteAsync((sessionResult, sessionError) -> {
                if (sessionError == null) {
                  future.complete(logResponse(KeepAliveResponse.builder()
                      .withStatus(RaftResponse.Status.OK)
                      .withLeader(raft.getCluster().getMember().memberId())
                      .withMembers(raft.getCluster().getMembers().stream()
                          .map(RaftMember::memberId)
                          .filter(m -> m != null)
                          .collect(Collectors.toList()))
                      .withSessionIds(sessionResult)
                      .build()));
                } else if (sessionError instanceof CompletionException && sessionError.getCause() instanceof RaftException) {
                  future.complete(logResponse(KeepAliveResponse.builder()
                      .withStatus(RaftResponse.Status.ERROR)
                      .withLeader(raft.getCluster().getMember().memberId())
                      .withError(((RaftException) sessionError.getCause()).getType(), sessionError.getMessage())
                      .build()));
                } else if (sessionError instanceof RaftException) {
                  future.complete(logResponse(KeepAliveResponse.builder()
                      .withStatus(RaftResponse.Status.ERROR)
                      .withLeader(raft.getCluster().getMember().memberId())
                      .withError(((RaftException) sessionError).getType(), sessionError.getMessage())
                      .build()));
                } else {
                  future.complete(logResponse(KeepAliveResponse.builder()
                      .withStatus(RaftResponse.Status.ERROR)
                      .withLeader(raft.getCluster().getMember().memberId())
                      .withError(RaftError.Type.PROTOCOL_ERROR, sessionError.getMessage())
                      .build()));
                }
              }, raft.getThreadContext());
            } else {
              future.complete(logResponse(KeepAliveResponse.builder()
                  .withStatus(RaftResponse.Status.ERROR)
                  .withLeader(raft.getCluster().getMember().memberId())
                  .withError(RaftError.Type.PROTOCOL_ERROR)
                  .build()));
            }
          } else {
            RaftMember leader = raft.getLeader();
            future.complete(logResponse(KeepAliveResponse.builder()
                .withStatus(RaftResponse.Status.ERROR)
                .withLeader(leader != null ? leader.memberId() : null)
                .withError(RaftError.Type.ILLEGAL_MEMBER_STATE)
                .build()));
          }
        });
      }, raft.getThreadContext());

  return future;
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId, KeepAliveRequest request) {
  return sendAndReceive(memberId, "keep-alive", request);
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId, KeepAliveRequest request) {
  return getServer(memberId).thenCompose(protocol -> protocol.keepAlive(encode(request))).thenApply(this::decode);
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId, KeepAliveRequest request) {
  return sendAndReceive(memberId, "keep-alive", request);
}
 

@Override
public void registerKeepAliveHandler(Function<KeepAliveRequest, CompletableFuture<KeepAliveResponse>> handler) {
  registerHandler("keep-alive", handler);
}
 

@Override
public CompletableFuture<KeepAliveResponse> keepAlive(MemberId memberId, KeepAliveRequest request) {
  return getServer(memberId).thenCompose(listener -> listener.keepAlive(encode(request))).thenApply(this::decode);
}
 

@Override
public void registerKeepAliveHandler(Function<KeepAliveRequest, CompletableFuture<KeepAliveResponse>> handler) {
  this.keepAliveHandler = handler;
}
 

/**
 * Handles a keep alive request.
 *
 * @param request The request to handle.
 * @return A completable future to be completed with the request response.
 */
CompletableFuture<KeepAliveResponse> onKeepAlive(KeepAliveRequest request);