[HWKMETRICS-199] add an endpoint to block until scheduler finishes work

There is a new endpoint to block for a specified duration. On the server side,
we subscribe to the task schedulers time slices observable which emits a
timestamp at the completion of each time slice. That is, a timestamp is emitted
when all of the work for the time slice is finished. A request like,

  GET /hawkulr/metrics/clock/wait?duration=5min

will block the request until the task scheduler has completed the work for five
time slices.

api/metrics-api-jaxrs/src/main/java/org/hawkular/metrics/api/jaxrs/handler/VirtualClockHandler.java

@@ -16,9 +16,11 @@
  */
 package org.hawkular.metrics.api.jaxrs.handler;
 
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static javax.ws.rs.core.MediaType.APPLICATION_JSON;
 
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 import javax.inject.Inject;
 import javax.ws.rs.Consumes;
@@ -27,12 +29,18 @@
 import javax.ws.rs.PUT;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
 import javax.ws.rs.core.Response;
 
+import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableMap;
 import org.hawkular.metrics.api.jaxrs.param.Duration;
 import org.hawkular.metrics.api.jaxrs.util.VirtualClock;
 import org.hawkular.metrics.tasks.api.TaskScheduler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import rx.observers.TestSubscriber;
+import rx.schedulers.Schedulers;
 
 
 /**
@@ -45,14 +53,13 @@ public class VirtualClockHandler {
 
     public static final String PATH = "/clock";
 
-    @Inject
+    private static Logger logger = LoggerFactory.getLogger(VirtualClockHandler.class);
+
     private VirtualClock virtualClock;
 
     @Inject
     private TaskScheduler taskScheduler;
 
-    private static boolean started;
-
     @GET
     public Response getTime() {
         return Response.ok(ImmutableMap.<String, Object>of("now", virtualClock.now())).build();
@@ -62,9 +69,8 @@ public Response getTime() {
     public Response setTime(Map<String, Object> params) {
         Long time = (Long) params.get("time");
         virtualClock.advanceTimeTo(time);
-        if (!started) {
+        if (!taskScheduler.isRunning()) {
             taskScheduler.start();
-            started = true;
         }
         return Response.ok().build();
     }
@@ -75,4 +81,27 @@ public Response incrementTime(Duration duration) {
         return Response.ok().build();
     }
 
+    @GET
+    @Path("/wait")
+    public Response waitForDuration(@QueryParam("duration") Duration duration) {
+        int numMinutes = (int) TimeUnit.MINUTES.convert(duration.getValue(), duration.getTimeUnit());
+        TestSubscriber<Long> timeSlicesSubscriber = new TestSubscriber<>();
+        taskScheduler.getFinishedTimeSlices()
+                .take(numMinutes)
+                .observeOn(Schedulers.immediate())
+                .subscribe(timeSlicesSubscriber);
+
+        try {
+            timeSlicesSubscriber.awaitTerminalEvent(10, SECONDS);
+            timeSlicesSubscriber.assertNoErrors();
+            timeSlicesSubscriber.assertTerminalEvent();
+
+            return Response.ok().build();
+        } catch (Exception e) {
+            logger.warn("Failed to wait " + numMinutes + " minutes for task scheduler to complete work", e);
+            return Response.serverError().entity(ImmutableMap.of("errorMsg", Throwables.getStackTraceAsString(e)))
+                    .build();
+        }
+    }
+
 }

rest-tests/src/test/groovy/org/hawkular/metrics/rest/CountersITest.groovy

@@ -336,7 +336,9 @@ class CountersITest extends RESTTest {
     assertEquals(200, response.status)
 
     setTime(start.plusMinutes(4))
-    Thread.sleep(2500)
+
+    response = hawkularMetrics.get(path: "clock/wait", query: [duration: "4mn"])
+    assertEquals("There was an error waiting: $response.data", 200, response.status)
 
     response = hawkularMetrics.get(
         path: "counters/$counter/rate",

task-queue/src/main/java/org/hawkular/metrics/tasks/api/TaskScheduler.java

@@ -42,6 +42,8 @@ Observable<Task2> scheduleTask(String name, String groupKey, int executionOrder,
 
     void shutdown();
 
+    Observable<Long> getFinishedTimeSlices();
 
+    boolean isRunning();
 
 }

task-queue/src/main/java/org/hawkular/metrics/tasks/impl/TaskSchedulerImpl.java

@@ -101,9 +101,7 @@ public class TaskSchedulerImpl implements TaskScheduler {
 
     private DateTimeService dateTimeService;
 
-    private boolean shutdown;
-
-    private boolean started;
+    private boolean running;
 
     /**
      * A subject to broadcast tasks that are to be executed. Other task scheduling libraries
@@ -193,6 +191,7 @@ public Subscription subscribe(Subscriber<Task2> subscriber) {
         return taskSubject.subscribe(new SubscriberWrapper(subscriber));
     }
 
+    @Override
     public Observable<Long> getFinishedTimeSlices() {
         return tickSubject;
     }
@@ -281,6 +280,7 @@ public Observable<Lease> start() {
         // primary motivation for having this method return a hot observable.
         PublishSubject<Lease> leasesSubject = PublishSubject.create();
         processedLeases.subscribe(leasesSubject);
+        running = true;
         return leasesSubject;
     }
 
@@ -305,7 +305,7 @@ private Observable<Date> createTicks() {
         // figure something out.
         return Observable.timer(0, 1, TimeUnit.MINUTES, tickScheduler)
                 .map(tick -> currentTimeSlice())
-                .takeUntil(d -> shutdown)
+                .takeUntil(d -> !running)
                 .doOnNext(tick -> logger.debug("Tick {}", tick))
                 .observeOn(leaseScheduler);
     }
@@ -427,7 +427,7 @@ private Observable<Task2Impl> execute(Task2Impl task) {
     public void shutdown() {
         try {
             logger.debug("shutting down");
-            shutdown = true;
+            running = false;
 
             if (leasesSubscription != null) {
                 leasesSubscription.unsubscribe();