[fix] User event reload when blockchain reload

[fix] better log on node start

duniter4j-es-core/src/main/java/org/duniter/elasticsearch/PluginInit.java

@@ -85,7 +85,7 @@ public class PluginInit extends AbstractLifecycleComponent<PluginInit> {
         // Reload All indices
         if (pluginSettings.reloadAllIndices() || pluginSettings.reloadBlockchainIndices()) {
             if (logger.isWarnEnabled()) {
-                logger.warn("Reloading [core-plugin] indices...");
+                logger.warn("Reloading indices...");
             }
 
             injector.getInstance(CurrencyService.class)
@@ -93,21 +93,21 @@ public class PluginInit extends AbstractLifecycleComponent<PluginInit> {
                     .createIndexIfNotExists();
 
             if (logger.isInfoEnabled()) {
-                logger.info("Reloading [core-plugin] indices. [OK]");
+                logger.info("Reloading indices [OK]");
             }
         }
 
         else {
 
             if (logger.isInfoEnabled()) {
-                logger.info("Checking if [core-plugin] indices exists...");
+                logger.info("Checking indices...");
             }
 
             injector.getInstance(CurrencyService.class)
                     .createIndexIfNotExists();
 
             if (logger.isInfoEnabled()) {
-                logger.info("Checking if [core-plugin] indices exists. [OK]");
+                logger.info("Checking indices [OK]");
             }
         }
     }
@@ -120,9 +120,13 @@ public class PluginInit extends AbstractLifecycleComponent<PluginInit> {
             Peer peer = pluginSettings.checkAndGetPeer();
 
             // Index (or refresh) node's currency
-            Currency currency = injector.getInstance(CurrencyService.class)
+            final Currency currency = injector.getInstance(CurrencyService.class)
                     .indexCurrencyFromPeer(peer, true);
 
+            if (logger.isInfoEnabled()) {
+                logger.info(String.format("[%s] Indexing blockchain...", currency.getCurrencyName()));
+            }
+
             injector.getInstance(RestSecurityController.class)
 
                     // Add access to <currency>/block index
@@ -161,6 +165,10 @@ public class PluginInit extends AbstractLifecycleComponent<PluginInit> {
                 injector.getInstance(PeerService.class)
                         .listenAndIndexPeers(peer);
 
+                if (logger.isInfoEnabled()) {
+                    logger.info(String.format("[%s] Indexing blockchain [OK]", currency.getCurrencyName()));
+                }
+
             }, ClusterHealthStatus.YELLOW, ClusterHealthStatus.GREEN);
 
 

duniter4j-es-core/src/main/java/org/duniter/elasticsearch/PluginSettings.java

@@ -197,7 +197,7 @@ public class PluginSettings extends AbstractLifecycleComponent<PluginSettings> {
     }
 
     public boolean reloadBlockchainIndices()  {
-        return settings.getAsBoolean("duniter.blockchain.indices.reload", false);
+        return settings.getAsBoolean("duniter.blockchain.reload", false);
     }
 
     public File getTempDirectory() {

duniter4j-es-core/src/main/java/org/duniter/elasticsearch/service/AbstractBlockchainListenerService.java

@@ -96,10 +96,21 @@ public abstract class AbstractBlockchainListenerService extends AbstractService
         if("current".equals(change.getId())) return;
 
         switch (change.getOperation()) {
+            // on INDEX
+            case CREATE:
+                if (change.getSource() != null) {
+                    synchronized (threadLock) {
+                        processBlockIndex(change);
+                    }
+                }
+                break;
+
             // on INDEX
             case INDEX:
-                synchronized (threadLock) {
-                    processBlockIndex(change);
+                if (change.getSource() != null) {
+                    synchronized (threadLock) {
+                        processBlockIndex(change);
+                    }
                 }
                 break;
 

duniter4j-es-subscription/src/main/java/org/duniter/elasticsearch/subscription/PluginInit.java

@@ -79,24 +79,24 @@ public class PluginInit extends AbstractLifecycleComponent<PluginInit> {
 
         if (reloadIndices) {
             if (logger.isInfoEnabled()) {
-                logger.info("Reloading all subscription indices...");
+                logger.info("Reloading indices...");
             }
             injector.getInstance(SubscriptionIndexDao.class)
                     .deleteIndex()
                     .createIndexIfNotExists();
 
             if (logger.isInfoEnabled()) {
-                logger.info("Reloading all subscription indices... [OK]");
+                logger.info("Reloading indices [OK]");
             }
         }
         else {
             if (logger.isInfoEnabled()) {
-                logger.info("Checking subscription indices...");
+                logger.info("Checking indices...");
             }
             injector.getInstance(SubscriptionIndexDao.class).createIndexIfNotExists();
 
             if (logger.isInfoEnabled()) {
-                logger.info("Checking subscription indices... [OK]");
+                logger.info("Checking indices [OK]");
             }
         }
     }

duniter4j-es-user/src/main/java/org/duniter/elasticsearch/user/PluginInit.java

@@ -83,7 +83,7 @@ public class PluginInit extends AbstractLifecycleComponent<PluginInit> {
         // Reload all indices
         if (pluginSettings.reloadAllIndices()) {
             if (logger.isInfoEnabled()) {
-                logger.info("Reloading [user-plugin] indices...");
+                logger.info("Reloading indices...");
             }
             injector.getInstance(HistoryService.class)
                     .deleteIndex()
@@ -102,14 +102,17 @@ public class PluginInit extends AbstractLifecycleComponent<PluginInit> {
                     .createIndexIfNotExists();
 
             if (logger.isInfoEnabled()) {
-                logger.info("Reloading [user-plugin] indices. [OK]");
+                logger.info("Reloading indices [OK]");
             }
         }
 
         else {
             if (logger.isInfoEnabled()) {
-                logger.info("Checking [user-plugin] indices...");
+                logger.info("Checking indices...");
             }
+
+            boolean cleanBlockchainUserEvents = injector.getInstance(UserService.class).isIndexExists() && pluginSettings.reloadBlockchainIndices();
+
             injector.getInstance(HistoryService.class).createIndexIfNotExists();
             injector.getInstance(UserService.class).createIndexIfNotExists();
             injector.getInstance(MessageService.class).createIndexIfNotExists();
@@ -117,19 +120,19 @@ public class PluginInit extends AbstractLifecycleComponent<PluginInit> {
             injector.getInstance(UserInvitationService.class).createIndexIfNotExists();
 
             if (logger.isInfoEnabled()) {
-                logger.info("Checking [user-plugin] indices. [OK]");
+                logger.info("Checking indices [OK]");
             }
 
-            // Reload blockchain indices : user/event
-            if (pluginSettings.reloadBlockchainIndices()) {
+            // Clean user events on blockchain
+            if (cleanBlockchainUserEvents) {
                 if (logger.isInfoEnabled()) {
-                    logger.info("Deleting existing user event, referencing a block...");
+                    logger.info("Deleting user events on blockchain (blockchain will be reload)...");
                 }
                 // Delete events that reference a block
                 injector.getInstance(UserEventService.class)
                         .deleteEventsByReference(new UserEvent.Reference(null/*all*/, BlockchainService.BLOCK_TYPE, null/*all*/));
                 if (logger.isInfoEnabled()) {
-                    logger.info("Deleting existing user event, referencing a block. [OK]");
+                    logger.info("Deleting user events on blockchain [OK]");
                 }
             }
         }

duniter4j-es-user/src/main/java/org/duniter/elasticsearch/user/service/BlockchainUserEventService.java

@@ -68,7 +68,6 @@ public class BlockchainUserEventService extends AbstractBlockchainListenerServic
     private final UserService userService;
     private final UserEventService userEventService;
     private final AdminService adminService;
-    private Queue<UserEvent.Reference> referencesToDelete = ConcurrentCollections.newBlockingQueue();
 
     @Inject
     public BlockchainUserEventService(Duniter4jClient client, PluginSettings settings, CryptoService cryptoService,
@@ -147,20 +146,13 @@ public class BlockchainUserEventService extends AbstractBlockchainListenerServic
             reference.setHash(block.getHash());
         }
 
-        // Add to queue
-        referencesToDelete.add(reference);
-
+        this.bulkRequest = userEventService.addDeleteEventsByReferenceToBulk(reference, this.bulkRequest, this.bulkSize, false);
         flushBulkRequestOrSchedule();
     }
 
 
     protected void beforeFlush() {
 
-        UserEvent.Reference reference = referencesToDelete.poll();
-        while (reference != null) {
-            this.bulkRequest = userEventService.addDeleteEventsByReferenceToBulk(reference, this.bulkRequest, this.bulkSize, false);
-            reference = referencesToDelete.poll();
-        }
     }
 
     /* -- internal method -- */

duniter4j-es-user/src/main/java/org/duniter/elasticsearch/user/service/UserEventService.java

@@ -187,15 +187,13 @@ public class UserEventService extends AbstractService implements ChangeService.C
     }
 
 
-    public ActionFuture<?> deleteEventsByReference(final UserEvent.Reference reference) {
+    public void deleteEventsByReference(final UserEvent.Reference reference) {
         Preconditions.checkNotNull(reference);
 
         final int bulkSize = pluginSettings.getIndexBulkSize();
 
-        return threadPool.schedule(() -> {
-            BulkRequestBuilder bulkRequest = client.prepareBulk();
-            addDeleteEventsByReferenceToBulk(reference, bulkRequest, bulkSize, true);
-        });
+       BulkRequestBuilder bulkRequest = client.prepareBulk();
+        addDeleteEventsByReferenceToBulk(reference, bulkRequest, bulkSize, true);
     }
 
     public ActionFuture<UpdateResponse> markEventAsRead(String id, String signature) {

duniter4j-es-user/src/main/java/org/duniter/elasticsearch/user/service/UserService.java

@@ -81,6 +81,13 @@ public class UserService extends AbstractService {
         return this;
     }
 
+    /**
+     * Create index need for blockchain mail, if need
+     */
+    public boolean isIndexExists() {
+        return client.existsIndex(INDEX);
+    }
+
     /**
      * Create index for mail
      * @throws JsonProcessingException