Merge pull request #29 from keyko-io/feature/flat_event_record

flatten eventBlock record

Author: josepablofm78

pom.xml

@@ -3,7 +3,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>io.keyko.monitoring</groupId>
   <artifactId>web3-event-streamer</artifactId>
-  <version>0.2.6</version>
+  <version>0.2.7</version>
   <name>Web3 Monitoring Event Streamer</name>
   <url>https://github.com/keyko-io/web3-event-streamer</url>
   <inceptionYear>2019</inceptionYear>
@@ -20,7 +20,7 @@
     <org.json.version>20190722</org.json.version>
     <typesafe.config.version>1.4.0</typesafe.config.version>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    <keyko.schemas.version>0.3.1</keyko.schemas.version>
+    <keyko.schemas.version>0.3.3</keyko.schemas.version>
     <junit.version>4.4</junit.version>
   </properties>
 

src/main/java/io/keyko/monitoring/preprocessing/Transformations.java

@@ -12,6 +12,8 @@
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.log4j.Logger;
 
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -247,4 +249,99 @@ public static KStream<String, LogRecordTopicsFlattened> flatLogs(KStream<String,
 
   }
 
+  public static KStream<String, FlatEventBlockRecord> flatEventBlockRecord(KStream<String, EventBlockRecord> eventBlockStream) {
+
+    return eventBlockStream.mapValues( eventBlock -> {
+
+      FlatEventBlockRecord flatEventBlockRecord = new FlatEventBlockRecord();
+
+      flatEventBlockRecord.setAddress(eventBlock.getEvent().getAddress());
+      flatEventBlockRecord.setBlockHash(eventBlock.getEvent().getBlockHash());
+      flatEventBlockRecord.setBlockNumber(eventBlock.getEvent().getBlockNumber());
+      flatEventBlockRecord.setContractName(eventBlock.getEvent().getContractName());
+      flatEventBlockRecord.setEventSpecificationSignature(eventBlock.getEvent().getEventSpecificationSignature());
+      flatEventBlockRecord.setFilterId(eventBlock.getEvent().getFilterId());
+      flatEventBlockRecord.setId(eventBlock.getEvent().getId());
+      flatEventBlockRecord.setLogIndex(eventBlock.getEvent().getLogIndex());
+      flatEventBlockRecord.setName(eventBlock.getEvent().getName());
+      flatEventBlockRecord.setNetworkName(eventBlock.getEvent().getNetworkName());
+      flatEventBlockRecord.setNodeName(eventBlock.getEvent().getNodeName());
+      flatEventBlockRecord.setRetries(eventBlock.getEvent().getRetries());
+      flatEventBlockRecord.setStatus(getEventStatus(eventBlock.getEvent().getStatus()));
+      flatEventBlockRecord.setTimestamp(eventBlock.getBlock().getTimestamp());
+
+      List<Object> indexedParameteres = eventBlock.getEvent().getIndexedParameters();
+      List<Object> nonIndexedParameteres = eventBlock.getEvent().getNonIndexedParameters();
+
+      setParametersOfFlatEvent(flatEventBlockRecord, indexedParameteres, true);
+      setParametersOfFlatEvent(flatEventBlockRecord, nonIndexedParameteres, false);
+
+      return flatEventBlockRecord;
+
+    });
+
+  }
+
+  private static void setParametersOfFlatEvent(FlatEventBlockRecord flatEvent, List<Object> parameters, Boolean indexed)  {
+
+
+    Integer maxIndexParameters = 4;
+    Integer maxNonIndexParameters = 5;
+
+    Integer max = indexed?maxIndexParameters:maxNonIndexParameters;
+
+    String nameMethod = indexed?"setIndexedParamName":"setNonIndexedParamName";
+    String typeMethod = indexed?"setIndexedParamType":"setNonIndexedParamType";
+    String valueMethod = indexed?"setIndexedParamValue":"setNonIndexedParamValue";
+
+    Method setNameMethod;
+    Method setTypeMethod;
+    Method setValueMethod;
+
+    String paramName = "";
+    String paramType = "";
+    String paramValue = "";
+
+    for (int i=0; i< parameters.size() && i< max; i++){
+
+      try {
+        setNameMethod = FlatEventBlockRecord.class.getMethod(nameMethod + i, String.class);
+        setTypeMethod = FlatEventBlockRecord.class.getMethod(typeMethod + i, String.class);
+        setValueMethod = FlatEventBlockRecord.class.getMethod(valueMethod + i, String.class);
+
+        if (parameters.get(i) instanceof NumberParameter) {
+          NumberParameter nParam = (NumberParameter) parameters.get(i);
+          paramName = nParam.getName();
+          paramType = nParam.getType();
+          paramValue = nParam.getValue();
+        } else {
+          StringParameter sParam = (StringParameter) parameters.get(i);
+          paramName = sParam.getName();
+          paramType = sParam.getType();
+          paramValue = sParam.getValue();
+        }
+
+        setNameMethod.invoke(flatEvent, paramName);
+        setTypeMethod.invoke(flatEvent, paramType);
+        setValueMethod.invoke(flatEvent, paramValue);
+
+      }catch (Exception e){
+        log.error("Error setting parameter " + e.getMessage());
+        continue;
+      }
+
+    }
+
+  }
+
+  private static FlatContractEventBlockStatus getEventStatus(ContractEventStatus status){
+    switch(status){
+      case CONFIRMED: return FlatContractEventBlockStatus.CONFIRMED;
+      case UNCONFIRMED: return FlatContractEventBlockStatus.UNCONFIRMED;
+      case INVALIDATED: return FlatContractEventBlockStatus.INVALIDATED;
+    }
+
+    return FlatContractEventBlockStatus.UNCONFIRMED;
+  }
+
   }

src/main/java/io/keyko/monitoring/serde/Web3MonitoringSerdes.java

@@ -20,6 +20,7 @@ public class Web3MonitoringSerdes {
   private final static SpecificAvroSerde<TimeSeriesRecord> timeSeriesSerde = new SpecificAvroSerde<>();
   private final static SpecificAvroSerde<LogRecord> logSerde = new SpecificAvroSerde<>();
   private final static SpecificAvroSerde<LogRecordTopicsFlattened> logFlattenedSerde = new SpecificAvroSerde<>();
+  private final static SpecificAvroSerde<FlatEventBlockRecord> flatEventBlockSerde = new SpecificAvroSerde<>();
   protected static Map<String, String> serdeConfig;
 
 
@@ -36,6 +37,7 @@ public static void configureSerdes(String schemaRegistryUrl) {
     timeSeriesSerde.configure(serdeConfig, false);
     logSerde.configure(serdeConfig, false);
     logFlattenedSerde.configure(serdeConfig, false);
+    flatEventBlockSerde.configure(serdeConfig, false);
   }
 
   protected static void configureSerde(SpecificAvroSerde serde) {
@@ -74,5 +76,9 @@ public static SpecificAvroSerde<LogRecordTopicsFlattened> getLogFlattenedSerde()
     return logFlattenedSerde;
   }
 
+  public static SpecificAvroSerde<FlatEventBlockRecord> getFlatEventBlockSerde() {
+    return flatEventBlockSerde;
+  }
+
 
 }