StatusResource implements Closeable { private Response<RunStateDataPayload> activeStates(RequestContext requestContext) { final Optional<String> componentOpt = requestContext.request().parameter("component"); final Optional<String> workflowOpt = requestContext.request().parameter("workflow"); final Optional<String> componentsOpt = requestContext.request().parameter("components"); final Map<WorkflowInstance, RunState> activeStates; final List<RunStateData> runStates = Lists.newArrayList(); try { activeStates = componentsOpt.isPresent() ? getActiveStates(componentsOpt.get()) : getActiveStates(componentOpt, workflowOpt); } catch (InvalidParametersException e) { return Response.forStatus(Status.BAD_REQUEST.withReasonPhrase(e.getMessage())); } catch (IOException e) { var errorMsg = "Could not read Active states: " + e; log.error(errorMsg); return Response.forStatus(Status.INTERNAL_SERVER_ERROR.withReasonPhrase(errorMsg)); } runStates.addAll( activeStates.values().stream().map(this::runStateToRunStateData).collect(toList())); return Response.forPayload(RunStateDataPayload.create(runStates)); } StatusResource(Storage storage, ServiceAccountUsageAuthorizer accountUsageAuthorizer); Stream<Route<AsyncHandler<Response<ByteString>>>> routes(); @Override void close();  }
@Test public void testGetAllActiveStates() throws Exception { sinceVersion(Api.Version.V3); storage.writeActiveState(WFI_1, PERSISTENT_STATE_1); storage.writeActiveState(WFI_2, PERSISTENT_STATE_2); assertThat(storage.readActiveStates().entrySet(), hasSize(2)); Response<ByteString> response = awaitResponse(serviceHelper.request("GET", path("/activeStates"))); assertThat(response, hasStatus(withCode(Status.OK))); String json = response.payload().orElseThrow().utf8(); RunStateDataPayload parsed = Json.OBJECT_MAPPER.readValue(json, RunStateDataPayload.class); assertThat(parsed.activeStates(), containsInAnyOrder( RunStateData.newBuilder() .workflowInstance(PERSISTENT_STATE_1.workflowInstance()) .state(PERSISTENT_STATE_1.state().name()) .stateData(PERSISTENT_STATE_1.data()) .latestTimestamp(PERSISTENT_STATE_1.timestamp()) .build(), RunStateData.newBuilder() .workflowInstance(PERSISTENT_STATE_2.workflowInstance()) .state(PERSISTENT_STATE_2.state().name()) .stateData(PERSISTENT_STATE_2.data()) .latestTimestamp(PERSISTENT_STATE_2.timestamp()) .build() )); }
@Test public void testFilterActiveStatesOnComponent() throws Exception { sinceVersion(Api.Version.V3); WorkflowInstance OTHER_WFI = WorkflowInstance.create(WorkflowId.create(C_ID_1 + "-other", WF_ID_1), PARAMETER); storage.writeActiveState(WFI_1, PERSISTENT_STATE_1); storage.writeActiveState(OTHER_WFI, PERSISTENT_STATE_2); storage.writeActiveState(WFI_3, PERSISTENT_STATE_3); assertThat(storage.readActiveStates().entrySet(), hasSize(3)); Response<ByteString> response = awaitResponse(serviceHelper.request("GET", path("/activeStates?component=" + C_ID_1))); assertThat(response, hasStatus(withCode(Status.OK))); String json = response.payload().orElseThrow().utf8(); RunStateDataPayload parsed = Json.OBJECT_MAPPER.readValue(json, RunStateDataPayload.class); assertThat(parsed.activeStates(), hasSize(2)); assertThat(parsed.activeStates().get(0).workflowInstance().workflowId().componentId(), is(C_ID_1)); assertThat(parsed.activeStates().get(0).workflowInstance().workflowId().id(), is(WF_ID_2)); assertThat(parsed.activeStates().get(1).workflowInstance().workflowId().componentId(), is(C_ID_1)); assertThat(parsed.activeStates().get(1).workflowInstance().workflowId().id(), is(WF_ID_1)); }
@Test public void testGetActiveStatesForMultiComponent() throws Exception { sinceVersion(Api.Version.V3); storage.writeActiveState(WFI_1, PERSISTENT_STATE_1); storage.writeActiveState(WFI_2, PERSISTENT_STATE_2); storage.writeActiveState(WFI_3, PERSISTENT_STATE_3); storage.writeActiveState(WFI_4, PERSISTENT_STATE_4); assertThat(storage.readActiveStates().entrySet(), hasSize(4)); Response<ByteString> response = awaitResponse(serviceHelper.request("GET", path("/activeStates?components=" + C_ID_1 + "," + C_ID_3 + "," + C_ID_1))); assertThat(response, hasStatus(withCode(Status.OK))); String json = response.payload().orElseThrow().utf8(); RunStateDataPayload parsed = Json.OBJECT_MAPPER.readValue(json, RunStateDataPayload.class); assertThat(parsed.activeStates(), hasSize(3)); var activeStates = ApiTestUtil.groupStatesByWorkflow(parsed.activeStates()); assertThat(activeStates.get(WorkflowId.create(C_ID_1, WF_ID_1)).size(), is(1)); assertThat(activeStates.get(WorkflowId.create(C_ID_1, WF_ID_2)).size(), is(1)); assertThat(activeStates.get(WorkflowId.create(C_ID_3, WF_ID_1)).size(), is(1)); }
@Test public void testFilterActiveStatesOnWorkflow() throws Exception { sinceVersion(Api.Version.V3); storage.writeActiveState(WFI_1, PERSISTENT_STATE_1); storage.writeActiveState(WFI_2, PERSISTENT_STATE_2); storage.writeActiveState(WFI_3, PERSISTENT_STATE_3); assertThat(storage.readActiveStates().entrySet(), hasSize(3)); Response<ByteString> response = awaitResponse(serviceHelper.request("GET", path("/activeStates?component=" + C_ID_1 + "&workflow=" + WF_ID_1))); assertThat(response, hasStatus(withCode(Status.OK))); String json = response.payload().orElseThrow().utf8(); RunStateDataPayload parsed = Json.OBJECT_MAPPER.readValue(json, RunStateDataPayload.class); assertThat(parsed.activeStates(), hasSize(1)); assertThat(parsed.activeStates().get(0).workflowInstance().workflowId().componentId(), is(C_ID_1)); assertThat(parsed.activeStates().get(0).workflowInstance().workflowId().id(), is(WF_ID_1)); }