Skip to content

Commit

Permalink
[HWKMETRICS-74] first pass at RxJava migration
Browse files Browse the repository at this point in the history 
Methods for reading/writing gauge metric data have been updated so that they
now use Observable instead of ListenableFuture. I had to comment some stuff out
like in InfluxSeriesHandler.java while I focus on getting a subset of tests
passing.
  • Loading branch information
John Sanda committed May 14, 2015
1 parent 1032472 commit bfcc952
Show file tree
Hide file tree
Showing 14 changed files with 471 additions and 328 deletions.
132 changes: 77 additions & 55 deletions api/metrics-api-jaxrs/src/main/java/org/hawkular/metrics/api/jaxrs/handler/GaugeHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,10 @@
import static javax.ws.rs.core.MediaType.APPLICATION_JSON;
import static org.hawkular.metrics.api.jaxrs.util.ApiUtils.badRequest;
import static org.hawkular.metrics.api.jaxrs.util.ApiUtils.executeAsync;
import static org.hawkular.metrics.api.jaxrs.util.ApiUtils.noContent;

import java.net.URI;
import java.util.Collections;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
Expand All @@ -49,6 +50,14 @@
import javax.ws.rs.core.Response.Status;
import javax.ws.rs.core.UriInfo;

import com.google.common.base.Function;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.wordnik.swagger.annotations.Api;
import com.wordnik.swagger.annotations.ApiOperation;
import com.wordnik.swagger.annotations.ApiParam;
import com.wordnik.swagger.annotations.ApiResponse;
import com.wordnik.swagger.annotations.ApiResponses;
import org.hawkular.metrics.api.jaxrs.ApiError;
import org.hawkular.metrics.api.jaxrs.callback.MetricCreatedCallback;
import org.hawkular.metrics.api.jaxrs.param.Duration;
Expand All @@ -64,15 +73,7 @@
import org.hawkular.metrics.core.api.MetricId;
import org.hawkular.metrics.core.api.MetricType;
import org.hawkular.metrics.core.api.MetricsService;

import com.google.common.base.Function;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.wordnik.swagger.annotations.Api;
import com.wordnik.swagger.annotations.ApiOperation;
import com.wordnik.swagger.annotations.ApiParam;
import com.wordnik.swagger.annotations.ApiResponse;
import com.wordnik.swagger.annotations.ApiResponses;
import rx.Observable;

/**
* @author Stefan Negrea
Expand Down Expand Up @@ -201,17 +202,20 @@ public void addDataForMetric(
@Suspended final AsyncResponse asyncResponse,
@HeaderParam("tenantId") final String tenantId,
@PathParam("id") String id,
@ApiParam(value = "List of datapoints containing timestamp and value", required = true)
@ApiParam(value = "List of data points containing timestamp and value", required = true)
List<GaugeData> data) {
executeAsync(asyncResponse, () -> {
if (data == null) {
return ApiUtils.emptyPayload();
}

if (data.isEmpty()) {
asyncResponse.resume(badRequest(new ApiError("Payload is empty")));
} else {
Gauge metric = new Gauge(tenantId, new MetricId(id));
metric.getData().addAll(data);
ListenableFuture<Void> future = metricsService.addGaugeData(Collections.singletonList(metric));
return Futures.transform(future, ApiUtils.MAP_VOID);
});
Observable<Void> observable = metricsService.addGaugeData(Observable.just(metric));
observable.subscribe(
noArg -> {},
t -> asyncResponse.resume(Response.serverError().entity(new ApiError(t.getMessage())).build()),
() -> asyncResponse.resume(Response.ok().build()));
}
}

@POST
Expand All @@ -223,14 +227,17 @@ public void addDataForMetric(
response = ApiError.class) })
public void addGaugeData(@Suspended final AsyncResponse asyncResponse, @HeaderParam("tenantId") String tenantId,
@ApiParam(value = "List of metrics", required = true) List<Gauge> metrics) {
executeAsync(asyncResponse, () -> {
if (metrics.isEmpty()) {
return Futures.immediateFuture(Response.ok().build());
}

if (metrics.isEmpty()) {
asyncResponse.resume(badRequest(new ApiError("Payload is empty")));
} else {
metrics.forEach(m -> m.setTenantId(tenantId));
ListenableFuture<Void> future = metricsService.addGaugeData(metrics);
return Futures.transform(future, ApiUtils.MAP_VOID);
});
Observable<Void> observable = metricsService.addGaugeData(Observable.from(metrics));
observable.subscribe(
noArg -> {},
t -> asyncResponse.resume(Response.serverError().entity(new ApiError(t.getMessage())).build()),
() -> asyncResponse.resume(Response.ok().build()));
}
}

@GET
Expand Down Expand Up @@ -271,44 +278,59 @@ public void findGaugeData(@Suspended AsyncResponse asyncResponse, @HeaderParam("
@ApiParam(value = "Defaults to now") @QueryParam("end") final Long end,
@ApiParam(value = "Total number of buckets") @QueryParam("buckets") Integer bucketsCount,
@ApiParam(value = "Bucket duration") @QueryParam("bucketDuration") Duration bucketDuration) {
executeAsync(
asyncResponse,
() -> {
long now = System.currentTimeMillis();
long startTime = start == null ? now - EIGHT_HOURS : start;
long endTime = end == null ? now : end;

Gauge metric = new Gauge(tenantId, new MetricId(id));

if (bucketsCount == null && bucketDuration == null) {
ListenableFuture<List<GaugeData>> dataFuture = metricsService.findGaugeData(tenantId,
new MetricId(id), startTime, endTime);
return Futures.transform(dataFuture, ApiUtils.MAP_COLLECTION);
if (bucketsCount == null && bucketDuration == null) {
long now = System.currentTimeMillis();
long startTime = start == null ? now - EIGHT_HOURS : start;
long endTime = end == null ? now : end;
Observable<GaugeData> observable = metricsService.findGaugeData(tenantId, new MetricId(id), startTime,
endTime);
List<GaugeData> dataPoints = new ArrayList<>();
observable.subscribe(
dataPoints::add,
t -> asyncResponse.resume(Response.serverError().entity(new ApiError(t.getMessage())).build()),
() -> {
if (dataPoints.isEmpty()) {
asyncResponse.resume(noContent());
} else {
asyncResponse.resume(Response.ok(dataPoints).build());
}
}
);
} else {
executeAsync(
asyncResponse,
() -> {
long now = System.currentTimeMillis();
long startTime = start == null ? now - EIGHT_HOURS : start;
long endTime = end == null ? now : end;

if (bucketsCount != null && bucketDuration != null) {
return badRequest(new ApiError("Both buckets and bucketDuration parameters are used"));
}
Gauge metric = new Gauge(tenantId, new MetricId(id));

Buckets buckets;
try {
if (bucketsCount != null) {
buckets = Buckets.fromCount(startTime, endTime, bucketsCount);
} else {
buckets = Buckets.fromStep(startTime, endTime, bucketDuration.toMillis());
if (bucketsCount != null && bucketDuration != null) {
return badRequest(new ApiError("Both buckets and bucketDuration parameters are used"));
}
} catch (IllegalArgumentException e) {
return badRequest(new ApiError("Bucket: " + e.getMessage()));
}

ListenableFuture<BucketedOutput<GaugeBucketDataPoint>> dataFuture;
dataFuture = metricsService.findGaugeStats(metric, startTime, endTime, buckets);
Buckets buckets;
try {
if (bucketsCount != null) {
buckets = Buckets.fromCount(startTime, endTime, bucketsCount);
} else {
buckets = Buckets.fromStep(startTime, endTime, bucketDuration.toMillis());
}
} catch (IllegalArgumentException e) {
return badRequest(new ApiError("Bucket: " + e.getMessage()));
}

ListenableFuture<List<GaugeBucketDataPoint>> outputFuture;
outputFuture = Futures.transform(dataFuture, BucketedOutput<GaugeBucketDataPoint>::getData);
ListenableFuture<BucketedOutput<GaugeBucketDataPoint>> dataFuture;
dataFuture = metricsService.findGaugeStats(metric, startTime, endTime, buckets);

return Futures.transform(outputFuture, ApiUtils.MAP_COLLECTION);
});
ListenableFuture<List<GaugeBucketDataPoint>> outputFuture;
outputFuture = Futures.transform(dataFuture, BucketedOutput<GaugeBucketDataPoint>::getData);

return Futures.transform(outputFuture, ApiUtils.MAP_COLLECTION);
});
}
}

@GET
Expand Down
58 changes: 28 additions & 30 deletions ...metrics-api-jaxrs/src/main/java/org/hawkular/metrics/api/jaxrs/handler/MetricHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
import static org.hawkular.metrics.api.jaxrs.util.ApiUtils.badRequest;
import static org.hawkular.metrics.api.jaxrs.util.ApiUtils.executeAsync;

import java.util.ArrayList;
import java.util.List;

import javax.inject.Inject;
Expand All @@ -33,14 +32,6 @@
import javax.ws.rs.QueryParam;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.container.Suspended;
import javax.ws.rs.core.Response;

import org.hawkular.metrics.api.jaxrs.ApiError;
import org.hawkular.metrics.api.jaxrs.request.MixedMetricsRequest;
import org.hawkular.metrics.api.jaxrs.util.ApiUtils;
import org.hawkular.metrics.core.api.Metric;
import org.hawkular.metrics.core.api.MetricType;
import org.hawkular.metrics.core.api.MetricsService;

import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
Expand All @@ -49,6 +40,12 @@
import com.wordnik.swagger.annotations.ApiParam;
import com.wordnik.swagger.annotations.ApiResponse;
import com.wordnik.swagger.annotations.ApiResponses;
import org.hawkular.metrics.api.jaxrs.ApiError;
import org.hawkular.metrics.api.jaxrs.request.MixedMetricsRequest;
import org.hawkular.metrics.api.jaxrs.util.ApiUtils;
import org.hawkular.metrics.core.api.Metric;
import org.hawkular.metrics.core.api.MetricType;
import org.hawkular.metrics.core.api.MetricsService;


/**
Expand Down Expand Up @@ -105,27 +102,28 @@ public void findMetrics(
response = ApiError.class) })
public void addMetricsData(@Suspended final AsyncResponse asyncResponse, @HeaderParam("tenantId") String tenantId,
@ApiParam(value = "List of metrics", required = true) MixedMetricsRequest metricsRequest) {
executeAsync(asyncResponse, () -> {
if ((metricsRequest.getGaugeMetrics() == null || !metricsRequest.getGaugeMetrics().isEmpty())
&& (metricsRequest.getAvailabilityMetrics() == null || metricsRequest.getAvailabilityMetrics()
.isEmpty())) {
return Futures.immediateFuture(Response.ok().build());
}

List<ListenableFuture<Void>> simpleFuturesList = new ArrayList<>();

if (metricsRequest.getGaugeMetrics() != null && !metricsRequest.getGaugeMetrics().isEmpty()) {
metricsRequest.getGaugeMetrics().forEach(m -> m.setTenantId(tenantId));
simpleFuturesList.add(metricsService.addGaugeData(metricsRequest.getGaugeMetrics()));
}

if (metricsRequest.getAvailabilityMetrics() != null && !metricsRequest.getAvailabilityMetrics().isEmpty()) {
metricsRequest.getAvailabilityMetrics().forEach(m -> m.setTenantId(tenantId));
simpleFuturesList.add(metricsService.addAvailabilityData(metricsRequest.getAvailabilityMetrics()));
}

return Futures.transform(Futures.successfulAsList(simpleFuturesList), ApiUtils.MAP_LIST_VOID);
});
// executeAsync(asyncResponse, () -> {
// if ((metricsRequest.getGaugeMetrics() == null || !metricsRequest.getGaugeMetrics().isEmpty())
// && (metricsRequest.getAvailabilityMetrics() == null || metricsRequest.getAvailabilityMetrics()
// .isEmpty())) {
// return Futures.immediateFuture(Response.ok().build());
// }
//
// List<ListenableFuture<Void>> simpleFuturesList = new ArrayList<>();
//
// if (metricsRequest.getGaugeMetrics() != null && !metricsRequest.getGaugeMetrics().isEmpty()) {
// metricsRequest.getGaugeMetrics().forEach(m -> m.setTenantId(tenantId));
// simpleFuturesList.add(metricsService.addGaugeData(metricsRequest.getGaugeMetrics()));
// }
//
// if (metricsRequest.getAvailabilityMetrics() != null &&
// !metricsRequest.getAvailabilityMetrics().isEmpty()) {
// metricsRequest.getAvailabilityMetrics().forEach(m -> m.setTenantId(tenantId));
// simpleFuturesList.add(metricsService.addAvailabilityData(metricsRequest.getAvailabilityMetrics()));
// }
//
// return Futures.transform(Futures.successfulAsList(simpleFuturesList), ApiUtils.MAP_LIST_VOID);
// });
}

}

0 comments on commit bfcc952

Please sign in to comment.