ActiveResource implements CommonRestResource, SingletonRestResource { @GET @Produces("application/json") @Path("/requests") public void activeRequests(@Suspended AsyncResponse asyncResponse, @QueryParam("type") final String type) { ReactiveSeq.of((type == null ? "default" : type)) .map(typeToUse -> activeQueries.getMap() .get(typeToUse) .toString()) .foldFuture(WorkerThreads.ioExecutor.get(), s->s.forEach(Long.MAX_VALUE,result -> asyncResponse.resume(result))); } @Autowired  ActiveResource(RequestTypes<T> activeQueries, JobsBeingExecuted activeJobs); @GET @Produces("application/json") @Path("/requests") void activeRequests(@Suspended AsyncResponse asyncResponse, @QueryParam("type") final String type); @GET @Produces("application/json") @Path("/all-requests") void allActiveRequests(@Suspended AsyncResponse asyncResponse); @GET @Produces("application/json") @Path("/jobs") void activeJobs(@Suspended AsyncResponse asyncResponse);  }
@Test public void testactiveRequests() { bus.post(RequestEvents.start("query", "1")); bus.post(RequestEvents.start("query", "2", "partition", ImmutableMap.of())); MockAsyncResponse<String> response = new MockAsyncResponse<>(); active.activeRequests(response, null); assertThat(convert(response.response()).get("events"), is(1)); active.activeRequests(response, "partition"); assertThat(convert(response.response()).get("events"), is(1)); }
@Test public void whenQueriesWithTheSameIdToDifferentTypesEventsIs1ForBoth() { bus.post(RequestEvents.start("query", "1")); bus.post(RequestEvents.start("query", "1", "partition", ImmutableMap.of())); MockAsyncResponse<String> response = new MockAsyncResponse<>(); active.activeRequests(response, null); assertThat(convert(response.response()).get("events"), is(1)); active.activeRequests(response, "partition"); assertThat(convert(response.response()).get("events"), is(1)); }
@Test public void whenQueriesWithTheSameIdButSameTypesEventsIs2ButSizeIs1() { bus.post(RequestEvents.start("query", "1")); bus.post(RequestEvents.start("query", "1")); MockAsyncResponse<String> response = new MockAsyncResponse<>(); active.activeRequests(response, null); assertThat(convert(response.response()).get("events"), is(2)); active.activeRequests(response, null); Map map = convert(response.response()); assertThat(((Map) map.get("active")).size(), is(1)); }