Merge pull request #50 from 317787106/hotfix/verify_data

hotfix: fix the bug of triggerProcessLoop

Author: CodeNinjaEvan

README.md

@@ -1,29 +1,28 @@
-# Tron eventsubscribe plugin
+# Tron event subscribe plugin
 
-This is an implementation of Tron eventsubscribe model. 
+This is an implementation of Tron event subscribe model.
 
 * **api** module defines IPluginEventListener, a protocol between Java-tron and event plugin. 
 * **app** module is an example for loading plugin, developers could use it for debugging.
 * **kafkaplugin** module is the implementation for kafka, it implements IPluginEventListener, it receives events subscribed from Java-tron and relay events to kafka server. 
 * **mongodbplugin** mongodbplugin module is the implementation for mongodb. 
 ### Setup/Build
-Eventplugin can be built with JDK 8 or JDK17.
+Event-plugin can be built with JDK 8 or JDK 17.
 1. Clone the repo
 2. Go to eventplugin `cd eventplugin` 
 3. run `./gradlew build`
 
 * This will produce plugin zips, named `plugin-kafka-1.0.0.zip` and `plugin-mongodb-1.0.0.zip`, located in the `eventplugin/build/plugins/` directory.
 
 
-### Edit **config.conf** of Java-tron, add the following fileds:
+### Edit **config.conf** of Java-tron, add the following fields:
 ```
 event.subscribe = {
     path = "" // absolute path of plugin
     server = "" // target server address to receive event triggers
-    # dbname|username|password, if you want to create indexes for collections when the collections
-    # are not exist, you can add version and set it to 2, as dbname|username|password|version
-    # if you use version 2 and one collection not exists, it will create index automaticaly;
-    # if you use version 2 and one collection exists, it will not create index, you must create index manually;
+    # dbname|username|password or dbname|username|password|version
+    # If you use version 2 and one collection not exists, it will create index automatically;
+    # In any other case, it will not create index, you must create index manually
     dbconfig = ""
     topics = [
         {
@@ -99,13 +98,14 @@ event.subscribe = {
            remove comment and set listeners=PLAINTEXT://:9092  
            remove comment and set advertised.listeners to PLAINTEXT://host_ip:9092 
 
+### How to use kafka plugin
 ##### Install Kafka
-**On Mac**:
+*On Mac*:
 ```
 brew install kafka
 ```
 
-**On Linux**:
+*On Linux*:
 ```
 cd /usr/local
 wget http://archive.apache.org/dist/kafka/0.10.2.2/kafka_2.10-0.10.2.2.tgz
@@ -119,12 +119,12 @@ source /etc/profile
 **Note**: make sure the version of Kafka is the same as the version set in build.gradle of eventplugin project.(kafka_2.10-0.10.2.2 kafka)
 
 ##### Run Kafka
-**On Mac**:
+*On Mac*:
 ```
 zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties & kafka-server-start /usr/local/etc/kafka/server.properties
 ```
 
-**On Linux**:
+*On Linux*:
 ```
 zookeeper-server-start.sh /usr/local/kafka/config/zookeeper.properties &
 Sleep about 3 seconds 
@@ -133,7 +133,7 @@ kafka-server-start.sh /usr/local/kafka/config/server.properties &
 
 #### Create topics to receive events, the topic is defined in config.conf
 
-**On Mac**:
+*On Mac*:
 ```
 kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic block
 kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic transaction
@@ -144,7 +144,7 @@ kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partit
 kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic soliditylog
 ```
 
-**On Linux**:
+*On Linux*:
 ```
 kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic block
 kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic transaction
@@ -157,7 +157,7 @@ kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --par
 
 #### Kafka consumer
 
-**On Mac**:
+*On Mac*:
 ```
 kafka-console-consumer --bootstrap-server localhost:9092  --topic block
 kafka-console-consumer --bootstrap-server localhost:9092  --topic transaction
@@ -168,7 +168,7 @@ kafka-console-consumer --bootstrap-server localhost:9092  --topic solidityevent
 kafka-console-consumer --bootstrap-server localhost:9092  --topic soliditylog
 ```
 
-**On Linux**:
+*On Linux*:
 ```
 kafka-console-consumer.sh --zookeeper localhost:2181 --topic block
 kafka-console-consumer.sh --zookeeper localhost:2181 --topic transaction
@@ -179,6 +179,29 @@ kafka-console-consumer.sh --zookeeper localhost:2181 --topic solidityevent
 kafka-console-consumer.sh --zookeeper localhost:2181 --topic soliditylog
 ```
 
+See more details on [developers](https://developers.tron.network/docs/event-plugin-deployment-kafka).
+
+### How to use MongoDB plugin
+These are default indexes when build automatically:
+```
+db.block.createIndex({ blockNumber: 1 },{ name: "blockNumber",unique: true});
+
+db.transaction.createIndex({ transactionId: 1 },{ name: "transactionId",unique: true });
+
+db.solidity.createIndex({ latestSolidifiedBlockNumber: 1 },{ name: "latestSolidifiedBlockNumber",unique: true });
+
+db.solidityevent.createIndex({ uniqueId: 1 },{ name: "uniqueId",unique: true });
+
+db.contractevent.createIndex({ uniqueId: 1 },{ name: "uniqueId",unique: true });
+
+db.soliditylog.createIndex({ uniqueId: 1 },{ name: "uniqueId",unique: true });
+db.soliditylog.createIndex({ contractAddress: 1 },{ name: "contractAddress" });
+
+db.contractlog.createIndex({ uniqueId: 1 },{ name: "uniqueId",unique: true });
+db.contractlog.createIndex({ contractAddress: 1 },{ name: "contractAddress" });
+```
+You can also create other indexes as necessary. See more details on [developers](https://developers.tron.network/docs/event-plugin-deployment-mongodb).
+
 ### Load plugin in Java-tron
 * add --es to command line, for example:
 ```

api/src/main/java/org/tron/common/logsfilter/IPluginEventListener.java

@@ -13,6 +13,8 @@ public interface IPluginEventListener extends ExtensionPoint {
   // start should be called after setServerAddress, setTopic, setDBConfig
   void start();
 
+  void stop();
+
   void handleBlockEvent(Object data);
 
   void handleTransactionTrigger(Object data);

api/src/main/java/org/tron/common/logsfilter/trigger/BlockLogTrigger.java

@@ -1,9 +1,7 @@
 package org.tron.common.logsfilter.trigger;
 
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import lombok.Getter;
 import lombok.Setter;
 
@@ -23,22 +21,11 @@ public class BlockLogTrigger extends Trigger {
 
   @Getter
   @Setter
-  private List<String> transactionList = new ArrayList<>();
-
-  /**
-   * address of witness
-   */
-  @Getter
-  @Setter
-  private String witnessAddress;
+  private long latestSolidifiedBlockNumber;
 
   @Getter
   @Setter
-  private long witnessPayPerBlock;
-
-  @Getter
-  @Setter
-  Map<String, Long> witnessMap = new HashMap<>();
+  private List<String> transactionList = new ArrayList<>();
 
   public BlockLogTrigger() {
     setTriggerName(Trigger.BLOCK_TRIGGER_NAME);
@@ -51,9 +38,7 @@ public String toString() {
         + ", blockNumber: " + blockNumber
         + ", blockhash: " + blockHash
         + ", transactionSize: " + transactionSize
-        + ", transactionList: " + transactionList
-        + ", witnessAddress: " + witnessAddress
-        + ", witnessPayPerBlock: " + witnessPayPerBlock
-        + ", witnessMap: " + witnessMap;
+        + ", latestSolidifiedBlockNumber: " + latestSolidifiedBlockNumber
+        + ", transactionList: " + transactionList;
   }
 }

app/build.gradle

@@ -3,33 +3,38 @@ apply plugin: 'application'
 mainClassName = 'org.tron.eventplugin.app.PluginLauncher'
 
 dependencies {
-  implementation project(':api')
-  implementation (group: 'org.pf4j', name: 'pf4j', version: "${pf4jVersion}") {
-    exclude group: "org.slf4j"
-  }
-  implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.5'
-  implementation group: 'org.slf4j', name: 'slf4j-simple', version: '1.7.25'
-  implementation group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.18.3'
-  implementation group: 'com.fasterxml.jackson.core', name: 'jackson-core', version: '2.18.3'
-  implementation "com.madgag.spongycastle:core:1.58.0.0"
-  implementation "com.madgag.spongycastle:prov:1.58.0.0"
+    implementation project(':api')
+
+    implementation(group: 'org.pf4j', name: 'pf4j', version: "${pf4jVersion}")
+    implementation 'org.apache.commons:commons-lang3:3.5'
+    implementation 'org.slf4j:slf4j-api:2.0.9'
+    implementation 'ch.qos.logback:logback-classic:1.3.14'
+    implementation 'com.fasterxml.jackson.core:jackson-databind:2.18.3'
+    implementation 'com.fasterxml.jackson.core:jackson-core:2.18.3'
+    implementation "com.madgag.spongycastle:core:1.58.0.0"
+    implementation "com.madgag.spongycastle:prov:1.58.0.0"
+    implementation 'com.alibaba:fastjson:1.2.83'
 }
 
 task uberjar(type: Jar, dependsOn: ['compileJava']) {
-  zip64 true
-  from configurations.runtimeClasspath.asFileTree.files.collect {
+    zip64 true
+
+    from {
+        configurations.runtimeClasspath.collect { file ->
+            file.isDirectory() ? file : zipTree(file)
+        }
+    }
+
     exclude "META-INF/*.SF"
     exclude "META-INF/*.DSA"
     exclude "META-INF/*.RSA"
-    zipTree(it)
-  }
-  from files(sourceSets.main.output.classesDirs)
-  from files(sourceSets.main.resources)
-  manifest {
-    attributes 'Main-Class': mainClassName
-  }
-
-  baseName = "${project.name}-plugin-demo"
-  classifier = "uberjar"
-}
 
+    from sourceSets.main.output
+
+    manifest {
+        attributes 'Main-Class': mainClassName
+    }
+
+    archiveBaseName.set("${project.name}-plugin-demo")
+    archiveClassifier.set("uberjar")
+}

app/src/main/java/org/tron/eventplugin/app/PluginLauncher.java

@@ -16,24 +16,24 @@
 
 package org.tron.eventplugin.app;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.File;
 import java.util.List;
 import java.util.Objects;
+import java.util.Random;
+import lombok.extern.slf4j.Slf4j;
 import org.pf4j.CompoundPluginDescriptorFinder;
 import org.pf4j.DefaultPluginManager;
 import org.pf4j.ManifestPluginDescriptorFinder;
 import org.pf4j.PluginManager;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.tron.common.logsfilter.IPluginEventListener;
 import org.tron.common.logsfilter.trigger.BlockLogTrigger;
 import org.tron.common.logsfilter.trigger.Trigger;
 
+@Slf4j
 public class PluginLauncher {
 
-  private static final Logger logger = LoggerFactory.getLogger(PluginLauncher.class);
-
   public static void main(String[] args) {
     String path = "/Users/tron/sourcecode/eventplugin/build/plugins/plugin-mongodb-1.0.0.zip";
 
@@ -55,6 +55,7 @@ protected CompoundPluginDescriptorFinder createPluginDescriptorFinder() {
     List<IPluginEventListener> eventListeners;
     eventListeners = pluginManager.getExtensions(IPluginEventListener.class);
 
+    log.info("start plugin...");
     if (Objects.isNull(eventListeners)) {
       return;
     }
@@ -80,30 +81,33 @@ protected CompoundPluginDescriptorFinder createPluginDescriptorFinder() {
     eventListeners.forEach(IPluginEventListener::start);
 
     ObjectMapper objectMapper = new ObjectMapper();
-    for (int index = 0; index < 1000; ++index) {
+    for (int index = 0; index < 2; ++index) {
       BlockLogTrigger trigger = new BlockLogTrigger();
-      trigger.setBlockNumber(index);
-      trigger.setBlockHash("000000000002f5834df6036318999576bfa23ff1a57e0538fa87d5a90319659e");
+      trigger.setBlockNumber(new Random().nextInt(10000)); //blockNumber is unique key
+      trigger.setBlockHash("000000000002f5834df6036318999576bfa23ff1a57e0538fa87d5a90319659f");
       trigger.setTimeStamp(System.currentTimeMillis());
       trigger.setTransactionSize(100);
 
+      String triggerData;
+      try {
+        triggerData = objectMapper.writeValueAsString(trigger);//convert to json
+      } catch (JsonProcessingException e) {
+        log.error("", e);
+        continue;
+      }
       eventListeners.forEach(listener -> {
-        try {
-          listener.handleBlockEvent(objectMapper.writeValueAsString(trigger));
-        } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
-          e.printStackTrace();
-        }
+        listener.handleBlockEvent(triggerData);
       });
     }
 
-    while (true) {
-      try {
-        Thread.sleep(1000);
-      } catch (InterruptedException e) {
-        e.printStackTrace();
-      }
+    try {
+      Thread.sleep(10_000);
+    } catch (InterruptedException e) {
+      //ignore
     }
+    log.info("try to close plugin...");
 
-    //pluginManager.stopPlugins();
+    // will invoke stop method of KafkaLogFilterPlugin or MongodbLogFilterPlugin
+    pluginManager.stopPlugins();
   }
 }

app/src/main/resources/log4j.properties

@@ -0,0 +1,18 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<configuration>
+
+  <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <pattern>
+        %d{HH:mm:ss.SSS} %-5level [%t] [%c{1}]\(%F:%L\) %m%n
+      </pattern>
+    </encoder>
+  </appender>
+
+  <root level="INFO">
+    <appender-ref ref="CONSOLE"/>
+  </root>
+
+  <logger name="org.pf4j" level="DEBUG"/>
+</configuration>