/**
 * Requests metadata from the cluster.
 *
 * @return A completable future to be completed with cluster metadata.
 */
private CompletableFuture<MetadataResponse> getMetadata() {
  CompletableFuture<MetadataResponse> future = new CompletableFuture<>();
  connection.metadata(MetadataRequest.builder().build()).whenComplete((response, error) -> {
    if (error == null) {
      if (response.status() == RaftResponse.Status.OK) {
        future.complete(response);
      } else {
        future.completeExceptionally(response.error().createException());
      }
    } else {
      future.completeExceptionally(error);
    }
  });
  return future;
}
 

@Override
public CompletableFuture<MetadataResponse> onMetadata(MetadataRequest request) {
  raft.checkThread();
  logRequest(request);

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

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

@Override
public CompletableFuture<MetadataResponse> onMetadata(MetadataRequest request) {
  raft.checkThread();
  logRequest(request);

  if (transferring) {
    return CompletableFuture.completedFuture(logResponse(MetadataResponse.builder()
        .withStatus(RaftResponse.Status.ERROR)
        .withError(RaftError.Type.ILLEGAL_MEMBER_STATE)
        .build()));
  }

  CompletableFuture<MetadataResponse> future = new CompletableFuture<>();
  Indexed<MetadataEntry> entry = new Indexed<>(
      raft.getLastApplied(),
      new MetadataEntry(raft.getTerm(), System.currentTimeMillis(), request.session()), 0);
  raft.getServiceManager().<MetadataResult>apply(entry).whenComplete((result, error) -> {
    if (error == null) {
      future.complete(logResponse(MetadataResponse.builder()
          .withStatus(RaftResponse.Status.OK)
          .withSessions(result.sessions())
          .build()));
    } else {
      future.complete(logResponse(MetadataResponse.builder()
          .withStatus(RaftResponse.Status.ERROR)
          .withError(RaftError.Type.PROTOCOL_ERROR)
          .build()));
    }
  });
  return future;
}
 

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

@Override
public CompletableFuture<MetadataResponse> metadata(MemberId memberId,
                                                    MetadataRequest request) {
  return sendAndReceive(memberId, "metadata", request);
}
 

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

@Override
public CompletableFuture<MetadataResponse> metadata(MemberId memberId,
                                                    MetadataRequest request) {
  return sendAndReceive(memberId, "metadata", request);
}
 

@Override
public void registerMetadataHandler(Function<MetadataRequest,
    CompletableFuture<MetadataResponse>> handler) {
  registerHandler("metadata", handler);
}
 

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

@Override
public void registerMetadataHandler(Function<MetadataRequest,
    CompletableFuture<MetadataResponse>> handler) {
  this.metadataHandler = handler;
}
 

@Override
public CompletableFuture<MetadataResponse> metadata(MemberId memberId,
                                                    MetadataRequest request) {
  return sendAndReceive(memberId, "metadata", request);
}
 

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

@Override
public CompletableFuture<MetadataResponse> metadata(MemberId memberId,
                                                    MetadataRequest request) {
  return sendAndReceive(memberId, "metadata", request);
}
 

@Override
public void registerMetadataHandler(Function<MetadataRequest,
    CompletableFuture<MetadataResponse>> handler) {
  registerHandler("metadata", handler);
}
 

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

@Override
public void registerMetadataHandler(Function<MetadataRequest,
    CompletableFuture<MetadataResponse>> handler) {
  this.metadataHandler = handler;
}
 

@Override
public CompletableFuture<MetadataResponse> metadata(MemberId memberId, MetadataRequest request) {
  return sendAndReceive(context.metadataSubject, request, memberId);
}
 

@Override
public void registerMetadataHandler(Function<MetadataRequest, CompletableFuture<MetadataResponse>> handler) {
  clusterCommunicator.subscribe(context.metadataSubject, serializer::decode, handler, serializer::encode);
}
 

@Override
public CompletableFuture<MetadataResponse> metadata(MemberId memberId, MetadataRequest request) {
  return sendAndReceive(context.metadataSubject, request, memberId);
}
 

@Override
public CompletableFuture<MetadataResponse> metadata(MemberId memberId, MetadataRequest request) {
  return sendAndReceive(memberId, "metadata", request);
}
 

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

@Override
public CompletableFuture<MetadataResponse> metadata(MemberId memberId, MetadataRequest request) {
  return sendAndReceive(memberId, "metadata", request);
}
 

@Override
public void registerMetadataHandler(Function<MetadataRequest, CompletableFuture<MetadataResponse>> handler) {
  registerHandler("metadata", handler);
}
 

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

@Override
public void registerMetadataHandler(Function<MetadataRequest, CompletableFuture<MetadataResponse>> handler) {
  this.metadataHandler = handler;
}
 

/**
 * Handles a metadata request.
 *
 * @param request The request to handle.
 * @return A completable future to be completed with the request response.
 */
CompletableFuture<MetadataResponse> onMetadata(MetadataRequest request);