Skip to content

Commit

Permalink
[refactor][java-client] Reduce code duplication in GET requests in ad…
Browse files Browse the repository at this point in the history 
…min client (#17023)
  • Loading branch information
Andras Beni authored Sep 14, 2022
1 parent 99b52eb commit d14b3e3
Show file tree
Hide file tree
Showing 10 changed files with 88 additions and 578 deletions.
35 changes: 35 additions & 0 deletions pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import java.util.function.Supplier;
import javax.ws.rs.ClientErrorException;
import javax.ws.rs.ServerErrorException;
Expand All @@ -34,6 +35,7 @@
import javax.ws.rs.client.Invocation.Builder;
import javax.ws.rs.client.InvocationCallback;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.GenericType;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import org.apache.pulsar.client.admin.PulsarAdminException;
Expand Down Expand Up @@ -186,6 +188,39 @@ public <T> CompletableFuture<T> asyncGetRequest(final WebTarget target, FutureCa
return callback.future();
}

protected <T> CompletableFuture<T> asyncGetRequest(final WebTarget target, Class<? extends T> type) {
return asyncGetRequest(target, response -> response.readEntity(type));
}

protected <T> CompletableFuture<T> asyncGetRequest(final WebTarget target, GenericType<T> type) {
return asyncGetRequest(target, response -> response.readEntity(type));
}

private <T> CompletableFuture<T> asyncGetRequest(final WebTarget target, Function<Response, T> readResponse) {
final CompletableFuture<T> future = new CompletableFuture<>();
asyncGetRequest(target,
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
try {
future.complete(readResponse.apply(response));
} catch (Exception e) {
future.completeExceptionally(getApiException(e));
}
}
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
}

public CompletableFuture<Void> asyncDeleteRequest(final WebTarget target) {
final CompletableFuture<Void> future = new CompletableFuture<>();
try {
Expand Down
14 changes: 1 addition & 13 deletions pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -162,19 +162,7 @@ public void backlogQuotaCheck() throws PulsarAdminException {
@Override
public CompletableFuture<Void> backlogQuotaCheckAsync() {
WebTarget path = adminBrokers.path("backlogQuotaCheck");
final CompletableFuture<Void> future = new CompletableFuture<>();
asyncGetRequest(path, new InvocationCallback<Void>() {
@Override
public void completed(Void unused) {
future.complete(null);
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(throwable);
}
});
return future;
return asyncGetRequest(path, new FutureCallback<Void>() {});
}

@Override
Expand Down
53 changes: 6 additions & 47 deletions pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ClustersImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.InvocationCallback;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.MediaType;
import org.apache.pulsar.client.admin.Clusters;
Expand Down Expand Up @@ -140,22 +139,9 @@ public Map<String, NamespaceIsolationData> getNamespaceIsolationPolicies(String
public CompletableFuture<Map<String, NamespaceIsolationData>> getNamespaceIsolationPoliciesAsync(
String cluster) {
WebTarget path = adminClusters.path(cluster).path("namespaceIsolationPolicies");
final CompletableFuture<Map<String, NamespaceIsolationData>> future = new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<Map<String, NamespaceIsolationDataImpl>>() {
@Override
public void completed(Map<String, NamespaceIsolationDataImpl> stringNamespaceIsolationDataMap) {
Map<String, NamespaceIsolationData> result = new HashMap<>();
stringNamespaceIsolationDataMap.forEach(result::put);
future.complete(result);
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
return asyncGetRequest(path, new FutureCallback<Map<String, NamespaceIsolationDataImpl>>() {})
.thenApply(HashMap::new);
}

@Override
Expand All @@ -168,22 +154,8 @@ public List<BrokerNamespaceIsolationData> getBrokersWithNamespaceIsolationPolicy
public CompletableFuture<List<BrokerNamespaceIsolationData>> getBrokersWithNamespaceIsolationPolicyAsync(
String cluster) {
WebTarget path = adminClusters.path(cluster).path("namespaceIsolationPolicies").path("brokers");
final CompletableFuture<List<BrokerNamespaceIsolationData>> future = new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<List<BrokerNamespaceIsolationDataImpl>>() {
@Override
public void completed(List<BrokerNamespaceIsolationDataImpl> brokerNamespaceIsolationData) {
List<BrokerNamespaceIsolationData> data =
new ArrayList<>(brokerNamespaceIsolationData);
future.complete(data);
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
return asyncGetRequest(path, new FutureCallback<List<BrokerNamespaceIsolationDataImpl>>() {})
.thenApply(ArrayList::new);
}

@Override
Expand Down Expand Up @@ -303,21 +275,8 @@ public Map<String, FailureDomain> getFailureDomains(String cluster) throws Pulsa
@Override
public CompletableFuture<Map<String, FailureDomain>> getFailureDomainsAsync(String cluster) {
WebTarget path = adminClusters.path(cluster).path("failureDomains");
final CompletableFuture<Map<String, FailureDomain>> future = new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<Map<String, FailureDomainImpl>>() {
@Override
public void completed(Map<String, FailureDomainImpl> failureDomains) {
Map<String, FailureDomain> result = new HashMap<>(failureDomains);
future.complete(result);
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
return asyncGetRequest(path, new FutureCallback<Map<String, FailureDomainImpl>>() {})
.thenApply(HashMap::new);
}

@Override
Expand Down
137 changes: 7 additions & 130 deletions pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -87,25 +87,7 @@ public List<String> getFunctions(String tenant, String namespace) throws PulsarA
@Override
public CompletableFuture<List<String>> getFunctionsAsync(String tenant, String namespace) {
WebTarget path = functions.path(tenant).path(namespace);
final CompletableFuture<List<String>> future = new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
List<String> functions = response.readEntity(new GenericType<List<String>>() {});
future.complete(functions);
}
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
return asyncGetRequest(path, new GenericType<List<String>>() {});
}

@Override
Expand All @@ -116,24 +98,7 @@ public FunctionConfig getFunction(String tenant, String namespace, String functi
@Override
public CompletableFuture<FunctionConfig> getFunctionAsync(String tenant, String namespace, String function) {
WebTarget path = functions.path(tenant).path(namespace).path(function);
final CompletableFuture<FunctionConfig> future = new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(FunctionConfig.class));
}
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
return asyncGetRequest(path, FunctionConfig.class);
}

@Override
Expand All @@ -145,24 +110,7 @@ public FunctionStatus getFunctionStatus(
@Override
public CompletableFuture<FunctionStatus> getFunctionStatusAsync(String tenant, String namespace, String function) {
WebTarget path = functions.path(tenant).path(namespace).path(function).path("status");
final CompletableFuture<FunctionStatus> future = new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(FunctionStatus.class));
}
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
return asyncGetRequest(path, FunctionStatus.class);
}

@Override
Expand All @@ -176,26 +124,7 @@ public CompletableFuture<FunctionStatus.FunctionInstanceStatus.FunctionInstanceS
String tenant, String namespace, String function, int id) {
WebTarget path =
functions.path(tenant).path(namespace).path(function).path(Integer.toString(id)).path("status");
final CompletableFuture<FunctionStatus.FunctionInstanceStatus.FunctionInstanceStatusData> future =
new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(
FunctionStatus.FunctionInstanceStatus.FunctionInstanceStatusData.class));
}
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
return asyncGetRequest(path, FunctionStatus.FunctionInstanceStatus.FunctionInstanceStatusData.class);
}

@Override
Expand All @@ -208,25 +137,7 @@ public FunctionInstanceStatsData getFunctionStats(
public CompletableFuture<FunctionInstanceStatsData> getFunctionStatsAsync(
String tenant, String namespace, String function, int id) {
WebTarget path = functions.path(tenant).path(namespace).path(function).path(Integer.toString(id)).path("stats");
final CompletableFuture<FunctionInstanceStatsData> future =
new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(FunctionInstanceStatsDataImpl.class));
}
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
return asyncGetRequest(path, FunctionInstanceStatsDataImpl.class);
}

@Override
Expand All @@ -239,24 +150,7 @@ public FunctionStats getFunctionStats(String tenant, String namespace, String fu
public CompletableFuture<FunctionStats> getFunctionStatsAsync(String tenant,
String namespace, String function) {
WebTarget path = functions.path(tenant).path(namespace).path(function).path("stats");
final CompletableFuture<FunctionStats> future = new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(FunctionStatsImpl.class));
}
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
return asyncGetRequest(path, FunctionStatsImpl.class);
}

@Override
Expand Down Expand Up @@ -788,24 +682,7 @@ public FunctionState getFunctionState(String tenant, String namespace, String fu
public CompletableFuture<FunctionState> getFunctionStateAsync(
String tenant, String namespace, String function, String key) {
WebTarget path = functions.path(tenant).path(namespace).path(function).path("state").path(key);
final CompletableFuture<FunctionState> future = new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(FunctionState.class));
}
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
return asyncGetRequest(path, FunctionState.class);
}

@Override
Expand Down
21 changes: 2 additions & 19 deletions pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/LookupImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import javax.ws.rs.client.InvocationCallback;
import javax.ws.rs.client.WebTarget;
import org.apache.pulsar.client.admin.Lookup;
import org.apache.pulsar.client.admin.PulsarAdminException;
Expand Down Expand Up @@ -57,24 +56,8 @@ public CompletableFuture<String> lookupTopicAsync(String topic) {
String prefix = topicName.isV2() ? "/topic" : "/destination";
WebTarget path = v2lookup.path(prefix).path(topicName.getLookupName());

final CompletableFuture<String> future = new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<LookupData>() {
@Override
public void completed(LookupData lookupData) {
if (useTls) {
future.complete(lookupData.getBrokerUrlTls());
} else {
future.complete(lookupData.getBrokerUrl());
}
}

@Override
public void failed(Throwable throwable) {
future.completeExceptionally(getApiException(throwable.getCause()));
}
});
return future;
return asyncGetRequest(path, new FutureCallback<LookupData>() {})
.thenApply(lookupData -> useTls ? lookupData.getBrokerUrlTls() : lookupData.getBrokerUrl());
}

@Override
Expand Down
Loading

0 comments on commit d14b3e3

Please sign in to comment.