svn commit: r1468477 - /jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation2/ObservationManagerImpl2.java

[md...@apache.org]

Author: mduerig
Date: Tue Apr 16 15:52:05 2013
New Revision: 1468477

URL: http://svn.apache.org/r1468477
Log:
OAK-775: Implement backward compatible observation
clear event queue after restart

Modified:
    jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation2/ObservationManagerImpl2.java

Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation2/ObservationManagerImpl2.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation2/ObservationManagerImpl2.java?rev=1468477&r1=1468476&r2=1468477&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation2/ObservationManagerImpl2.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation2/ObservationManagerImpl2.java Tue Apr 16 15:52:05 2013
@@ -40,6 +40,7 @@ import org.apache.jackrabbit.commons.ite
 import org.apache.jackrabbit.oak.api.CommitFailedException;
 import org.apache.jackrabbit.oak.api.ContentSession;
 import org.apache.jackrabbit.oak.api.Root;
+import org.apache.jackrabbit.oak.api.Tree;
 import org.apache.jackrabbit.oak.namepath.NamePathMapper;
 import org.apache.jackrabbit.oak.plugins.nodetype.ReadOnlyNodeTypeManager;
 import org.slf4j.Logger;
@@ -51,7 +52,7 @@ import org.slf4j.LoggerFactory;
 public class ObservationManagerImpl2 implements ObservationManager {
     private static final Logger log = LoggerFactory.getLogger(ObservationManagerImpl2.class);
 
-    private final Root root;
+    private final ContentSession contentSession;
     private final NamePathMapper namePathMapper;
     private final ScheduledExecutorService executor;
     private final ReadOnlyNodeTypeManager ntMgr;
@@ -59,10 +60,29 @@ public class ObservationManagerImpl2 imp
     private final AtomicBoolean hasEvents = new AtomicBoolean(false);
 
     public ObservationManagerImpl2(Root root, NamePathMapper namePathMapper, ScheduledExecutorService executor) {
-        this.root = checkNotNull(root);
+        this.contentSession = checkNotNull(root).getContentSession();
         this.namePathMapper = checkNotNull(namePathMapper);
         this.executor = checkNotNull(executor);
         this.ntMgr = ReadOnlyNodeTypeManager.getInstance(root, namePathMapper);
+        clearEventQueueOnRestart();
+    }
+
+    // FIXME: we need a better way to communicate BUNDLE_ID across.
+    // Preferably through persisting it to th repository
+    private void clearEventQueueOnRestart() {
+        if (EventQueueWriterProvider.BUNDLE_ID.get() == 0) {
+            try {
+                Root root = contentSession.getLatestRoot();
+                Tree events = root.getTree(ObservationConstants.EVENTS_PATH);
+                if (events != null) {
+                    events.remove();
+                    root.commit();
+                }
+            }
+            catch (CommitFailedException e) {
+                log.warn("Error clearing event queue after restart", e);
+            }
+        }
     }
 
     private static void stop(EventCollector collector) {
@@ -181,7 +201,7 @@ public class ObservationManagerImpl2 imp
     }
 
     ContentSession getContentSession() {
-        return root.getContentSession();
+        return contentSession;
     }
 
     NamePathMapper getNamePathMapper() {