util: add toLeaseTimeToLiveResponse() and listenableToCompletableFuture()
toLeaseTimeToLiveResponse() converts grpc's LeaseTimeToLiveResponse to Lease's LeaseTimeToLiveResponse.
listenableToCompletableFuture() converts a listenableFuture to CompletableFuture.
@@ -5,10 +5,15 @@ import com.coreos.jetcd.api.ResponseHeader;
import com.coreos.jetcd.data.ByteSequence;
import com.coreos.jetcd.data.Header;
import com.coreos.jetcd.data.KeyValue;
+import com.coreos.jetcd.lease.LeaseTimeToLiveResponse;
import com.coreos.jetcd.watch.WatchEvent;
+import com.google.common.util.concurrent.ListenableFuture;
import com.google.protobuf.ByteString;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
/**
* This util is to convert api class to client class.
@@ -79,4 +84,48 @@ class Util {
return new Header(header.getClusterId(), header.getMemberId(), header.getRevision(),
header.getRaftTerm(), compactRevision);
}
+
+ /**
+ * convert API LeaseTimeToLiveResponse to self defined LeaseTimeToLiveResponse.
+ */
+ static LeaseTimeToLiveResponse toLeaseTimeToLiveResponse(
+ com.coreos.jetcd.api.LeaseTimeToLiveResponse response) {
+ List<ByteSequence> byteSequenceKeys = response.getKeysList().stream()
+ .map(byteStringKey -> ByteSequence.fromBytes(byteStringKey.toByteArray()))
+ .collect(Collectors.toList());
+ return new LeaseTimeToLiveResponse(apiToClientHeader(response.getHeader(), 0),
+ response.getID(), response.getTTL(), response.getGrantedTTL(), byteSequenceKeys);
+ }
+
+ /**
+ * convert ListenableFuture of Type S to CompletableFuture of Type T.
+ */
+ static <S, T> CompletableFuture<T> listenableToCompletableFuture(
+ final ListenableFuture<S> sourceFuture, final FutureResultConvert<S, T> resultConvert,
+ Executor executor) {
+ CompletableFuture<T> targetFuture = new CompletableFuture<T>() {
+ // the cancel of targetFuture also cancels the sourceFuture.
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ super.cancel(mayInterruptIfRunning);
+ return sourceFuture.cancel(mayInterruptIfRunning);
+ }
+ };
+
+ sourceFuture.addListener(() -> {
+ try {
+ targetFuture.complete(resultConvert.convert(sourceFuture.get()));
+ } catch (Exception e) {
+ targetFuture.completeExceptionally(e);
+ }
+ }, executor);
+
+ return targetFuture;
+ }
+
+ @FunctionalInterface
+ interface FutureResultConvert<S, T> {
+
+ T convert(S source);
+ }
}
