jackrabbit-oak-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mdue...@apache.org
Subject svn commit: r1540983 - in /jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak: plugins/mongomk/CommitQueue.java plugins/segment/SegmentNodeStore.java spi/commit/ChangeDispatcher.java spi/state/AbstractNodeStoreBranch.java
Date Tue, 12 Nov 2013 09:33:34 GMT
Author: mduerig
Date: Tue Nov 12 09:33:33 2013
New Revision: 1540983

URL: http://svn.apache.org/r1540983
Log:
OAK-1143: [scala] Repository init throws "illegal cyclic reference involving class ChangeDispatcher"
Make ChangeDispatcher implement Observer and replace beforeCommit, localCommit and afterCommit
method with Observer.contentChanged

Modified:
    jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/mongomk/CommitQueue.java
    jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentNodeStore.java
    jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/ChangeDispatcher.java
    jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/state/AbstractNodeStoreBranch.java

Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/mongomk/CommitQueue.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/mongomk/CommitQueue.java?rev=1540983&r1=1540982&r2=1540983&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/mongomk/CommitQueue.java
(original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/mongomk/CommitQueue.java
Tue Nov 12 09:33:33 2013
@@ -109,11 +109,11 @@ class CommitQueue {
                 store.setHeadRevision(c.getRevision());
                 NodeState root = store.getRoot();
                 // TODO: correct?
-                dispatcher.beforeCommit(store.getRoot(before));
+                dispatcher.contentChanged(store.getRoot(before), null);
                 try {
-                    dispatcher.localCommit(root, info);
+                    dispatcher.contentChanged(root, info);
                 } finally {
-                    dispatcher.afterCommit(root);
+                    dispatcher.contentChanged(root, null);
                 }
             } finally {
                 // notify next if there is any

Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentNodeStore.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentNodeStore.java?rev=1540983&r1=1540982&r2=1540983&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentNodeStore.java
(original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentNodeStore.java
Tue Nov 12 09:33:33 2013
@@ -78,16 +78,16 @@ public class SegmentNodeStore implements
     }
 
     boolean setHead(SegmentNodeState base, SegmentNodeState head, CommitInfo info) {
-        changeDispatcher.beforeCommit(base.getChildNode(ROOT));
+        changeDispatcher.contentChanged(base.getChildNode(ROOT), null);
         try {
             if (journal.setHead(base.getRecordId(), head.getRecordId())) {
-                changeDispatcher.localCommit(head.getChildNode(ROOT), info);
+                changeDispatcher.contentChanged(head.getChildNode(ROOT), info);
                 return true;
             } else {
                 return false;
             }
         } finally {
-            changeDispatcher.afterCommit(getRoot());
+            changeDispatcher.contentChanged(getRoot(), null);
         }
     }
 

Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/ChangeDispatcher.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/ChangeDispatcher.java?rev=1540983&r1=1540982&r2=1540983&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/ChangeDispatcher.java
(original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/ChangeDispatcher.java
Tue Nov 12 09:33:33 2013
@@ -19,47 +19,35 @@
 package org.apache.jackrabbit.oak.spi.commit;
 
 import static com.google.common.base.Preconditions.checkNotNull;
-import static com.google.common.base.Preconditions.checkState;
 
 import java.io.Closeable;
-import java.util.concurrent.atomic.AtomicLong;
 
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 
 import org.apache.jackrabbit.oak.spi.state.NodeState;
 
 /**
- * A {@code ChangeDispatcher} instance records changes to a
- * {@link org.apache.jackrabbit.oak.spi.state.NodeStore}
- * and dispatches them to interested parties.
+ * A {@code ChangeDispatcher} instance dispatches content changes
+ * to registered {@link Observer}s.
  * <p>
- * Actual changes are reported by calling {@link #beforeCommit(NodeState)},
- * {@link #localCommit(NodeState, CommitInfo)} and {@link #afterCommit(NodeState)} in that
order:
- * <pre>
-      NodeState root = store.getRoot();
-      branch.rebase();
-      changeDispatcher.beforeCommit(root);
-      try {
-          NodeState head = branch.getHead();
-          branch.merge();
-          changeDispatcher.localCommit(head);
-      } finally {
-          changeDispatcher.afterCommit(store.getRoot());
-      }
- * </pre>
+ * Changes are reported by calling {@link #contentChanged(NodeState, CommitInfo)}.
  * <p>
  * The {@link #addObserver(Observer)} method registers an {@link Observer} for receiving
- * notifications about all changes reported to this instance.
+ * notifications for all changes reported to this instance.
  */
-public class ChangeDispatcher implements Observable {
+public class ChangeDispatcher implements Observable, Observer {
+    // TODO make the queue size configurable
+    private static final int QUEUE_SIZE = 8192;
+
     private final CompositeObserver observers = new CompositeObserver();
 
     @Nonnull
     private NodeState root;
 
     /**
-     * Create a new instance for recording changes to a {@code NodeStore}
-     * @param root  current root node state of the node store
+     * Create a new instance for dispatching content changes
+     * @param root  current root node state
      */
     public ChangeDispatcher(@Nonnull NodeState root) {
         this.root = checkNotNull(root);
@@ -76,8 +64,7 @@ public class ChangeDispatcher implements
     @Override
     @Nonnull
     public Closeable addObserver(Observer observer) {
-        // FIXME don't hard code queue size
-        final BackgroundObserver backgroundObserver = new BackgroundObserver(observer, 8192);
+        final BackgroundObserver backgroundObserver = new BackgroundObserver(observer, QUEUE_SIZE);
         backgroundObserver.contentChanged(root, null);
         observers.addObserver(backgroundObserver);
         return new Closeable() {
@@ -89,72 +76,10 @@ public class ChangeDispatcher implements
         };
     }
 
-    private final AtomicLong changeCount = new AtomicLong(0);
-
-    private boolean inLocalCommit() {
-        return changeCount.get() % 2 == 1;
-    }
-
-    /**
-     * Call with the latest persisted root node state right before persisting further changes.
-     * Calling this method marks this instance to be inside a local commit.
-     * <p>
-     * The differences from the root node state passed to the last call to
-     * {@link #afterCommit(NodeState)} to {@code root} are reported as cluster external
-     * changes to any listener.
-     *
-     * @param root  latest persisted root node state.
-     * @throws IllegalStateException  if inside a local commit
-     */
-    public synchronized void beforeCommit(@Nonnull NodeState root) {
-        checkState(!inLocalCommit());
-        checkNotNull(root);
-        changeCount.incrementAndGet();
-        observers.contentChanged(root, null);
-        this.root = root;
-    }
-
-    /**
-     * Call right after changes have been successfully persisted passing
-     * the new root node state resulting from the persist operation.
-     * <p>
-     * The differences from the root node state passed to the last call to
-     * {@link #beforeCommit(NodeState)} to {@code root} are reported as
-     * cluster local changes to any listener in case non-{@code null}
-     * commit information is provided. If no local commit information is
-     * given, then no local changes are reported and the committed changes
-     * will only show up as an aggregate with any concurrent external changes
-     * reported during the {@link #afterCommit(NodeState)} call.
-     *
-     * @param root root node state just persisted
-     * @param info commit information
-     * @throws IllegalStateException  if not inside a local commit
-     */
-    public synchronized void localCommit(
-            @Nonnull NodeState root, @Nonnull CommitInfo info) {
-        checkState(inLocalCommit());
+    @Override
+    public synchronized void contentChanged(@Nonnull NodeState root, @Nullable CommitInfo
info) {
         checkNotNull(root);
         observers.contentChanged(root, info);
         this.root = root;
     }
-
-    /**
-     * Call to mark the end of a persist operation passing the latest persisted root node
state.
-     * Calling this method marks this instance to not be inside a local commit.
-     * <p>
-     * The difference from the root node state passed to the las call to
-     * {@link #localCommit(NodeState, CommitInfo)} to {@code root} are reported as cluster
external
-     * changes to any listener.
-
-     * @param root  latest persisted root node state.
-     * @throws IllegalStateException  if not inside a local commit
-     */
-    public synchronized void afterCommit(@Nonnull NodeState root) {
-        checkState(inLocalCommit());
-        checkNotNull(root);
-        observers.contentChanged(root, null);
-        this.root = root;
-        changeCount.incrementAndGet();
-    }
-
 }

Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/state/AbstractNodeStoreBranch.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/state/AbstractNodeStoreBranch.java?rev=1540983&r1=1540982&r2=1540983&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/state/AbstractNodeStoreBranch.java
(original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/state/AbstractNodeStoreBranch.java
Tue Nov 12 09:33:33 2013
@@ -389,14 +389,14 @@ public abstract class AbstractNodeStoreB
                 throws CommitFailedException {
             try {
                 rebase();
-                dispatcher.beforeCommit(base);
+                dispatcher.contentChanged(base, null);
                 NodeState toCommit = checkNotNull(hook).processCommit(base, head);
                 NodeState newHead = AbstractNodeStoreBranch.this.persist(toCommit, base,
info);
-                dispatcher.localCommit(newHead, info);
+                dispatcher.contentChanged(newHead, info);
                 branchState = new Merged(base);
                 return newHead;
             } finally {
-                dispatcher.afterCommit(getRoot());
+                dispatcher.contentChanged(getRoot(), null);
             }
         }
     }
@@ -469,7 +469,7 @@ public abstract class AbstractNodeStoreB
         NodeState merge(@Nonnull CommitHook hook, CommitInfo info) throws CommitFailedException
{
             try {
                 rebase();
-                dispatcher.beforeCommit(base);
+                dispatcher.contentChanged(base, null);
                 NodeState toCommit = checkNotNull(hook).processCommit(base, head);
                 if (toCommit.equals(base)) {
                     branchState = new Merged(base);
@@ -477,12 +477,12 @@ public abstract class AbstractNodeStoreB
                 } else {
                     head = AbstractNodeStoreBranch.this.persist(toCommit, head, info);
                     NodeState newRoot = AbstractNodeStoreBranch.this.merge(head, info);
-                    dispatcher.localCommit(newRoot, info);
+                    dispatcher.contentChanged(newRoot, info);
                     branchState = new Merged(base);
                     return newRoot;
                 }
             } finally {
-                dispatcher.afterCommit(getRoot());
+                dispatcher.contentChanged(getRoot(), null);
             }
         }
 



Mime
View raw message