add interface to set dbconfig to plugin

Author: jiangyy0824

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

@@ -7,6 +7,11 @@ public interface IPluginEventListener extends ExtensionPoint {
 
     public void setTopic(int eventType, String topic);
 
+    public void setDBConfig(String dbConfig);
+
+    // start should be called after setServerAddress, setTopic, setDBConfig
+    public void start();
+
     public void handleBlockEvent(Object data);
 
     public void handleTransactionTrigger(Object data);

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

@@ -34,7 +34,7 @@ public class PluginLauncher {
     private static final Logger logger = LoggerFactory.getLogger(PluginLauncher.class);
 
     public static void main(String[] args) {
-        String path = "/Users/tron/workplace/java-tronSubmit/java-tronUseSubmit/develop_event_subscribe/eventplugin/build/plugins/plugin-mongodb-1.0.0.zip";
+        String path = "/Users/tron/sourcecode/eventplugin/build/plugins/plugin-mongodb-1.0.0.zip";
 
         File dir = new File(path);
         // create the plugin manager
@@ -60,15 +60,23 @@ protected CompoundPluginDescriptorFinder createPluginDescriptorFinder() {
             listener.setServerAddress("127.0.0.1:27017");
         });
 
+        eventListeners.forEach(listener -> {
+            listener.setDBConfig("eventlog|tron|123456");
+        });
+
         eventListeners.forEach(listener -> {
             listener.setTopic(Trigger.BLOCK_TRIGGER, "block");
             listener.setTopic(Trigger.TRANSACTION_TRIGGER, "transaction");
             listener.setTopic(Trigger.CONTRACTEVENT_TRIGGER, "contractevent");
             listener.setTopic(Trigger.CONTRACTLOG_TRIGGER, "contractlog");
         });
 
+        eventListeners.forEach(listener -> {
+            listener.start();
+        });
+
         ObjectMapper objectMapper = new ObjectMapper();
-        for (int index = 0; index < 666; ++index){
+        for (int index = 0; index < 1000; ++index){
             BlockLogTrigger trigger = new BlockLogTrigger();
             trigger.setBlockNumber(index);
             trigger.setBlockHash("000000000002f5834df6036318999576bfa23ff1a57e0538fa87d5a90319659e");

plugins/kafkaplugin/src/main/java/org/tron/eventplugin/KafkaEventListener.java

@@ -19,16 +19,24 @@ public void setServerAddress(String address) {
         }
 
         MessageSenderImpl.getInstance().setServerAddress(address);
-
-        // MessageSenderImpl should never init until server address is set
-        MessageSenderImpl.getInstance().init();
     }
 
     @Override
     public void setTopic(int eventType, String topic) {
         MessageSenderImpl.getInstance().setTopic(eventType, topic);
     }
 
+    @Override
+    public void setDBConfig(String dbConfig) {
+        // empty implementation
+    }
+
+    @Override
+    public void start() {
+        // MessageSenderImpl should never init until server address is set
+        MessageSenderImpl.getInstance().init();
+    }
+
     @Override
     public void handleBlockEvent(Object data) {
 

plugins/mongodbplugin/src/main/java/org/tron/eventplugin/MongodbEventListener.java

@@ -19,16 +19,24 @@ public void setServerAddress(String address) {
         }
 
         MongodbSenderImpl.getInstance().setServerAddress(address);
-
-        // MessageSenderImpl should never init until server address is set
-        MongodbSenderImpl.getInstance().init();
     }
 
     @Override
     public void setTopic(int eventType, String topic) {
         MongodbSenderImpl.getInstance().setTopic(eventType, topic);
     }
 
+    @Override
+    public void setDBConfig(String dbConfig) {
+        MongodbSenderImpl.getInstance().setDBConfig(dbConfig);
+    }
+
+    @Override
+    public void start() {
+        // MessageSenderImpl should never init until server address is set
+        MongodbSenderImpl.getInstance().init();
+    }
+
     @Override
     public void handleBlockEvent(Object data) {
 

plugins/mongodbplugin/src/main/java/org/tron/eventplugin/MongodbSenderImpl.java

@@ -35,6 +35,10 @@ public class MongodbSenderImpl{
     private MongoManager mongoManager;
     private Map<String, MongoTemplate> mongoTemplateMap;
 
+    private String dbName;
+    private String dbUserName;
+    private String dbPassword;
+
     private MongoConfig mongoConfig;
 
     public static MongodbSenderImpl getInstance(){
@@ -49,6 +53,23 @@ public static MongodbSenderImpl getInstance(){
         return instance;
     }
 
+    public void setDBConfig(String dbConfig){
+        if (StringUtils.isNullOrEmpty(dbConfig)){
+            return;
+        }
+
+        String[] params = dbConfig.split("\\|");
+        if (params.length != 3){
+            return;
+        }
+
+        dbName = params[0];
+        dbUserName = params[1];
+        dbPassword = params[2];
+
+        loadMongoConfig();
+    }
+
     public void setServerAddress(final String serverAddress){
         if (StringUtils.isNullOrEmpty(serverAddress)){
             return;
@@ -77,13 +98,6 @@ public void setServerAddress(final String serverAddress){
 
         mongoConfig.setHost(mongoHostName);
         mongoConfig.setPort(mongoPort);
-
-        loadMongoConfig();
-
-        if (Objects.isNull(mongoManager)){
-            mongoManager = new MongoManager();
-            mongoManager.initConfig(mongoConfig);
-        }
     }
 
     public void init(){
@@ -92,16 +106,40 @@ public void init(){
             return;
         }
 
+        if (Objects.isNull(mongoManager)){
+            mongoManager = new MongoManager();
+            mongoManager.initConfig(mongoConfig);
+        }
+
+        mongoTemplateMap = new HashMap<>();
+        createCollections();
+
         triggerProcessThread = new Thread(triggerProcessLoop);
         triggerProcessThread.start();
 
-        mongoTemplateMap = new HashMap<>();
         loaded = true;
     }
 
-    private void loadMongoConfig(){
+    private void createCollections(){
+        mongoManager.createCollection(blockTopic);
+        createMongoTemplate(blockTopic);
+
+        mongoManager.createCollection(transactionTopic);
+        createMongoTemplate(transactionTopic);
 
+        mongoManager.createCollection(contractLogTopic);
+        createMongoTemplate(contractLogTopic);
+
+        mongoManager.createCollection(contractEventTopic);
+        createMongoTemplate(contractEventTopic);
+    }
+
+    private void loadMongoConfig(){
         if (Objects.isNull(mongoConfig)){
+            mongoConfig = new MongoConfig();
+        }
+
+        if (StringUtils.isNullOrEmpty(dbName)){
             return;
         }
 
@@ -114,26 +152,20 @@ private void loadMongoConfig(){
             }
             properties.load(input);
 
-            String dbName = properties.getProperty("mongo.dbname");
-            String userName = properties.getProperty("mongo.username");
-            String password = properties.getProperty("mongo.password");
-
             int connectionsPerHost = Integer.parseInt(properties.getProperty("mongo.connectionsPerHost"));
             int threadsAllowedToBlockForConnectionMultiplie = Integer.parseInt(
                     properties.getProperty("mongo.threadsAllowedToBlockForConnectionMultiplier"));
 
             mongoConfig.setDbName(dbName);
-            mongoConfig.setUsername(userName);
-            mongoConfig.setPassword(password);
-
+            mongoConfig.setUsername(dbUserName);
+            mongoConfig.setPassword(dbPassword);
             mongoConfig.setConnectionsPerHost(connectionsPerHost);
             mongoConfig.setThreadsAllowedToBlockForConnectionMultiplier(threadsAllowedToBlockForConnectionMultiplie);
         } catch (IOException e) {
             e.printStackTrace();
         } catch (Exception e){
             e.printStackTrace();
         }
-
     }
 
     private MongoTemplate createMongoTemplate(final String collectionName){
@@ -177,9 +209,6 @@ else if (triggerType == Constant.CONTRACTLOG_TRIGGER){
         else {
             return;
         }
-
-        mongoManager.createCollection(topic);
-        createMongoTemplate(topic);
     }
 
     public void close() {
@@ -209,6 +238,7 @@ public void handleTransactionTrigger(Object data) {
             return;
         }
 
+        System.out.println(data);
 
         MongoTemplate template = mongoTemplateMap.get(transactionTopic);
         if (Objects.nonNull(template)) {

plugins/mongodbplugin/src/main/java/org/tron/mongodb/MongoManager.java

@@ -30,6 +30,11 @@ public void initConfig(MongoConfig config) {
         String username = config.getUsername();
         String password = config.getPassword();
         String databaseName = config.getDbName();
+
+        if (StringUtils.isNullOrEmpty(databaseName)){
+            return;
+        }
+
         MongoCredential credential = MongoCredential.createScramSha1Credential(username, databaseName,
                 password.toCharArray());
         List<MongoCredential> credentials = new ArrayList<MongoCredential>();

plugins/mongodbplugin/src/main/resources/mongodb.properties

@@ -1,7 +1,3 @@
-mongo.dbname=eventlog
-mongo.username=tron
-mongo.password=123456
-
 mongo.connectionsPerHost=8
 mongo.threadsAllowedToBlockForConnectionMultiplier=4