[ADD] List of event keys for working with multiple streams
@@ -45,7 +45,7 @@ public abstract class SiddhiEventEngine<B extends EventProcessorBindingParams> i
private Map<String, InputHandler> siddhiInputHandlers;
private List<String> inputStreamNames;
- private List<String> sortedEventKeys;
+ private Map<String, List> listOfEventKeys;
private List<String> outputEventKeys;
private Boolean debugMode;
@@ -59,7 +59,7 @@ public abstract class SiddhiEventEngine<B extends EventProcessorBindingParams> i
this.siddhiAppString = new StringBuilder();
this.siddhiInputHandlers = new HashMap<>();
this.inputStreamNames = new ArrayList<>();
- sortedEventKeys = new ArrayList<>();
+ listOfEventKeys = new HashMap<>();
outputEventKeys = new ArrayList<>();
this.debugMode = false;
}
@@ -164,11 +164,14 @@ public abstract class SiddhiEventEngine<B extends EventProcessorBindingParams> i
String defineStreamPrefix = "define stream " + prepareName(eventTypeName);
StringJoiner joiner = new StringJoiner(",");
+ List<String> sortedEventKeys = new ArrayList<>();
for (String key : typeMap.keySet()) {
sortedEventKeys.add(key);
Collections.sort(sortedEventKeys);
}
+ listOfEventKeys.put(prepareName(eventTypeName), sortedEventKeys)
+
for (String key : sortedEventKeys) {
// TODO: get timestamp field from user params
if(key.equalsIgnoreCase(this.timestampField)) {
@@ -217,16 +220,20 @@ public abstract class SiddhiEventEngine<B extends EventProcessorBindingParams> i
@Override
public void onEvent(org.apache.streampipes.model.runtime.Event event, SpOutputCollector collector) {
try {
- siddhiInputHandlers.get(event.getSourceInfo().getSourceId()).send(toObjArr(event.getRaw()));
+ String sourceId = event.getSourceInfo().getSourceId()
+ InputHandler inputHandler = siddhiInputHandlers.get(sourceId);
+ List<String> eventKeys = listOfEventKeys.get(prepareName(sourceId));
+
+ inputHandler.send(toObjArr(eventKeys, event.getRaw()));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
- private Object[] toObjArr(Map<String, Object> event) {
- Object[] result = new Object[sortedEventKeys.size()];
- for (int i = 0; i < sortedEventKeys.size(); i++) {
- result[i] = event.get(sortedEventKeys.get(i));
+ 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;
