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 6cdeb31bdc5e3f5862f7c96b9bba60ae2a6cf6e6 Author: Dominik Riemer <[email protected]> AuthorDate: Sun Oct 18 16:03:08 2020 +0200 [STREAMPIPES-250] Refactor Siddhi wrapper --- .../SiddhiConstants.java} | 16 +- .../wrapper/siddhi/engine/SiddhiEngine.java | 122 +++++++++ .../wrapper/siddhi/engine/SiddhiEventEngine.java | 281 ++++----------------- .../siddhi/engine/SiddhiInvocationConfig.java | 74 ++++++ ...Callback.java => SiddhiStatementGenerator.java} | 10 +- .../siddhi/engine/StreamPipesSiddhiProcessor.java | 54 ++++ .../engine/{ => callback}/SiddhiDebugCallback.java | 2 +- .../callback/SiddhiOutputStreamCallback.java | 51 ++++ .../SiddhiOutputStreamDebugCallback.java} | 24 +- .../engine/generator/EventTypeGenerator.java | 73 ++++++ .../engine/generator/InputStreamNameGenerator.java | 44 ++++ .../engine/generator/SiddhiAppGenerator.java | 92 +++++++ .../EventType.java} | 26 +- .../wrapper/siddhi/utils/SiddhiUtils.java | 73 ++++++ 14 files changed, 689 insertions(+), 253 deletions(-) diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/constants/SiddhiConstants.java similarity index 55% copy from streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java copy to streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/constants/SiddhiConstants.java index 540d608..3d2a2a3 100644 --- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/constants/SiddhiConstants.java @@ -15,12 +15,20 @@ * limitations under the License. * */ -package org.apache.streampipes.wrapper.siddhi.engine; +package org.apache.streampipes.wrapper.siddhi.constants; +public class SiddhiConstants { -import io.siddhi.core.event.Event; + public static final String SELECT = "select"; + public static final String INSERT = "insert"; -public interface SiddhiDebugCallback { + public static final String FIRST_STREAM_PREFIX = "s0"; + public static final String SECOND_STREAM_PREFIX = "s1"; - void onEvent(Event event); + public static final String SIDDHI_LONG_TYPE = "LONG"; + public static final String SIDDHI_INT_TYPE = "INT"; + public static final String SIDDHI_DOUBLE_TYPE = "DOUBLE"; + public static final String SIDDHI_BOOLEAN_TYPE = "BOOL"; + public static final String SIDDHI_STRING_TYPE = "STRING"; + public static final String SIDDHI_OBJECT_TYPE = "OBJECT"; } diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiEngine.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiEngine.java new file mode 100644 index 0000000..173f30f --- /dev/null +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiEngine.java @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.streampipes.wrapper.siddhi.engine; + +import io.siddhi.core.SiddhiAppRuntime; +import io.siddhi.core.SiddhiManager; +import io.siddhi.core.stream.input.InputHandler; +import io.siddhi.core.stream.output.StreamCallback; +import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext; +import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams; +import org.apache.streampipes.wrapper.routing.SpOutputCollector; +import org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback; +import org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiOutputStreamCallback; +import org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiOutputStreamDebugCallback; +import org.apache.streampipes.wrapper.siddhi.model.EventType; +import org.apache.streampipes.wrapper.siddhi.manager.SpSiddhiManager; +import org.apache.streampipes.wrapper.siddhi.utils.SiddhiUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +public class SiddhiEngine { + + private static final Logger LOG = LoggerFactory.getLogger(SiddhiEngine.class); + + private SiddhiAppRuntime siddhiAppRuntime; + private final Map<String, InputHandler> siddhiInputHandlers; + private Map<String, List<EventType>> typeInfo; + + private Boolean debugMode; + private SiddhiDebugCallback debugCallback; + + public SiddhiEngine() { + this.siddhiInputHandlers = new HashMap<>(); + this.debugMode = false; + } + + public SiddhiEngine(SiddhiDebugCallback debugCallback) { + this(); + this.debugCallback = debugCallback; + this.debugMode = true; + } + + public void initializeEngine(SiddhiInvocationConfig<? extends EventProcessorBindingParams> settings, + SpOutputCollector spOutputCollector, + EventProcessorRuntimeContext runtimeContext) { + + EventProcessorBindingParams params = settings.getParams(); + this.typeInfo = settings.getEventTypeInfo(); + SiddhiManager siddhiManager = SpSiddhiManager.INSTANCE.getSiddhiManager(); + + //this.timestampField = removeStreamIdFromTimestamp(setTimestamp(parameters)); + + siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(settings.getSiddhiAppString()); + settings.getParams() + .getInEventTypes() + .forEach((key, value) -> { + String preparedKey = SiddhiUtils.prepareName(key); + siddhiInputHandlers.put(key, siddhiAppRuntime.getInputHandler(preparedKey)); + }); + + StreamCallback callback; + + if (!debugMode) { + callback = new SiddhiOutputStreamCallback(spOutputCollector, settings.getOutputEventKeys(), runtimeContext); + } else { + callback = new SiddhiOutputStreamDebugCallback(debugCallback); + } + + siddhiAppRuntime.addCallback(SiddhiUtils.prepareName(SiddhiUtils.getOutputTopicName(params)), callback); + siddhiAppRuntime.start(); + } + + public void processEvent(org.apache.streampipes.model.runtime.Event event) { + try { + String sourceId = event.getSourceInfo().getSourceId(); + InputHandler inputHandler = siddhiInputHandlers.get(sourceId); + List<String> eventKeys = this.typeInfo + .get(sourceId) + .stream() + .map(EventType::getEventTypeName) + .collect(Collectors.toList()); + + inputHandler.send(SiddhiUtils.toObjArr(eventKeys, event.getRaw())); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + + public void shutdownEngine() { + this.siddhiAppRuntime.shutdown(); + } + +// public void setSortedEventKeys(List<String> sortedEventKeys) { +// String streamId = (String) this.listOfEventKeys.keySet().toArray()[0]; // only reliable if there is only one stream, else use changeEventKeys() to respective streamId +// changeEventKeys(streamId, sortedEventKeys); +// } +// +// public void changeEventKeys(String streamId, List<String> newEventKeys) { +// this.listOfEventKeys.put(streamId, newEventKeys); +// } + +} diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiEventEngine.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiEventEngine.java index a642f6c..585ac5a 100644 --- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiEventEngine.java +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiEventEngine.java @@ -17,290 +17,97 @@ */ package org.apache.streampipes.wrapper.siddhi.engine; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.apache.streampipes.model.graph.DataProcessorInvocation; -import org.apache.streampipes.model.runtime.EventFactory; -import org.apache.streampipes.model.runtime.SchemaInfo; -import org.apache.streampipes.model.runtime.SourceInfo; import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext; import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams; import org.apache.streampipes.wrapper.routing.SpOutputCollector; import org.apache.streampipes.wrapper.runtime.EventProcessor; -import org.apache.streampipes.wrapper.siddhi.manager.SpSiddhiManager; -import io.siddhi.core.SiddhiAppRuntime; -import io.siddhi.core.SiddhiManager; -import io.siddhi.core.event.Event; -import io.siddhi.core.stream.input.InputHandler; -import io.siddhi.core.stream.output.StreamCallback; +import org.apache.streampipes.wrapper.siddhi.constants.SiddhiConstants; +import org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback; +import org.apache.streampipes.wrapper.siddhi.engine.generator.InputStreamNameGenerator; +import org.apache.streampipes.wrapper.siddhi.utils.SiddhiUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -import java.util.*; +import java.util.ArrayList; +import java.util.List; public abstract class SiddhiEventEngine<B extends EventProcessorBindingParams> implements - EventProcessor<B> { - - private StringBuilder siddhiAppString; - - private SiddhiAppRuntime siddhiAppRuntime; - private Map<String, InputHandler> siddhiInputHandlers; - private List<String> inputStreamNames; - - private Map<String, List> listOfEventKeys; - private List<String> outputEventKeys; - - private Boolean debugMode; - private SiddhiDebugCallback debugCallback; - - private String timestampField; + EventProcessor<B>, SiddhiStatementGenerator<B> { private static final Logger LOG = LoggerFactory.getLogger(SiddhiEventEngine.class); + private SiddhiEngine siddhiEngine; + private SiddhiInvocationConfig<B> siddhiConfig; + private List<String> outputEventKeys = new ArrayList<>(); + public SiddhiEventEngine() { - this.siddhiAppString = new StringBuilder(); - this.siddhiInputHandlers = new HashMap<>(); - this.inputStreamNames = new ArrayList<>(); - listOfEventKeys = new HashMap<>(); - outputEventKeys = new ArrayList<>(); - this.debugMode = false; + this.siddhiEngine = new SiddhiEngine(); } public SiddhiEventEngine(SiddhiDebugCallback debugCallback) { - this(); - this.debugCallback = debugCallback; - this.debugMode = true; + this.siddhiEngine = new SiddhiEngine(debugCallback); } @Override public void onInvocation(B parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) { - if (parameters.getInEventTypes().size() != parameters.getGraph().getInputStreams().size()) { - throw new IllegalArgumentException("Input parameters do not match!"); - } - - SiddhiManager siddhiManager = SpSiddhiManager.INSTANCE.getSiddhiManager(); - - this.timestampField = removeStreamIdFromTimestamp(setTimestamp(parameters)); - - LOG.info("Configuring event types for graph " + parameters.getGraph().getName()); - parameters.getInEventTypes().forEach((key, value) -> { - // TODO why is the prefix not in the parameters.getInEventType - registerEventTypeIfNotExists(key, value); - this.inputStreamNames.add(prepareName(key)); - }); - - LOG.info("Configuring output event keys for graph " + parameters.getGraph().getName()); - //System.out.println("output key: " + key); - outputEventKeys.addAll(parameters.getOutEventType().keySet()); - + List<String> inputStreamNames = new InputStreamNameGenerator<>(parameters).generateInputStreamNames(); + this.outputEventKeys = new ArrayList<>(parameters.getOutEventType().keySet()); String fromStatement = fromStatement(inputStreamNames, parameters); - String selectStatement = selectStatement(parameters); - registerStatements(fromStatement, selectStatement, getOutputTopicName(parameters)); - - siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(siddhiAppString.toString()); - parameters - .getInEventTypes() - .forEach((key, value) -> siddhiInputHandlers.put(key, siddhiAppRuntime.getInputHandler(prepareName(key)))); - - if (!debugMode) { - siddhiAppRuntime.addCallback(prepareName(getOutputTopicName(parameters)), new StreamCallback() { - @Override - public void receive(Event[] events) { - if (events.length > 0) { - Event lastEvent = events[events.length - 1]; - spOutputCollector.collect(toSpEvent(lastEvent, parameters, - runtimeContext.getOutputSchemaInfo - (), runtimeContext.getOutputSourceInfo())); - } - } - }); - } else { - siddhiAppRuntime.addCallback(prepareName(getOutputTopicName(parameters)), new StreamCallback() { - @Override - public void receive(Event[] events) { - LOG.info("Siddhi is firing"); - if (events.length > 0) { - SiddhiEventEngine.this.debugCallback.onEvent(events[events.length - 1]); - } - } - }); - } - - siddhiAppRuntime.start(); - - } - - private String removeStreamIdFromTimestamp(String timestampField) { - return timestampField !=null ? timestampField.replaceAll("s0::", "") : null; - } - - private String getOutputTopicName(B parameters) { - return parameters - .getGraph() - .getOutputStream() - .getEventGrounding() - .getTransportProtocol() - .getTopicDefinition() - .getActualTopicName(); - } - - private org.apache.streampipes.model.runtime.Event toSpEvent(Event event, B parameters, SchemaInfo - schemaInfo, SourceInfo sourceInfo) { - Map<String, Object> outMap = new HashMap<>(); - for (int i = 0; i < outputEventKeys.size(); i++) { - - if (event.getData(i) instanceof LinkedList) { - List<Object> tmp = (List<Object>) event.getData(i); - outMap.put(outputEventKeys.get(i), tmp.get(0)); - } - else { - outMap.put(outputEventKeys.get(i), event.getData(i)); - } - - } - return EventFactory.fromMap(outMap, sourceInfo, schemaInfo); - } - - - private void registerEventTypeIfNotExists(String eventTypeName, Map<String, Object> typeMap) { - String defineStreamPrefix = "define stream " + prepareName(eventTypeName); - StringJoiner joiner = new StringJoiner(","); - int currentNoOfStreams = this.listOfEventKeys.size(); - - List<String> sortedEventKeys = new ArrayList<>(); - for (String key : typeMap.keySet()) { - sortedEventKeys.add(key); - Collections.sort(sortedEventKeys); - } - - listOfEventKeys.put(eventTypeName, sortedEventKeys); - - for (String key : sortedEventKeys) { - // TODO: get timestamp field from user params - if(key.equalsIgnoreCase(this.timestampField)) { - joiner.add("s" + currentNoOfStreams + key + " LONG"); - } - else { - joiner.add("s" + currentNoOfStreams + key + " " + toType((Class<?>) typeMap.get(key))); - } - } - - this.siddhiAppString.append(defineStreamPrefix); - this.siddhiAppString.append("("); - this.siddhiAppString.append(joiner.toString()); - this.siddhiAppString.append(");\n"); - } - - private String toType(Class<?> o) { - if (o.equals(Long.class)) { - return "LONG"; - } else if (o.equals(Integer.class)) { - return "INT"; - } else if (o.equals(Double.class)) { - return "DOUBLE"; - } else if (o.equals(Float.class)) { - return "DOUBLE"; - } else if (o.equals(Boolean.class)) { - return "BOOL"; - } else { - return "STRING"; - } - } - - private void registerStatements(String fromStatement, String selectStatement, String outputStream) { - this.siddhiAppString.append(fromStatement) - .append("\n") - .append(selectStatement) - .append("\n") - .append("insert into ") - .append(prepareName(outputStream)) - .append(";"); - - LOG.info("Registering statement: \n" + this.siddhiAppString.toString()); - + String selectStatement = selectStatement(outputEventKeys, parameters); + this.siddhiConfig = new SiddhiInvocationConfig<>(parameters, fromStatement, selectStatement, inputStreamNames); + this.siddhiEngine.initializeEngine(this.siddhiConfig, spOutputCollector, runtimeContext); } @Override public void onEvent(org.apache.streampipes.model.runtime.Event event, SpOutputCollector collector) { - try { - String sourceId = event.getSourceInfo().getSourceId(); - InputHandler inputHandler = siddhiInputHandlers.get(sourceId); - List<String> eventKeys = listOfEventKeys.get(sourceId); - - inputHandler.send(toObjArr(eventKeys, event.getRaw())); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - - private Object[] toObjArr(List<String> eventKeys, Map<String, Object> event) { - Object[] result = new Object[eventKeys.size()]; - for (int i = 0; i < eventKeys.size(); i++) { - result[i] = event.get(eventKeys.get(i)); - } - - return result; + this.siddhiEngine.processEvent(event); } @Override public void onDetach() { - this.siddhiAppRuntime.shutdown(); + this.siddhiEngine.shutdownEngine(); } - protected abstract String fromStatement(List<String> inputStreamNames, final B bindingParameters); - - protected abstract String selectStatement(final B bindingParameters); - - protected String setTimestamp(final B bindingparameters) { - return null; + public SiddhiInvocationConfig<B> getSiddhiConfig() { + return this.siddhiConfig; } - protected String prepareName(String eventName) { - return eventName - .replaceAll("\\.", "") - .replaceAll("-", "") - .replaceAll("::", ""); + public String prepareName(String name) { + return SiddhiUtils.prepareName(name); } - - - protected String getCustomOutputSelectStatement(DataProcessorInvocation invocation, - String eventName) { + public String getCustomOutputSelectStatement(DataProcessorInvocation invocation) { StringBuilder selectString = new StringBuilder(); - selectString.append("select "); + selectString.append(SiddhiConstants.SELECT).append(" "); if (outputEventKeys.size() > 0) { - for (int i = 0; i < outputEventKeys.size() - 1; i++) { - selectString.append(eventName + ".s0" + outputEventKeys.get(i) + ","); + for (int i=0; i<outputEventKeys.size() - 1; i++) { + selectString + .append(SiddhiConstants.FIRST_STREAM_PREFIX) + .append(outputEventKeys.get(i)) + .append(","); } - selectString.append(eventName + ".s0" + outputEventKeys.get(outputEventKeys.size() - 1)); - + selectString + .append(SiddhiConstants.FIRST_STREAM_PREFIX) + .append(outputEventKeys.get(outputEventKeys.size() - 1)); } - return selectString.toString(); } -// protected String getCustomOutputSelectStatement(DataProcessorInvocation invocation) { -// return getCustomOutputSelectStatement(invocation, "e1"); -// } - - protected String getCustomOutputSelectStatement(DataProcessorInvocation invocation) { + public String getCustomOutputSelectStatement(DataProcessorInvocation invocation, + String eventName) { StringBuilder selectString = new StringBuilder(); - selectString.append("select "); + selectString.append(SiddhiConstants.SELECT).append(" "); if (outputEventKeys.size() > 0) { - for (int i=0; i<outputEventKeys.size() - 1; i++) { - selectString.append("s0" + outputEventKeys.get(i) + ","); + for (int i = 0; i < outputEventKeys.size() - 1; i++) { + selectString.append(eventName).append(".s0").append(outputEventKeys.get(i)).append(","); } - selectString.append("s0" + outputEventKeys.get(outputEventKeys.size() - 1)); - } - return selectString.toString(); - } + selectString.append(eventName).append(".s0").append(outputEventKeys.get(outputEventKeys.size() - 1)); - public void setSortedEventKeys(List<String> sortedEventKeys) { - String streamId = (String) this.listOfEventKeys.keySet().toArray()[0]; // only reliable if there is only one stream, else use changeEventKeys() to respective streamId - changeEventKeys(streamId, sortedEventKeys); - } + } - public void changeEventKeys(String streamId, List<String> newEventKeys) { - this.listOfEventKeys.put(streamId, newEventKeys); + return selectString.toString(); } } diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiInvocationConfig.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiInvocationConfig.java new file mode 100644 index 0000000..62eb29c --- /dev/null +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiInvocationConfig.java @@ -0,0 +1,74 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.streampipes.wrapper.siddhi.engine; + +import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams; +import org.apache.streampipes.wrapper.siddhi.engine.generator.EventTypeGenerator; +import org.apache.streampipes.wrapper.siddhi.engine.generator.SiddhiAppGenerator; +import org.apache.streampipes.wrapper.siddhi.model.EventType; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +public class SiddhiInvocationConfig<B extends EventProcessorBindingParams> { + + private final B params; + private final String siddhiAppString; + private final List<String> inputStreamNames; + private final Map<String, List<EventType>> eventTypeInfo; + private final List<String> outputEventKeys; + + public SiddhiInvocationConfig(B params, + String fromStatement, + String selectStatement, + List<String> inputStreamNames) { + this.params = params; + this.inputStreamNames = inputStreamNames; + this.eventTypeInfo = new EventTypeGenerator<>(params).generateEventTypes(); + this.siddhiAppString = new SiddhiAppGenerator<>(params, inputStreamNames, eventTypeInfo, fromStatement, selectStatement) + .generateSiddhiApp(); + this.outputEventKeys = new ArrayList<>(this.params.getOutEventType().keySet()); + } + + public B getParams() { + return params; + } + + public String getSiddhiAppString() { + return siddhiAppString; + } + + public List<String> getInputStreamNames() { + return inputStreamNames; + } + + public Map<String, List<EventType>> getEventTypeInfo() { + return eventTypeInfo; + } + + public List<String> getOutputEventKeys() { + return this.outputEventKeys; + } + + public String getSiddhiApp() { + return this.siddhiAppString; + } + + +} diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiStatementGenerator.java similarity index 70% copy from streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java copy to streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiStatementGenerator.java index 540d608..d7adcb9 100644 --- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiStatementGenerator.java @@ -17,10 +17,14 @@ */ package org.apache.streampipes.wrapper.siddhi.engine; +import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams; -import io.siddhi.core.event.Event; +import java.util.List; -public interface SiddhiDebugCallback { +public interface SiddhiStatementGenerator<B extends EventProcessorBindingParams> { + + String fromStatement(List<String> inputStreamNames, final B bindingParameters); + + String selectStatement(List<String> outputEventKeys, final B bindingParameters); - void onEvent(Event event); } diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/StreamPipesSiddhiProcessor.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/StreamPipesSiddhiProcessor.java new file mode 100644 index 0000000..d603e9a --- /dev/null +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/StreamPipesSiddhiProcessor.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.streampipes.wrapper.siddhi.engine; + +import org.apache.streampipes.commons.exceptions.SpRuntimeException; +import org.apache.streampipes.model.runtime.Event; +import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext; +import org.apache.streampipes.wrapper.routing.SpOutputCollector; +import org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback; +import org.apache.streampipes.wrapper.standalone.ProcessorParams; +import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor; + +public abstract class StreamPipesSiddhiProcessor extends StreamPipesDataProcessor implements SiddhiStatementGenerator<ProcessorParams> { + + private SiddhiEngine siddhiEngine; + + public StreamPipesSiddhiProcessor() { + this.siddhiEngine = new SiddhiEngine(); + } + + public StreamPipesSiddhiProcessor(SiddhiDebugCallback debugCallback) { + this.siddhiEngine = new SiddhiEngine(debugCallback); + } + + @Override + public void onInvocation(ProcessorParams parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException { + + } + + @Override + public void onEvent(Event event, SpOutputCollector collector) throws SpRuntimeException { + this.siddhiEngine.processEvent(event); + } + + @Override + public void onDetach() throws SpRuntimeException { + this.siddhiEngine.shutdownEngine(); + } +} diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/callback/SiddhiDebugCallback.java similarity index 93% copy from streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java copy to streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/callback/SiddhiDebugCallback.java index 540d608..838071f 100644 --- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/callback/SiddhiDebugCallback.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.wrapper.siddhi.engine; +package org.apache.streampipes.wrapper.siddhi.engine.callback; import io.siddhi.core.event.Event; diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/callback/SiddhiOutputStreamCallback.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/callback/SiddhiOutputStreamCallback.java new file mode 100644 index 0000000..947e99a --- /dev/null +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/callback/SiddhiOutputStreamCallback.java @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.streampipes.wrapper.siddhi.engine.callback; + +import io.siddhi.core.event.Event; +import io.siddhi.core.stream.output.StreamCallback; +import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext; +import org.apache.streampipes.wrapper.routing.SpOutputCollector; +import org.apache.streampipes.wrapper.siddhi.utils.SiddhiUtils; + +import java.util.List; + +public class SiddhiOutputStreamCallback extends StreamCallback { + + private SpOutputCollector collector; + private List<String> outputEventKeys; + private EventProcessorRuntimeContext runtimeContext; + + public SiddhiOutputStreamCallback(SpOutputCollector collector, + List<String> outputEventKeys, + EventProcessorRuntimeContext runtimeContext) { + this.collector = collector; + this.outputEventKeys = outputEventKeys; + this.runtimeContext = runtimeContext; + } + + @Override + public void receive(Event[] events) { + if (events.length > 0) { + Event lastEvent = events[events.length - 1]; + collector.collect(SiddhiUtils.toSpEvent(lastEvent, outputEventKeys, + runtimeContext.getOutputSchemaInfo + (), runtimeContext.getOutputSourceInfo())); + } + } +} diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/callback/SiddhiOutputStreamDebugCallback.java similarity index 55% copy from streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java copy to streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/callback/SiddhiOutputStreamDebugCallback.java index 540d608..a79f911 100644 --- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/callback/SiddhiOutputStreamDebugCallback.java @@ -15,12 +15,28 @@ * limitations under the License. * */ -package org.apache.streampipes.wrapper.siddhi.engine; - +package org.apache.streampipes.wrapper.siddhi.engine.callback; import io.siddhi.core.event.Event; +import io.siddhi.core.stream.output.StreamCallback; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class SiddhiOutputStreamDebugCallback extends StreamCallback { + + private static final Logger LOG = LoggerFactory.getLogger(SiddhiOutputStreamDebugCallback.class); + + private SiddhiDebugCallback callback; -public interface SiddhiDebugCallback { + public SiddhiOutputStreamDebugCallback(SiddhiDebugCallback callback) { + this.callback = callback; + } - void onEvent(Event event); + @Override + public void receive(Event[] events) { + LOG.info("Siddhi is firing"); + if (events.length > 0) { + this.callback.onEvent(events[events.length - 1]); + } + } } diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/EventTypeGenerator.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/EventTypeGenerator.java new file mode 100644 index 0000000..1aa95a4 --- /dev/null +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/EventTypeGenerator.java @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.streampipes.wrapper.siddhi.engine.generator; + +import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams; +import org.apache.streampipes.wrapper.siddhi.constants.SiddhiConstants; +import org.apache.streampipes.wrapper.siddhi.model.EventType; + +import java.util.*; +import java.util.concurrent.atomic.AtomicReference; + +public class EventTypeGenerator<B extends EventProcessorBindingParams> { + + private B params; + + public EventTypeGenerator(B params) { + this.params = params; + } + + public Map<String, List<EventType>> generateEventTypes() { + Map<String, List<EventType>> listOfEventKeys = new HashMap<>(); + AtomicReference<Integer> currentStreamIndex = new AtomicReference<>(0); + + params.getInEventTypes().forEach((key, value) -> { + List<EventType> sortedEventKeys = new ArrayList<>(); + for (String propertyKey : value.keySet()) { + sortedEventKeys.add(makeEventType(currentStreamIndex.get(), propertyKey, value.get(propertyKey))); + sortedEventKeys.sort(Comparator.comparing(EventType::getEventTypeName)); + } + listOfEventKeys.put(key, sortedEventKeys); + currentStreamIndex.getAndSet(currentStreamIndex.get() + 1); + }); + + return listOfEventKeys; + } + + private EventType makeEventType(Integer currentStreamIndex, String propertyName, Object propertyType) { + return new EventType(currentStreamIndex, propertyName, toType((Class<?>) propertyType)); + } + + private String toType(Class<?> o) { + if (o.equals(Long.class)) { + return SiddhiConstants.SIDDHI_LONG_TYPE; + } else if (o.equals(Integer.class)) { + return SiddhiConstants.SIDDHI_INT_TYPE; + } else if (o.equals(Double.class)) { + return SiddhiConstants.SIDDHI_DOUBLE_TYPE; + } else if (o.equals(Float.class)) { + return SiddhiConstants.SIDDHI_DOUBLE_TYPE; + } else if (o.equals(Boolean.class)) { + return SiddhiConstants.SIDDHI_BOOLEAN_TYPE; + } else if (o.equals(String.class)){ + return SiddhiConstants.SIDDHI_STRING_TYPE; + } else { + return SiddhiConstants.SIDDHI_OBJECT_TYPE; + } + } +} diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/InputStreamNameGenerator.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/InputStreamNameGenerator.java new file mode 100644 index 0000000..0bd8ee9 --- /dev/null +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/InputStreamNameGenerator.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.streampipes.wrapper.siddhi.engine.generator; + +import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams; +import org.apache.streampipes.wrapper.siddhi.utils.SiddhiUtils; + +import java.util.ArrayList; +import java.util.List; + +public class InputStreamNameGenerator<B extends EventProcessorBindingParams> { + + private B params; + + public InputStreamNameGenerator(B params) { + this.params = params; + } + + public List<String> generateInputStreamNames() { + List<String> inputStreamNames = new ArrayList<>(); + this.params.getInEventTypes().forEach((key, value) -> { + // TODO why is the prefix not in the parameters.getInEventType + //registerEventTypeIfNotExists(key, value); + inputStreamNames.add(SiddhiUtils.prepareName(key)); + }); + + return inputStreamNames; + } +} diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/SiddhiAppGenerator.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/SiddhiAppGenerator.java new file mode 100644 index 0000000..0c54f16 --- /dev/null +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/SiddhiAppGenerator.java @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.streampipes.wrapper.siddhi.engine.generator; + +import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams; +import org.apache.streampipes.wrapper.siddhi.model.EventType; +import org.apache.streampipes.wrapper.siddhi.utils.SiddhiUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.Map; +import java.util.StringJoiner; + +public class SiddhiAppGenerator<B extends EventProcessorBindingParams> { + + private static final Logger LOG = LoggerFactory.getLogger(SiddhiAppGenerator.class); + + private B params; + private List<String> inputStreamNames; + private Map<String, List<EventType>> eventTypes; + private final String fromStatement; + private final String selectStatement; + + private final StringBuilder siddhiAppString; + + public SiddhiAppGenerator(B params, + List<String> inputStreamNames, + Map<String, List<EventType>> eventTypes, + String fromStatement, + String selectStatement) { + this.params = params; + this.inputStreamNames = inputStreamNames; + this.eventTypes = eventTypes; + this.fromStatement = fromStatement; + this.selectStatement = selectStatement; + this.siddhiAppString = new StringBuilder(); + } + + public String generateSiddhiApp() { + LOG.info("Configuring event types for graph " + params.getGraph().getName()); + + this.eventTypes.forEach(this::registerEventType); + registerStatements(fromStatement, selectStatement, SiddhiUtils.getOutputTopicName(params)); + + return this.siddhiAppString.toString(); + } + + private void registerEventType(String eventTypeName, List<EventType> eventSchema) { + String defineStreamPrefix = "define stream " + SiddhiUtils.prepareName(eventTypeName); + StringJoiner joiner = new StringJoiner(","); + + eventSchema.forEach(typeInfo -> { + joiner.add("s" + typeInfo.getStreamIdentifier() + typeInfo.getEventTypeName() + " " + typeInfo.getEventType()); + }); + + this.siddhiAppString + .append(defineStreamPrefix) + .append("(") + .append(joiner.toString()) + .append(");\n"); + } + + private void registerStatements(String fromStatement, String selectStatement, String outputStream) { + this.siddhiAppString + .append(fromStatement) + .append("\n") + .append(selectStatement) + .append("\n") + .append("insert into ") + .append(SiddhiUtils.prepareName(outputStream)) + .append(";"); + + LOG.info("Registering statement: \n" + this.siddhiAppString.toString()); + + } +} diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/model/EventType.java similarity index 58% rename from streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java rename to streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/model/EventType.java index 540d608..1264caf 100644 --- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/SiddhiDebugCallback.java +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/model/EventType.java @@ -15,12 +15,30 @@ * limitations under the License. * */ -package org.apache.streampipes.wrapper.siddhi.engine; +package org.apache.streampipes.wrapper.siddhi.model; +public class EventType { -import io.siddhi.core.event.Event; + private Integer streamIdentifier; + private String eventTypeName; + private String eventType; -public interface SiddhiDebugCallback { + public EventType(Integer streamIdentifier, String eventTypeName, String eventType) { + this.streamIdentifier = streamIdentifier; + this.eventTypeName = eventTypeName; + this.eventType = eventType; + } + + public Integer getStreamIdentifier() { + return streamIdentifier; + } + + public String getEventTypeName() { + return eventTypeName; + } + + public String getEventType() { + return eventType; + } - void onEvent(Event event); } diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/utils/SiddhiUtils.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/utils/SiddhiUtils.java new file mode 100644 index 0000000..2292f18 --- /dev/null +++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/utils/SiddhiUtils.java @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.streampipes.wrapper.siddhi.utils; + +import io.siddhi.core.event.Event; +import org.apache.streampipes.model.runtime.EventFactory; +import org.apache.streampipes.model.runtime.SchemaInfo; +import org.apache.streampipes.model.runtime.SourceInfo; +import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class SiddhiUtils { + + public static org.apache.streampipes.model.runtime.Event toSpEvent(Event event, List<String> outputEventKeys, SchemaInfo + schemaInfo, SourceInfo sourceInfo) { + Map<String, Object> outMap = new HashMap<>(); + + for (int i = 0; i < outputEventKeys.size(); i++) { + if (event.getData(i) instanceof List) { + List<Object> tmp = (List<Object>) event.getData(i); + outMap.put(outputEventKeys.get(i), tmp.get(0)); + } + else { + outMap.put(outputEventKeys.get(i), event.getData(i)); + } + } + return EventFactory.fromMap(outMap, sourceInfo, schemaInfo); + } + + public static Object[] toObjArr(List<String> eventKeys, Map<String, Object> event) { + Object[] result = new Object[eventKeys.size()]; + for (int i = 0; i < eventKeys.size(); i++) { + result[i] = event.get(eventKeys.get(i)); + } + + return result; + } + + public static String getOutputTopicName(EventProcessorBindingParams parameters) { + return parameters + .getGraph() + .getOutputStream() + .getEventGrounding() + .getTransportProtocol() + .getTopicDefinition() + .getActualTopicName(); + } + + public static String prepareName(String eventName) { + return eventName + .replaceAll("\\.", "") + .replaceAll("-", "") + .replaceAll("::", ""); + } +}
