SONAR-5237 - Moved Latching into Queue. Defaults to SYNCH

2 files changed, 25 insertions, 8 deletions

diff --git a/sonar-server/src/main/java/org/sonar/server/cluster/LocalNonBlockingWorkQueue.java b/sonar-server/src/main/java/org/sonar/server/cluster/LocalNonBlockingWorkQueue.java
index 8afcdb10c64..c11204912c7 100644
--- a/sonar-server/src/main/java/org/sonar/server/cluster/LocalNonBlockingWorkQueue.java
+++ b/sonar-server/src/main/java/org/sonar/server/cluster/LocalNonBlockingWorkQueue.java
@@ -19,33 +19,50 @@
  */
 package org.sonar.server.cluster;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.sonar.api.ServerComponent;
 import org.sonar.core.cluster.QueueAction;
 import org.sonar.core.cluster.WorkQueue;
 
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 public class LocalNonBlockingWorkQueue extends LinkedBlockingQueue<Runnable>
-  implements ServerComponent, WorkQueue{
+  implements ServerComponent, WorkQueue {
+
+  private static final Logger LOG = LoggerFactory.getLogger(LocalNonBlockingWorkQueue.class);
 
-  public LocalNonBlockingWorkQueue(){
+  public LocalNonBlockingWorkQueue() {
     super();
   }
 
   @Override
   public void enqueue(QueueAction action) {
+    CountDownLatch latch = new CountDownLatch(1);
+    action.setLatch(latch);
     try {
-      this.offer(action, 1000,TimeUnit.SECONDS);
+      this.offer(action, 1000, TimeUnit.SECONDS);
+      latch.await();
     } catch (InterruptedException e) {
-      //TODO throw a runtime error here.
+      LOG.error("ES update has been interrupted: {}", e.getMessage());
     }
   }
 
   @Override
-  public void enqueue(Iterable<QueueAction> actions) {
-    for (QueueAction action : actions) {
-      enqueue(action);
+  public void enqueue(Collection<QueueAction> actions) {
+    CountDownLatch latch = new CountDownLatch(actions.size());
+    try {
+      for (QueueAction action : actions) {
+        action.setLatch(latch);
+        this.offer(action, 1000, TimeUnit.SECONDS);
+      }
+      latch.await();
+    } catch (InterruptedException e) {
+      LOG.error("ES update has been interrupted: {}", e.getMessage());
     }
   }
 }
diff --git a/sonar-server/src/main/java/org/sonar/server/platform/ServerComponents.java b/sonar-server/src/main/java/org/sonar/server/platform/ServerComponents.java
index 4b3521cac18..28bda726b6b 100644
--- a/sonar-server/src/main/java/org/sonar/server/platform/ServerComponents.java
+++ b/sonar-server/src/main/java/org/sonar/server/platform/ServerComponents.java
@@ -176,7 +176,7 @@ class ServerComponents {
       System2.INSTANCE,
 
       /* new RuleDao working with ES */
-      RuleDao.class
+      org.sonar.server.rule2.RuleDao.class
     ));
     components.addAll(CorePropertyDefinitions.all());
     components.addAll(DatabaseMigrations.CLASSES);