svn commit: r1377710 - /jackrabbit/oak/trunk/oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java

[md...@apache.org]

Author: mduerig
Date: Mon Aug 27 15:19:32 2012
New Revision: 1377710

URL: http://svn.apache.org/viewvc?rev=1377710&view=rev
Log:
OAK-279: ChangeProcessor getting stuck while shutdown

Modified:
    jackrabbit/oak/trunk/oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java

Modified: jackrabbit/oak/trunk/oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java?rev=1377710&r1=1377709&r2=1377710&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java (original)
+++ jackrabbit/oak/trunk/oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java Mon Aug 27 15:19:32 2012
@@ -19,7 +19,6 @@ package org.apache.jackrabbit.oak.jcr.ob
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -46,7 +45,6 @@ class ChangeProcessor implements Runnabl
     private final ChangeExtractor changeExtractor;
     private final EventListener listener;
     private final AtomicReference<ChangeFilter> filterRef;
-    private final CountDownLatch stopped = new CountDownLatch(1);
     private volatile boolean running;
     private volatile boolean stopping;
     private ScheduledFuture<?> future;
@@ -66,6 +64,7 @@ class ChangeProcessor implements Runnabl
     /**
      * Stop this change processor if running. After returning from this methods no further
      * events will be delivered.
+     * @throws IllegalStateException if not yet started or stopped already
      */
     public synchronized void stop() {
         if (future == null) {
@@ -75,11 +74,10 @@ class ChangeProcessor implements Runnabl
         try {
             stopping = true;
             future.cancel(true);
-            if (running) {
-                stopped.await();
+            while (running) {
+                wait();
             }
-        }
-        catch (InterruptedException e) {
+        } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
         }
         finally {
@@ -96,6 +94,7 @@ class ChangeProcessor implements Runnabl
         if (future != null) {
             throw new IllegalStateException("Change processor started already");
         }
+        stopping = false;
         future = executor.scheduleWithFixedDelay(this, 100, 1000, TimeUnit.MILLISECONDS);
     }
 
@@ -108,11 +107,11 @@ class ChangeProcessor implements Runnabl
             if (!stopping) {
                 diff.sendEvents();
             }
-        } finally{
-            if (stopping) {
-                stopped.countDown();
+        } finally {
+            synchronized (this) {
+                running = false;
+                notifyAll();
             }
-            running = false;
         }
     }