This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch dev in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit faa0d6d4857829470b0bc29d745ea4b11c8d5090 Author: Dominik Riemer <[email protected]> AuthorDate: Tue Dec 28 16:21:07 2021 +0100 [hotfix] Use standard method to check for debug flag --- .../api/InvocablePipelineElementResource.java | 235 +++++++++++---------- 1 file changed, 119 insertions(+), 116 deletions(-) diff --git a/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java b/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java index 2e4cebd..0cab11b 100644 --- a/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java +++ b/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java @@ -18,6 +18,7 @@ package org.apache.streampipes.container.api; +import org.apache.streampipes.commons.constants.Envs; import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.container.declarer.Declarer; import org.apache.streampipes.container.declarer.InvocableDeclarer; @@ -42,138 +43,140 @@ import java.util.Map; public abstract class InvocablePipelineElementResource<I extends InvocableStreamPipesEntity, D extends Declarer<?>, P extends AbstractParameterExtractor<I>> extends AbstractPipelineElementResource<D> { - private static final Logger LOG = LoggerFactory.getLogger(InvocablePipelineElementResource.class); - - protected abstract Map<String, D> getElementDeclarers(); - protected abstract String getInstanceId(String uri, String elementId); - - protected Class<I> clazz; - - public InvocablePipelineElementResource(Class<I> clazz) { - this.clazz = clazz; - } - - @POST - @Path("{elementId}") - @Produces(MediaType.APPLICATION_JSON) - @Consumes(MediaType.APPLICATION_JSON) - @JacksonSerialized - public javax.ws.rs.core.Response invokeRuntime(@PathParam("elementId") String elementId, I graph) { - - try { - if (isDebug()) { - graph = createGroundingDebugInformation(graph); - } - - InvocableDeclarer declarer = (InvocableDeclarer) getDeclarerById(elementId); - - if (declarer != null) { - String runningInstanceId = getInstanceId(graph.getElementId(), elementId); - if (!RunningInstances.INSTANCE.exists(runningInstanceId)) { - RunningInstances.INSTANCE.add(runningInstanceId, graph, declarer.getClass().newInstance()); - Response resp = RunningInstances.INSTANCE.getInvocation(runningInstanceId).invokeRuntime(graph, getServiceGroup()); - if (!resp.isSuccess()) { - LOG.error("Could not invoke pipeline element {} due to the following error: {}", - graph.getName(), - resp.getOptionalMessage()); - RunningInstances.INSTANCE.remove(runningInstanceId); - } - return ok(resp); - } else { - LOG.info("Pipeline element {} with id {} seems to be already running, skipping invocation request.", graph.getName(), runningInstanceId); - Response resp = new Response(graph.getElementId(), true); - return ok(resp); - } - - } - } catch (InstantiationException | IllegalAccessException e) { - e.printStackTrace(); - return ok(new Response(elementId, false, e.getMessage())); + private static final Logger LOG = LoggerFactory.getLogger(InvocablePipelineElementResource.class); + + protected abstract Map<String, D> getElementDeclarers(); + + protected abstract String getInstanceId(String uri, String elementId); + + protected Class<I> clazz; + + public InvocablePipelineElementResource(Class<I> clazz) { + this.clazz = clazz; + } + + @POST + @Path("{elementId}") + @Produces(MediaType.APPLICATION_JSON) + @Consumes(MediaType.APPLICATION_JSON) + @JacksonSerialized + public javax.ws.rs.core.Response invokeRuntime(@PathParam("elementId") String elementId, I graph) { + + try { + if (isDebug()) { + LOG.info("SP_DEBUG env variable is set - overriding broker hostname and port for local development"); + graph = createGroundingDebugInformation(graph); + } + + InvocableDeclarer declarer = (InvocableDeclarer) getDeclarerById(elementId); + + if (declarer != null) { + String runningInstanceId = getInstanceId(graph.getElementId(), elementId); + if (!RunningInstances.INSTANCE.exists(runningInstanceId)) { + RunningInstances.INSTANCE.add(runningInstanceId, graph, declarer.getClass().newInstance()); + Response resp = RunningInstances.INSTANCE.getInvocation(runningInstanceId).invokeRuntime(graph, getServiceGroup()); + if (!resp.isSuccess()) { + LOG.error("Could not invoke pipeline element {} due to the following error: {}", + graph.getName(), + resp.getOptionalMessage()); + RunningInstances.INSTANCE.remove(runningInstanceId); + } + return ok(resp); + } else { + LOG.info("Pipeline element {} with id {} seems to be already running, skipping invocation request.", graph.getName(), runningInstanceId); + Response resp = new Response(graph.getElementId(), true); + return ok(resp); } - return ok(new Response(elementId, false, "Could not find the element with id: " + elementId)); - } - - @POST - @Path("{elementId}/configurations") - @Produces(MediaType.APPLICATION_JSON) - @Consumes(MediaType.APPLICATION_JSON) - @JacksonSerialized - public javax.ws.rs.core.Response fetchConfigurations(@PathParam("elementId") String elementId, - RuntimeOptionsRequest runtimeOptionsRequest) { - - ResolvesContainerProvidedOptions resolvesOptions = (ResolvesContainerProvidedOptions) getDeclarerById(elementId); - - List<Option> availableOptions = - resolvesOptions.resolveOptions(runtimeOptionsRequest.getRequestId(), - StaticPropertyExtractor.from( - runtimeOptionsRequest.getStaticProperties(), - runtimeOptionsRequest.getInputStreams(), - runtimeOptionsRequest.getAppId() - )); - - return ok(new RuntimeOptionsResponse(runtimeOptionsRequest, availableOptions)); + } + } catch (InstantiationException | IllegalAccessException e) { + e.printStackTrace(); + return ok(new Response(elementId, false, e.getMessage())); } - @POST - @Path("{elementId}/output") - @Produces(MediaType.APPLICATION_JSON) - @Consumes(MediaType.APPLICATION_JSON) - @JacksonSerialized - public javax.ws.rs.core.Response fetchOutputStrategy(@PathParam("elementId") String elementId, I runtimeOptionsRequest) { - try { - //I runtimeOptionsRequest = JacksonSerializer.getObjectMapper().readValue(payload, clazz); - ResolvesContainerProvidedOutputStrategy<I, P> resolvesOutput = - (ResolvesContainerProvidedOutputStrategy<I, P>) - getDeclarerById - (elementId); - return ok(resolvesOutput.resolveOutputStrategy - (runtimeOptionsRequest, getExtractor(runtimeOptionsRequest))); - } catch (SpRuntimeException e) { - e.printStackTrace(); - return ok(new Response(elementId, false)); - } + return ok(new Response(elementId, false, "Could not find the element with id: " + elementId)); + } + + @POST + @Path("{elementId}/configurations") + @Produces(MediaType.APPLICATION_JSON) + @Consumes(MediaType.APPLICATION_JSON) + @JacksonSerialized + public javax.ws.rs.core.Response fetchConfigurations(@PathParam("elementId") String elementId, + RuntimeOptionsRequest runtimeOptionsRequest) { + + ResolvesContainerProvidedOptions resolvesOptions = (ResolvesContainerProvidedOptions) getDeclarerById(elementId); + + List<Option> availableOptions = + resolvesOptions.resolveOptions(runtimeOptionsRequest.getRequestId(), + StaticPropertyExtractor.from( + runtimeOptionsRequest.getStaticProperties(), + runtimeOptionsRequest.getInputStreams(), + runtimeOptionsRequest.getAppId() + )); + + return ok(new RuntimeOptionsResponse(runtimeOptionsRequest, availableOptions)); + } + + @POST + @Path("{elementId}/output") + @Produces(MediaType.APPLICATION_JSON) + @Consumes(MediaType.APPLICATION_JSON) + @JacksonSerialized + public javax.ws.rs.core.Response fetchOutputStrategy(@PathParam("elementId") String elementId, I runtimeOptionsRequest) { + try { + //I runtimeOptionsRequest = JacksonSerializer.getObjectMapper().readValue(payload, clazz); + ResolvesContainerProvidedOutputStrategy<I, P> resolvesOutput = + (ResolvesContainerProvidedOutputStrategy<I, P>) + getDeclarerById + (elementId); + return ok(resolvesOutput.resolveOutputStrategy + (runtimeOptionsRequest, getExtractor(runtimeOptionsRequest))); + } catch (SpRuntimeException e) { + e.printStackTrace(); + return ok(new Response(elementId, false)); } + } - // TODO move endpoint to /elementId/instances/runningInstanceId - @DELETE - @Path("{elementId}/{runningInstanceId}") - @Produces(MediaType.APPLICATION_JSON) - public javax.ws.rs.core.Response detach(@PathParam("elementId") String elementId, @PathParam("runningInstanceId") String runningInstanceId) { - - InvocableDeclarer runningInstance = RunningInstances.INSTANCE.getInvocation(runningInstanceId); + // TODO move endpoint to /elementId/instances/runningInstanceId + @DELETE + @Path("{elementId}/{runningInstanceId}") + @Produces(MediaType.APPLICATION_JSON) + public javax.ws.rs.core.Response detach(@PathParam("elementId") String elementId, @PathParam("runningInstanceId") String runningInstanceId) { - if (runningInstance != null) { - Response resp = runningInstance.detachRuntime(runningInstanceId, getServiceGroup()); + InvocableDeclarer runningInstance = RunningInstances.INSTANCE.getInvocation(runningInstanceId); - if (resp.isSuccess()) { - RunningInstances.INSTANCE.remove(runningInstanceId); - } + if (runningInstance != null) { + Response resp = runningInstance.detachRuntime(runningInstanceId, getServiceGroup()); - return ok(resp); - } + if (resp.isSuccess()) { + RunningInstances.INSTANCE.remove(runningInstanceId); + } - return ok(new Response(elementId, false, "Could not find the running instance with id: " + runningInstanceId)); + return ok(resp); } - @GET - @Path("{elementId}/instances") - @Produces(MediaType.APPLICATION_JSON) - public javax.ws.rs.core.Response listRunningInstances(@PathParam("elementId") String elementId) { - return ok(RunningInstances.INSTANCE.getRunningInstanceIdsForElement(elementId)); - } + return ok(new Response(elementId, false, "Could not find the running instance with id: " + runningInstanceId)); + } - protected abstract P getExtractor(I graph); + @GET + @Path("{elementId}/instances") + @Produces(MediaType.APPLICATION_JSON) + public javax.ws.rs.core.Response listRunningInstances(@PathParam("elementId") String elementId) { + return ok(RunningInstances.INSTANCE.getRunningInstanceIdsForElement(elementId)); + } - protected abstract I createGroundingDebugInformation(I graph); + protected abstract P getExtractor(I graph); - private Boolean isDebug() { - return "true".equals(System.getenv("SP_DEBUG")); - } + protected abstract I createGroundingDebugInformation(I graph); - private String getServiceGroup() { - return DeclarersSingleton.getInstance().getServiceDefinition().getServiceGroup(); - } + private Boolean isDebug() { + return Envs.SP_DEBUG.exists() && Envs.SP_DEBUG.getValueAsBoolean(); + } + + private String getServiceGroup() { + return DeclarersSingleton.getInstance().getServiceDefinition().getServiceGroup(); + } }
