This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch STREAMPIPES-319
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
The following commit(s) were added to refs/heads/STREAMPIPES-319 by this push:
new 6a569a9 [STREAMPIPES-319] Add config extraction feature to
onInvocation method
6a569a9 is described below
commit 6a569a910d3b354f1c101fa8de8c02acd0f51b6e
Author: Dominik Riemer <[email protected]>
AuthorDate: Mon May 10 13:49:53 2021 +0200
[STREAMPIPES-319] Add config extraction feature to onInvocation method
---
.../api/InvocablePipelineElementResource.java | 9 ++++++--
.../container/config/ConfigExtractor.java | 25 ++++++++++++----------
.../container/declarer/InvocableDeclarer.java | 7 +++---
.../container/init/DeclarersSingleton.java | 7 ++++++
.../apache/streampipes/rest/impl/RdfEndpoint.java | 2 +-
.../distributed/runtime/DistributedRuntime.java | 1 -
.../wrapper/flink/FlinkDataProcessorRuntime.java | 3 ++-
.../wrapper/flink/FlinkDataSinkRuntime.java | 3 ++-
.../wrapper/spark/AbstractSparkDeclarer.java | 4 ++--
.../StandaloneEventProcessingDeclarer.java | 13 +++++++----
.../StandaloneEventProcessorDeclarerSingleton.java | 12 ++++++++---
.../declarer/StandaloneEventSinkDeclarer.java | 10 ++++++---
.../StandaloneEventSinkDeclarerSingleton.java | 10 ++++++---
.../StandaloneExternalEventProcessingDeclarer.java | 8 +++++--
.../StandaloneExternalEventSinkDeclarer.java | 8 +++++--
streampipes-wrapper/pom.xml | 7 +++++-
.../wrapper/context/RuntimeContext.java | 8 +++++++
.../context/SpEventProcessorRuntimeContext.java | 9 ++++++--
.../wrapper/context/SpEventSinkRuntimeContext.java | 8 +++++--
.../wrapper/context/SpRuntimeContext.java | 20 ++++++++++++++++-
.../wrapper/declarer/EventProcessorDeclarer.java | 4 ++--
.../wrapper/declarer/EventSinkDeclarer.java | 4 ++--
.../wrapper/declarer/PipelineElementDeclarer.java | 24 ++++++++++++++-------
.../runtime/EventProcessorRuntimeParams.java | 11 +++++++---
.../params/runtime/EventSinkRuntimeParams.java | 11 +++++++---
.../wrapper/params/runtime/RuntimeParams.java | 11 +++++++++-
26 files changed, 175 insertions(+), 64 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 4530f8a..1adce39 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
@@ -21,6 +21,7 @@ package org.apache.streampipes.container.api;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.container.declarer.Declarer;
import org.apache.streampipes.container.declarer.InvocableDeclarer;
+import org.apache.streampipes.container.init.DeclarersSingleton;
import org.apache.streampipes.container.init.RunningInstances;
import org.apache.streampipes.model.Response;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
@@ -70,7 +71,7 @@ public abstract class InvocablePipelineElementResource<I
extends InvocableStream
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);
+ Response resp =
RunningInstances.INSTANCE.getInvocation(runningInstanceId).invokeRuntime(graph,
getServiceId());
return ok(resp);
} else {
LOG.info("Pipeline element {} with id {} seems to be
already running, skipping invocation request.", graph.getName(),
runningInstanceId);
@@ -138,7 +139,7 @@ public abstract class InvocablePipelineElementResource<I
extends InvocableStream
InvocableDeclarer runningInstance =
RunningInstances.INSTANCE.getInvocation(runningInstanceId);
if (runningInstance != null) {
- Response resp = runningInstance.detachRuntime(runningInstanceId);
+ Response resp = runningInstance.detachRuntime(runningInstanceId,
getServiceId());
if (resp.isSuccess()) {
RunningInstances.INSTANCE.remove(runningInstanceId);
@@ -157,5 +158,9 @@ public abstract class InvocablePipelineElementResource<I
extends InvocableStream
private Boolean isDebug() {
return "true".equals(System.getenv("SP_DEBUG"));
}
+
+ private String getServiceId() {
+ return DeclarersSingleton.getInstance().getServiceId();
+ }
}
diff --git
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/RuntimeContext.java
b/streampipes-container/src/main/java/org/apache/streampipes/container/config/ConfigExtractor.java
similarity index 62%
copy from
streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/RuntimeContext.java
copy to
streampipes-container/src/main/java/org/apache/streampipes/container/config/ConfigExtractor.java
index 83c2580..a43d9d6 100644
---
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/RuntimeContext.java
+++
b/streampipes-container/src/main/java/org/apache/streampipes/container/config/ConfigExtractor.java
@@ -15,21 +15,24 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.wrapper.context;
+package org.apache.streampipes.container.config;
-import org.apache.streampipes.logging.api.Logger;
-import org.apache.streampipes.model.runtime.SchemaInfo;
-import org.apache.streampipes.model.runtime.SourceInfo;
+import org.apache.streampipes.config.SpConfig;
+import org.apache.streampipes.config.consul.ConsulSpConfig;
-import java.util.List;
+public class ConfigExtractor {
-public interface RuntimeContext {
+ private SpConfig config;
- Logger getLogger();
+ public static ConfigExtractor from(String serviceId) {
+ return new ConfigExtractor(serviceId);
+ }
- List<SchemaInfo> getInputSchemaInfo();
+ private ConfigExtractor(String serviceId) {
+ this.config = new ConsulSpConfig(serviceId);
+ }
- List<SourceInfo> getInputSourceInfo();
-
- String getCorrespondingUser();
+ public SpConfig getConfig() {
+ return this.config;
+ }
}
diff --git
a/streampipes-container/src/main/java/org/apache/streampipes/container/declarer/InvocableDeclarer.java
b/streampipes-container/src/main/java/org/apache/streampipes/container/declarer/InvocableDeclarer.java
index 9792132..e3e05f7 100644
---
a/streampipes-container/src/main/java/org/apache/streampipes/container/declarer/InvocableDeclarer.java
+++
b/streampipes-container/src/main/java/org/apache/streampipes/container/declarer/InvocableDeclarer.java
@@ -22,9 +22,10 @@ import
org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
import org.apache.streampipes.model.Response;
-public interface InvocableDeclarer<D extends NamedStreamPipesEntity, I extends
InvocableStreamPipesEntity> extends Declarer<D> {
+public interface InvocableDeclarer<D extends NamedStreamPipesEntity,
+ I extends InvocableStreamPipesEntity> extends Declarer<D> {
- Response invokeRuntime(I invocationGraph);
+ Response invokeRuntime(I invocationGraph, String serviceId);
- Response detachRuntime(String pipelineId);
+ Response detachRuntime(String pipelineId, String serviceId);
}
diff --git
a/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java
b/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java
index 4de3c8d..1b908e8 100644
---
a/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java
+++
b/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java
@@ -49,6 +49,8 @@ public class DeclarersSingleton {
private Map<String, TransportProtocol> supportedProtocols;
private Map<String, TransportFormat> supportedFormats;
+ private String serviceId;
+
private int port;
private String route;
private String hostName;
@@ -77,6 +79,7 @@ public class DeclarersSingleton {
this.addDeclarers(serviceDef.getDeclarers());
this.registerProtocols(serviceDef.getProtocolDefinitionFactories());
this.registerDataFormats(serviceDef.getDataFormatFactories());
+ this.serviceId = serviceDef.getServiceId();
}
public void addDeclarers(List<Declarer<?>> allDeclarers) {
@@ -222,6 +225,10 @@ public class DeclarersSingleton {
return Http + hostName + Colon + port + route;
}
+ public String getServiceId() {
+ return serviceId;
+ }
+
private void checkAndStartExecutableStreams(DataStreamDeclarer declarer) {
if (declarer.isExecutable()) {
declarer.executeStream();
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/RdfEndpoint.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/RdfEndpoint.java
index 7de93c5..5842003 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/RdfEndpoint.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/RdfEndpoint.java
@@ -125,7 +125,7 @@ public class RdfEndpoint extends AbstractRestResource {
private List<RdfEndpointItem> getAllDataSinkEndpoints(String username,
List<RdfEndpointItem> existingItems) {
return getAllDataSinkUris(username)
.stream()
- .filter(s -> existingItems.stream().noneMatch(item ->
item.getElementId().equals(s)))
+ .filter(s -> existingItems.stream().noneMatch(item ->
s.equals(item.getElementId())))
.map(s ->
getTripleStorage().getPipelineElementStorage().getDataSinkById(s))
.map(source -> makeItem(source, "action"))
.collect(Collectors.toList());
diff --git
a/streampipes-wrapper-distributed/src/main/java/org/apache/streampipes/wrapper/distributed/runtime/DistributedRuntime.java
b/streampipes-wrapper-distributed/src/main/java/org/apache/streampipes/wrapper/distributed/runtime/DistributedRuntime.java
index dcd1694..2c508c7 100644
---
a/streampipes-wrapper-distributed/src/main/java/org/apache/streampipes/wrapper/distributed/runtime/DistributedRuntime.java
+++
b/streampipes-wrapper-distributed/src/main/java/org/apache/streampipes/wrapper/distributed/runtime/DistributedRuntime.java
@@ -17,7 +17,6 @@
*/
package org.apache.streampipes.wrapper.distributed.runtime;
-import org.apache.streampipes.config.backend.SpDataFormat;
import org.apache.streampipes.dataformat.SpDataFormatDefinition;
import org.apache.streampipes.dataformat.SpDataFormatManager;
import org.apache.streampipes.messaging.kafka.config.ConsumerConfigFactory;
diff --git
a/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataProcessorRuntime.java
b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataProcessorRuntime.java
index 519725e..b9a49d8 100644
---
a/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataProcessorRuntime.java
+++
b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataProcessorRuntime.java
@@ -107,6 +107,7 @@ public abstract class FlinkDataProcessorRuntime<B extends
EventProcessorBindingP
}
protected EventProcessorRuntimeParams<B> makeRuntimeParams() {
- return new EventProcessorRuntimeParams<>(bindingParams, false);
+ // TODO add support for config extractor & client
+ return new EventProcessorRuntimeParams<>(bindingParams, false, null, null);
}
}
diff --git
a/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataSinkRuntime.java
b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataSinkRuntime.java
index 611f671..761c6de 100644
---
a/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataSinkRuntime.java
+++
b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataSinkRuntime.java
@@ -58,7 +58,8 @@ public abstract class FlinkDataSinkRuntime<B extends
EventSinkBindingParams> ext
public abstract void getSink(DataStream<Event>... convertedStream1);
protected EventSinkRuntimeParams<B> makeRuntimeParams() {
- return new EventSinkRuntimeParams<>(bindingParams, false);
+ // TODO add support for config extractor & client
+ return new EventSinkRuntimeParams<>(bindingParams, false, null, null);
}
}
diff --git
a/streampipes-wrapper-spark/src/main/java/org/apache/streampipes/wrapper/spark/AbstractSparkDeclarer.java
b/streampipes-wrapper-spark/src/main/java/org/apache/streampipes/wrapper/spark/AbstractSparkDeclarer.java
index 4798ca5..5c41055 100644
---
a/streampipes-wrapper-spark/src/main/java/org/apache/streampipes/wrapper/spark/AbstractSparkDeclarer.java
+++
b/streampipes-wrapper-spark/src/main/java/org/apache/streampipes/wrapper/spark/AbstractSparkDeclarer.java
@@ -28,7 +28,7 @@ public abstract class AbstractSparkDeclarer<D extends
NamedStreamPipesEntity, I
protected I graph;
@Override
- public Response invokeRuntime(I sepaInvocation) {
+ public Response invokeRuntime(I sepaInvocation, String serviceId) {
runtime = getRuntime(sepaInvocation);
graph = sepaInvocation;
@@ -40,7 +40,7 @@ public abstract class AbstractSparkDeclarer<D extends
NamedStreamPipesEntity, I
}
@Override
- public Response detachRuntime(String s) {
+ public Response detachRuntime(String pipelineId, String serviceId) {
if (runtime.stop()) {
return new Response(graph.getElementId(), true);
} else {
diff --git
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventProcessingDeclarer.java
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventProcessingDeclarer.java
index 953c758..9cba639 100644
---
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventProcessingDeclarer.java
+++
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventProcessingDeclarer.java
@@ -18,6 +18,8 @@
package org.apache.streampipes.wrapper.standalone.declarer;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import
org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
import org.apache.streampipes.wrapper.declarer.EventProcessorDeclarer;
@@ -27,16 +29,19 @@ import
org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
import
org.apache.streampipes.wrapper.standalone.runtime.StandaloneEventProcessorRuntime;
public abstract class StandaloneEventProcessingDeclarer<B extends
- EventProcessorBindingParams> extends EventProcessorDeclarer<B,
StandaloneEventProcessorRuntime> {
+ EventProcessorBindingParams> extends EventProcessorDeclarer<B,
StandaloneEventProcessorRuntime<B>> {
- public abstract ConfiguredEventProcessor<B>
onInvocation(DataProcessorInvocation graph, ProcessingElementParameterExtractor
extractor);
+ public abstract ConfiguredEventProcessor<B>
onInvocation(DataProcessorInvocation graph,
+
ProcessingElementParameterExtractor extractor);
@Override
public StandaloneEventProcessorRuntime<B> getRuntime(DataProcessorInvocation
graph,
- ProcessingElementParameterExtractor extractor) {
+
ProcessingElementParameterExtractor extractor,
+ ConfigExtractor
configExtractor,
+ StreamPipesClient
streamPipesClient) {
ConfiguredEventProcessor<B> configuredEngine = onInvocation(graph,
extractor);
EventProcessorRuntimeParams<B> runtimeParams = new
EventProcessorRuntimeParams<>
- (configuredEngine.getBindingParams(), false);
+ (configuredEngine.getBindingParams(), false, configExtractor,
streamPipesClient);
return new
StandaloneEventProcessorRuntime<>(configuredEngine.getEngineSupplier(),
runtimeParams);
diff --git
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventProcessorDeclarerSingleton.java
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventProcessorDeclarerSingleton.java
index e9435ab..2fcf23f 100644
---
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventProcessorDeclarerSingleton.java
+++
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventProcessorDeclarerSingleton.java
@@ -18,6 +18,8 @@
package org.apache.streampipes.wrapper.standalone.declarer;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import
org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
import org.apache.streampipes.wrapper.declarer.EventProcessorDeclarer;
@@ -26,14 +28,18 @@ import
org.apache.streampipes.wrapper.params.runtime.EventProcessorRuntimeParams
import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
import
org.apache.streampipes.wrapper.standalone.runtime.StandaloneEventProcessorRuntime;
-public abstract class StandaloneEventProcessorDeclarerSingleton<B extends
EventProcessorBindingParams> extends EventProcessorDeclarer<B,
StandaloneEventProcessorRuntime> {
+public abstract class StandaloneEventProcessorDeclarerSingleton<B extends
EventProcessorBindingParams>
+ extends EventProcessorDeclarer<B, StandaloneEventProcessorRuntime<B>> {
@Override
- public StandaloneEventProcessorRuntime getRuntime(DataProcessorInvocation
graph, ProcessingElementParameterExtractor extractor) {
+ public StandaloneEventProcessorRuntime<B> getRuntime(DataProcessorInvocation
graph,
+
ProcessingElementParameterExtractor extractor,
+ ConfigExtractor
configExtractor,
+ StreamPipesClient
streamPipesClient) {
ConfiguredEventProcessor<B> configuredEngine = onInvocation(graph,
extractor);
EventProcessorRuntimeParams<B> runtimeParams = new
EventProcessorRuntimeParams<>
- (configuredEngine.getBindingParams(), true);
+ (configuredEngine.getBindingParams(), true, configExtractor,
streamPipesClient);
return new
StandaloneEventProcessorRuntime<>(configuredEngine.getEngineSupplier(),
runtimeParams);
diff --git
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventSinkDeclarer.java
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventSinkDeclarer.java
index 7842316..048b857 100644
---
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventSinkDeclarer.java
+++
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventSinkDeclarer.java
@@ -18,6 +18,8 @@
package org.apache.streampipes.wrapper.standalone.declarer;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.graph.DataSinkInvocation;
import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
import org.apache.streampipes.wrapper.declarer.EventSinkDeclarer;
@@ -27,15 +29,17 @@ import
org.apache.streampipes.wrapper.standalone.ConfiguredEventSink;
import
org.apache.streampipes.wrapper.standalone.runtime.StandaloneEventSinkRuntime;
public abstract class StandaloneEventSinkDeclarer<B extends
- EventSinkBindingParams> extends EventSinkDeclarer<B,
StandaloneEventSinkRuntime> {
+ EventSinkBindingParams> extends EventSinkDeclarer<B,
StandaloneEventSinkRuntime<B>> {
@Override
public StandaloneEventSinkRuntime<B> getRuntime(DataSinkInvocation graph,
- DataSinkParameterExtractor extractor) {
+ DataSinkParameterExtractor
extractor,
+ ConfigExtractor
configExtractor,
+ StreamPipesClient
streamPipesClient) {
ConfiguredEventSink<B> configuredEngine = onInvocation(graph, extractor);
EventSinkRuntimeParams<B> runtimeParams = new EventSinkRuntimeParams<>
- (configuredEngine.getBindingParams(), false);
+ (configuredEngine.getBindingParams(), false, configExtractor,
streamPipesClient);
return new
StandaloneEventSinkRuntime<>(configuredEngine.getEngineSupplier(),
runtimeParams);
}
diff --git
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventSinkDeclarerSingleton.java
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventSinkDeclarerSingleton.java
index 82aab6f..aa2a27b 100644
---
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventSinkDeclarerSingleton.java
+++
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneEventSinkDeclarerSingleton.java
@@ -18,6 +18,8 @@
package org.apache.streampipes.wrapper.standalone.declarer;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.graph.DataSinkInvocation;
import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
import org.apache.streampipes.wrapper.declarer.EventSinkDeclarer;
@@ -27,15 +29,17 @@ import
org.apache.streampipes.wrapper.standalone.ConfiguredEventSink;
import
org.apache.streampipes.wrapper.standalone.runtime.StandaloneEventSinkRuntime;
public abstract class StandaloneEventSinkDeclarerSingleton<B extends
- EventSinkBindingParams> extends EventSinkDeclarer<B,
StandaloneEventSinkRuntime> {
+ EventSinkBindingParams> extends EventSinkDeclarer<B,
StandaloneEventSinkRuntime<B>> {
@Override
public StandaloneEventSinkRuntime<B> getRuntime(DataSinkInvocation graph,
- DataSinkParameterExtractor extractor) {
+ DataSinkParameterExtractor
extractor,
+ ConfigExtractor
configExtractor,
+ StreamPipesClient
streamPipesClient) {
ConfiguredEventSink<B> configuredEngine = onInvocation(graph, extractor);
EventSinkRuntimeParams<B> runtimeParams = new EventSinkRuntimeParams<>
- (configuredEngine.getBindingParams(), true);
+ (configuredEngine.getBindingParams(), true, configExtractor,
streamPipesClient);
return new
StandaloneEventSinkRuntime<>(configuredEngine.getEngineSupplier(),
runtimeParams);
}
diff --git
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneExternalEventProcessingDeclarer.java
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneExternalEventProcessingDeclarer.java
index feb702c..8e020cf 100644
---
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneExternalEventProcessingDeclarer.java
+++
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneExternalEventProcessingDeclarer.java
@@ -17,6 +17,8 @@
*/
package org.apache.streampipes.wrapper.standalone.declarer;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import
org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
import org.apache.streampipes.wrapper.declarer.EventProcessorDeclarer;
@@ -34,10 +36,12 @@ public abstract class
StandaloneExternalEventProcessingDeclarer<B extends
@Override
public StandaloneExternalEventProcessorRuntime<B>
getRuntime(DataProcessorInvocation graph,
-
ProcessingElementParameterExtractor extractor) {
+
ProcessingElementParameterExtractor extractor,
+ ConfigExtractor
configExtractor,
+
StreamPipesClient streamPipesClient) {
ConfiguredExternalEventProcessor<B> configuredEngine = onInvocation(graph,
extractor);
EventProcessorRuntimeParams<B> runtimeParams = new
EventProcessorRuntimeParams<>
- (configuredEngine.getBindingParams(), false);
+ (configuredEngine.getBindingParams(), false, configExtractor,
streamPipesClient);
return new
StandaloneExternalEventProcessorRuntime<>(configuredEngine.getEngineSupplier(),
runtimeParams);
diff --git
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneExternalEventSinkDeclarer.java
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneExternalEventSinkDeclarer.java
index 7ba8c97..2815c08 100644
---
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneExternalEventSinkDeclarer.java
+++
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/declarer/StandaloneExternalEventSinkDeclarer.java
@@ -17,6 +17,8 @@
*/
package org.apache.streampipes.wrapper.standalone.declarer;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.graph.DataSinkInvocation;
import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
import org.apache.streampipes.wrapper.declarer.EventSinkDeclarer;
@@ -31,11 +33,13 @@ public abstract class StandaloneExternalEventSinkDeclarer<B
extends
@Override
public StandaloneExternalEventSinkRuntime<B> getRuntime(DataSinkInvocation
graph,
- DataSinkParameterExtractor
extractor) {
+
DataSinkParameterExtractor extractor,
+ ConfigExtractor
configExtractor,
+ StreamPipesClient
streamPipesClient) {
ConfiguredExternalEventSink<B> configuredEngine = onInvocation(graph,
extractor);
EventSinkRuntimeParams<B> runtimeParams = new EventSinkRuntimeParams<>
- (configuredEngine.getBindingParams(), false);
+ (configuredEngine.getBindingParams(), false, configExtractor,
streamPipesClient);
return new
StandaloneExternalEventSinkRuntime<>(configuredEngine.getEngineSupplier(),
runtimeParams);
diff --git a/streampipes-wrapper/pom.xml b/streampipes-wrapper/pom.xml
index 5a92e38..1801bd8 100644
--- a/streampipes-wrapper/pom.xml
+++ b/streampipes-wrapper/pom.xml
@@ -34,6 +34,11 @@
</dependency>
<dependency>
<groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-client</artifactId>
+ <version>0.68.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
<artifactId>streampipes-dataformat</artifactId>
<version>0.68.0-SNAPSHOT</version>
</dependency>
@@ -48,4 +53,4 @@
<version>0.68.0-SNAPSHOT</version>
</dependency>
</dependencies>
-</project>
\ No newline at end of file
+</project>
diff --git
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/RuntimeContext.java
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/RuntimeContext.java
index 83c2580..5570767 100644
---
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/RuntimeContext.java
+++
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/RuntimeContext.java
@@ -17,6 +17,8 @@
*/
package org.apache.streampipes.wrapper.context;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.logging.api.Logger;
import org.apache.streampipes.model.runtime.SchemaInfo;
import org.apache.streampipes.model.runtime.SourceInfo;
@@ -32,4 +34,10 @@ public interface RuntimeContext {
List<SourceInfo> getInputSourceInfo();
String getCorrespondingUser();
+
+ ConfigExtractor getConfigStore();
+
+ StreamPipesClient getStreamPipesClient();
+
+
}
diff --git
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpEventProcessorRuntimeContext.java
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpEventProcessorRuntimeContext.java
index a8159fe..3d8579b 100644
---
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpEventProcessorRuntimeContext.java
+++
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpEventProcessorRuntimeContext.java
@@ -17,6 +17,8 @@
*/
package org.apache.streampipes.wrapper.context;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.runtime.SchemaInfo;
import org.apache.streampipes.model.runtime.SourceInfo;
@@ -33,8 +35,10 @@ public class SpEventProcessorRuntimeContext extends
SpRuntimeContext implements
List<SchemaInfo> inputSchemaInfo,
SourceInfo outputSourceInfo,
SchemaInfo outputSchemaInfo,
- String correspondingUser) {
- super(inputSourceInfo, inputSchemaInfo, correspondingUser);
+ String correspondingUser,
+ ConfigExtractor configExtractor,
+ StreamPipesClient streamPipesClient) {
+ super(inputSourceInfo, inputSchemaInfo, correspondingUser,
configExtractor, streamPipesClient);
this.outputSchemaInfo = outputSchemaInfo;
this.outputSourceInfo = outputSourceInfo;
}
@@ -52,4 +56,5 @@ public class SpEventProcessorRuntimeContext extends
SpRuntimeContext implements
public SourceInfo getOutputSourceInfo() {
return outputSourceInfo;
}
+
}
diff --git
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpEventSinkRuntimeContext.java
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpEventSinkRuntimeContext.java
index 19c79aa..033918a 100644
---
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpEventSinkRuntimeContext.java
+++
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpEventSinkRuntimeContext.java
@@ -17,6 +17,8 @@
*/
package org.apache.streampipes.wrapper.context;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.runtime.SchemaInfo;
import org.apache.streampipes.model.runtime.SourceInfo;
@@ -28,7 +30,9 @@ public class SpEventSinkRuntimeContext extends
SpRuntimeContext implements
public SpEventSinkRuntimeContext(List<SourceInfo> sourceInfo,
List<SchemaInfo> inputSchemaInfo,
- String correspondingUser) {
- super(sourceInfo, inputSchemaInfo, correspondingUser);
+ String correspondingUser,
+ ConfigExtractor configExtractor,
+ StreamPipesClient streamPipesClient) {
+ super(sourceInfo, inputSchemaInfo, correspondingUser, configExtractor,
streamPipesClient);
}
}
diff --git
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpRuntimeContext.java
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpRuntimeContext.java
index 40b9f34..44d6d61 100644
---
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpRuntimeContext.java
+++
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/context/SpRuntimeContext.java
@@ -17,6 +17,8 @@
*/
package org.apache.streampipes.wrapper.context;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.logging.api.Logger;
import org.apache.streampipes.model.runtime.SchemaInfo;
import org.apache.streampipes.model.runtime.SourceInfo;
@@ -28,13 +30,19 @@ public class SpRuntimeContext implements RuntimeContext {
private List<SchemaInfo> inputSchemaInfo;
private List<SourceInfo> sourceInfo;
private String correspondingUser;
+ private ConfigExtractor configExtractor;
+ private StreamPipesClient streamPipesClient;
public SpRuntimeContext(List<SourceInfo> sourceInfo,
List<SchemaInfo> inputSchemaInfo,
- String correspondingUser) {
+ String correspondingUser,
+ ConfigExtractor configExtractor,
+ StreamPipesClient streamPipesClient) {
this.inputSchemaInfo = inputSchemaInfo;
this.sourceInfo = sourceInfo;
this.correspondingUser = correspondingUser;
+ this.configExtractor = configExtractor;
+ this.streamPipesClient = streamPipesClient;
}
public SpRuntimeContext() {
@@ -61,4 +69,14 @@ public class SpRuntimeContext implements RuntimeContext {
public String getCorrespondingUser() {
return correspondingUser;
}
+
+ @Override
+ public ConfigExtractor getConfigStore() {
+ return this.configExtractor;
+ }
+
+ @Override
+ public StreamPipesClient getStreamPipesClient() {
+ return streamPipesClient;
+ }
}
diff --git
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/EventProcessorDeclarer.java
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/EventProcessorDeclarer.java
index fa8c5bf..a8427da 100644
---
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/EventProcessorDeclarer.java
+++
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/EventProcessorDeclarer.java
@@ -40,8 +40,8 @@ public abstract class EventProcessorDeclarer<B extends
EventProcessorBindingPara
}
@Override
- public Response invokeRuntime(DataProcessorInvocation graph) {
- return invokeEPRuntime(graph);
+ public Response invokeRuntime(DataProcessorInvocation graph, String
serviceId) {
+ return invokeEPRuntime(graph, serviceId);
}
}
diff --git
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/EventSinkDeclarer.java
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/EventSinkDeclarer.java
index f510f86..03974e0 100644
---
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/EventSinkDeclarer.java
+++
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/EventSinkDeclarer.java
@@ -36,8 +36,8 @@ public abstract class EventSinkDeclarer<B extends
EventSinkBindingParams, ES ext
}
@Override
- public Response invokeRuntime(DataSinkInvocation graph) {
- return invokeEPRuntime(graph);
+ public Response invokeRuntime(DataSinkInvocation graph, String serviceId) {
+ return invokeEPRuntime(graph, serviceId);
}
}
diff --git
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/PipelineElementDeclarer.java
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/PipelineElementDeclarer.java
index 97877d6..73bb9ff 100644
---
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/PipelineElementDeclarer.java
+++
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/declarer/PipelineElementDeclarer.java
@@ -18,6 +18,8 @@
package org.apache.streampipes.wrapper.declarer;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.Response;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.sdk.extractor.AbstractParameterExtractor;
@@ -31,22 +33,24 @@ public abstract class PipelineElementDeclarer<B extends
BindingParams, EPR exten
protected EPR epRuntime;
protected String elementId;
- public Response invokeEPRuntime(I graph) {
+ public Response invokeEPRuntime(I graph, String serviceId) {
try {
elementId = graph.getElementId();
- // change to getRuntime(graph, extractor)
- epRuntime = getRuntime(graph, getExtractor(graph));
+ ConfigExtractor configExtractor = makeConfigExtractor(serviceId);
+ // TODO add StreamPipes Client support
+ StreamPipesClient streamPipesClient = null;
+ epRuntime = getRuntime(graph, getExtractor(graph), configExtractor,
streamPipesClient);
epRuntime.bindRuntime();
return new Response(graph.getElementId(), true);
} catch (Exception e) {
e.printStackTrace();
return new Response(graph.getElementId(), false, e.getMessage());
}
-
}
- public Response detachRuntime(String pipelineId) {
+ public Response detachRuntime(String pipelineId,
+ String serviceId) {
try {
epRuntime.discardRuntime();
return new Response(elementId, true);
@@ -58,9 +62,13 @@ public abstract class PipelineElementDeclarer<B extends
BindingParams, EPR exten
protected abstract EX getExtractor(I graph);
- public abstract EPR getRuntime(I graph, EX extractor);
-
-
+ public abstract EPR getRuntime(I graph,
+ EX extractor,
+ ConfigExtractor configExtractor,
+ StreamPipesClient streamPipesClient);
+ private ConfigExtractor makeConfigExtractor(String serviceId) {
+ return ConfigExtractor.from(serviceId);
+ }
}
diff --git
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/EventProcessorRuntimeParams.java
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/EventProcessorRuntimeParams.java
index 3160e9c..b26c7ae 100644
---
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/EventProcessorRuntimeParams.java
+++
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/EventProcessorRuntimeParams.java
@@ -18,6 +18,8 @@
package org.apache.streampipes.wrapper.params.runtime;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
import org.apache.streampipes.wrapper.context.SpEventProcessorRuntimeContext;
@@ -26,8 +28,11 @@ import
org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams
public class EventProcessorRuntimeParams<B extends
EventProcessorBindingParams> extends
RuntimeParams<B, DataProcessorInvocation,
EventProcessorRuntimeContext> { // B - Bind Type
- public EventProcessorRuntimeParams(B bindingParams, Boolean singletonEngine)
{
- super(bindingParams, singletonEngine);
+ public EventProcessorRuntimeParams(B bindingParams,
+ Boolean singletonEngine,
+ ConfigExtractor configExtractor,
+ StreamPipesClient streamPipesClient) {
+ super(bindingParams, singletonEngine, configExtractor, streamPipesClient);
}
@Override
@@ -35,7 +40,7 @@ public class EventProcessorRuntimeParams<B extends
EventProcessorBindingParams>
return new SpEventProcessorRuntimeContext(getSourceInfo(),
getSchemaInfo(), bindingParams.getOutputStreamParams()
.getSourceInfo(),
bindingParams.getOutputStreamParams().getSchemaInfo(),
- bindingParams.getGraph().getCorrespondingUser());
+ bindingParams.getGraph().getCorrespondingUser(), configExtractor,
streamPipesClient);
}
}
diff --git
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/EventSinkRuntimeParams.java
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/EventSinkRuntimeParams.java
index 0286991..58ef58d 100644
---
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/EventSinkRuntimeParams.java
+++
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/EventSinkRuntimeParams.java
@@ -18,6 +18,8 @@
package org.apache.streampipes.wrapper.params.runtime;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.graph.DataSinkInvocation;
import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
import org.apache.streampipes.wrapper.context.SpEventSinkRuntimeContext;
@@ -26,14 +28,17 @@ import
org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
public class EventSinkRuntimeParams<B extends EventSinkBindingParams> extends
RuntimeParams<B, DataSinkInvocation, EventSinkRuntimeContext> {
- public EventSinkRuntimeParams(B bindingParams, Boolean singletonEngine) {
- super(bindingParams, singletonEngine);
+ public EventSinkRuntimeParams(B bindingParams,
+ Boolean singletonEngine,
+ ConfigExtractor configExtractor,
+ StreamPipesClient streamPipesClient) {
+ super(bindingParams, singletonEngine, configExtractor, streamPipesClient);
}
@Override
protected EventSinkRuntimeContext makeRuntimeContext() {
return new SpEventSinkRuntimeContext(getSourceInfo(), getSchemaInfo(),
- bindingParams.getGraph().getCorrespondingUser());
+ bindingParams.getGraph().getCorrespondingUser(), configExtractor,
streamPipesClient);
}
}
diff --git
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/RuntimeParams.java
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/RuntimeParams.java
index fb77a92..d866a15 100644
---
a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/RuntimeParams.java
+++
b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/params/runtime/RuntimeParams.java
@@ -18,6 +18,8 @@
package org.apache.streampipes.wrapper.params.runtime;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.container.config.ConfigExtractor;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.model.runtime.EventFactory;
@@ -38,14 +40,21 @@ public abstract class RuntimeParams<B extends
BindingParams<I>, I extends
protected final B bindingParams;
protected RC runtimeContext;
+ protected ConfigExtractor configExtractor;
+ protected StreamPipesClient streamPipesClient;
private Map<String, Integer> eventInfoMap = new HashMap<>();
private Boolean singletonEngine;
- public RuntimeParams(B bindingParams, Boolean singletonEngine) {
+ public RuntimeParams(B bindingParams,
+ Boolean singletonEngine,
+ ConfigExtractor configExtractor,
+ StreamPipesClient streamPipesClient) {
this.bindingParams = bindingParams;
this.singletonEngine = singletonEngine;
+ this.configExtractor = configExtractor;
+ this.streamPipesClient = streamPipesClient;
buildEventInfoMap();
this.runtimeContext = makeRuntimeContext();
}
