process events concurrently

docs-core/src/main/java/com/sismics/docs/core/listener/async/AclCreatedAsyncListener.java

@@ -1,5 +1,6 @@
 package com.sismics.docs.core.listener.async;
 
+import com.google.common.eventbus.AllowConcurrentEvents;
 import com.google.common.eventbus.Subscribe;
 import com.sismics.docs.core.event.AclCreatedAsyncEvent;
 import com.sismics.docs.core.model.context.AppContext;
@@ -24,6 +25,7 @@ public class AclCreatedAsyncListener {
      * @param event ACL created event
      */
     @Subscribe
+    @AllowConcurrentEvents
     public void on(final AclCreatedAsyncEvent event) {
         if (log.isInfoEnabled()) {
             log.info("ACL created event: " + event.toString());

docs-core/src/main/java/com/sismics/docs/core/listener/async/AclDeletedAsyncListener.java

@@ -1,5 +1,6 @@
 package com.sismics.docs.core.listener.async;
 
+import com.google.common.eventbus.AllowConcurrentEvents;
 import com.google.common.eventbus.Subscribe;
 import com.sismics.docs.core.event.AclDeletedAsyncEvent;
 import com.sismics.docs.core.model.context.AppContext;
@@ -24,6 +25,7 @@ public class AclDeletedAsyncListener {
      * @param event ACL deleted event
      */
     @Subscribe
+    @AllowConcurrentEvents
     public void on(final AclDeletedAsyncEvent event) {
         if (log.isInfoEnabled()) {
             log.info("ACL deleted event: " + event.toString());

docs-core/src/main/java/com/sismics/docs/core/listener/async/DocumentCreatedAsyncListener.java

@@ -1,5 +1,6 @@
 package com.sismics.docs.core.listener.async;
 
+import com.google.common.eventbus.AllowConcurrentEvents;
 import com.google.common.eventbus.Subscribe;
 import com.sismics.docs.core.dao.ContributorDao;
 import com.sismics.docs.core.event.DocumentCreatedAsyncEvent;
@@ -26,6 +27,7 @@ public class DocumentCreatedAsyncListener {
      * @param event Document created event
      */
     @Subscribe
+    @AllowConcurrentEvents
     public void on(final DocumentCreatedAsyncEvent event) {
         if (log.isInfoEnabled()) {
             log.info("Document created event: " + event.toString());

docs-core/src/main/java/com/sismics/docs/core/listener/async/DocumentDeletedAsyncListener.java

@@ -1,5 +1,6 @@
 package com.sismics.docs.core.listener.async;
 
+import com.google.common.eventbus.AllowConcurrentEvents;
 import com.google.common.eventbus.Subscribe;
 import com.sismics.docs.core.event.DocumentDeletedAsyncEvent;
 import com.sismics.docs.core.model.context.AppContext;
@@ -22,9 +23,9 @@ public class DocumentDeletedAsyncListener {
      * Document deleted.
      * 
      * @param event Document deleted event
-     * @throws Exception e
      */
     @Subscribe
+    @AllowConcurrentEvents
     public void on(final DocumentDeletedAsyncEvent event) {
         if (log.isInfoEnabled()) {
             log.info("Document deleted event: " + event.toString());

docs-core/src/main/java/com/sismics/docs/core/listener/async/DocumentUpdatedAsyncListener.java

@@ -1,5 +1,6 @@
 package com.sismics.docs.core.listener.async;
 
+import com.google.common.eventbus.AllowConcurrentEvents;
 import com.google.common.eventbus.Subscribe;
 import com.sismics.docs.core.dao.ContributorDao;
 import com.sismics.docs.core.dao.DocumentDao;
@@ -30,6 +31,7 @@ public class DocumentUpdatedAsyncListener {
      * @param event Document updated event
      */
     @Subscribe
+    @AllowConcurrentEvents
     public void on(final DocumentUpdatedAsyncEvent event) {
         if (log.isInfoEnabled()) {
             log.info("Document updated event: " + event.toString());

docs-core/src/main/java/com/sismics/docs/core/listener/async/FileDeletedAsyncListener.java

@@ -1,5 +1,6 @@
 package com.sismics.docs.core.listener.async;
 
+import com.google.common.eventbus.AllowConcurrentEvents;
 import com.google.common.eventbus.Subscribe;
 import com.sismics.docs.core.event.FileDeletedAsyncEvent;
 import com.sismics.docs.core.model.context.AppContext;
@@ -27,6 +28,7 @@ public class FileDeletedAsyncListener {
      * @throws Exception e
      */
     @Subscribe
+    @AllowConcurrentEvents
     public void on(final FileDeletedAsyncEvent event) throws Exception {
         if (log.isInfoEnabled()) {
             log.info("File deleted event: " + event.toString());

docs-core/src/main/java/com/sismics/docs/core/listener/async/FileProcessingAsyncListener.java

@@ -1,5 +1,6 @@
 package com.sismics.docs.core.listener.async;
 
+import com.google.common.eventbus.AllowConcurrentEvents;
 import com.google.common.eventbus.Subscribe;
 import com.sismics.docs.core.dao.FileDao;
 import com.sismics.docs.core.dao.UserDao;
@@ -45,6 +46,7 @@ public class FileProcessingAsyncListener {
      * @param event File created event
      */
     @Subscribe
+    @AllowConcurrentEvents
     public void on(final FileCreatedAsyncEvent event) {
         if (log.isInfoEnabled()) {
             log.info("File created event: " + event.toString());

docs-core/src/main/java/com/sismics/docs/core/listener/async/PasswordLostAsyncListener.java

@@ -1,5 +1,6 @@
 package com.sismics.docs.core.listener.async;
 
+import com.google.common.eventbus.AllowConcurrentEvents;
 import com.google.common.eventbus.Subscribe;
 import com.sismics.docs.core.constant.Constants;
 import com.sismics.docs.core.dao.dto.UserDto;
@@ -27,17 +28,18 @@ public class PasswordLostAsyncListener {
     /**
      * Handle events.
      * 
-     * @param passwordLostEvent Event
+     * @param event Event
      */
     @Subscribe
-    public void onPasswordLost(final PasswordLostEvent passwordLostEvent) {
+    @AllowConcurrentEvents
+    public void on(final PasswordLostEvent event) {
         if (log.isInfoEnabled()) {
-            log.info("Password lost event: " + passwordLostEvent.toString());
+            log.info("Password lost event: " + event.toString());
         }
         
         TransactionUtil.handle(() -> {
-            final UserDto user = passwordLostEvent.getUser();
+            final UserDto user = event.getUser();
-            final PasswordRecovery passwordRecovery = passwordLostEvent.getPasswordRecovery();
+            final PasswordRecovery passwordRecovery = event.getPasswordRecovery();
 
             // Send the password recovery email
             Map<String, Object> paramRootMap = new HashMap<>();

docs-core/src/main/java/com/sismics/docs/core/listener/async/RebuildIndexAsyncListener.java

@@ -1,5 +1,6 @@
 package com.sismics.docs.core.listener.async;
 
+import com.google.common.eventbus.AllowConcurrentEvents;
 import com.google.common.eventbus.Subscribe;
 import com.sismics.docs.core.dao.DocumentDao;
 import com.sismics.docs.core.dao.FileDao;
@@ -27,12 +28,13 @@ public class RebuildIndexAsyncListener {
     /**
      * Rebuild Lucene index.
      * 
-     * @param rebuildIndexAsyncEvent Index rebuild event
+     * @param event Index rebuild event
      */
     @Subscribe
-    public void on(final RebuildIndexAsyncEvent rebuildIndexAsyncEvent) {
+    @AllowConcurrentEvents
+    public void on(final RebuildIndexAsyncEvent event) {
         if (log.isInfoEnabled()) {
-            log.info("Rebuild index event: " + rebuildIndexAsyncEvent.toString());
+            log.info("Rebuild index event: " + event.toString());
         }
         
         // Fetch all documents and files

docs-core/src/main/java/com/sismics/docs/core/listener/async/RouteStepValidateAsyncListener.java

@@ -1,5 +1,6 @@
 package com.sismics.docs.core.listener.async;
 
+import com.google.common.eventbus.AllowConcurrentEvents;
 import com.google.common.eventbus.Subscribe;
 import com.sismics.docs.core.constant.Constants;
 import com.sismics.docs.core.dao.dto.UserDto;
@@ -26,22 +27,23 @@ public class RouteStepValidateAsyncListener {
     /**
      * Handle events.
      * 
-     * @param routeStepValidateEvent Event
+     * @param event Event
      */
     @Subscribe
-    public void onRouteStepValidate(final RouteStepValidateEvent routeStepValidateEvent) {
+    @AllowConcurrentEvents
+    public void on(final RouteStepValidateEvent event) {
         if (log.isInfoEnabled()) {
-            log.info("Route step validate event: " + routeStepValidateEvent.toString());
+            log.info("Route step validate event: " + event.toString());
         }
         
         TransactionUtil.handle(() -> {
-            final UserDto user = routeStepValidateEvent.getUser();
+            final UserDto user = event.getUser();
 
             // Send route step validated email
             Map<String, Object> paramRootMap = new HashMap<>();
             paramRootMap.put("user_name", user.getUsername());
-            paramRootMap.put("document_id", routeStepValidateEvent.getDocument().getId());
+            paramRootMap.put("document_id", event.getDocument().getId());
-            paramRootMap.put("document_title", routeStepValidateEvent.getDocument().getTitle());
+            paramRootMap.put("document_title", event.getDocument().getTitle());
 
             EmailUtil.sendEmail(Constants.EMAIL_TEMPLATE_ROUTE_STEP_VALIDATE, user, paramRootMap);
         });

docs-core/src/main/java/com/sismics/docs/core/model/context/AppContext.java

@@ -174,7 +174,7 @@ public class AppContext {
             return new EventBus();
         } else {
             ThreadPoolExecutor executor = new ThreadPoolExecutor(8, 8,
-                    0L, TimeUnit.MILLISECONDS,
+                    1L, TimeUnit.MINUTES,
                     new LinkedBlockingQueue<>());
             asyncExecutorList.add(executor);
             return new AsyncEventBus(executor);
@@ -206,7 +206,7 @@ public class AppContext {
             // Shutdown executor, don't accept any more tasks (can cause error with nested events)
             try {
                 executor.shutdown();
-                executor.awaitTermination(60, TimeUnit.SECONDS);
+                executor.awaitTermination(1, TimeUnit.MINUTES);
             } catch (InterruptedException e) {
                 // NOP
             }