You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by ju...@apache.org on 2013/11/10 22:00:36 UTC
svn commit: r1540548 - in
/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit:
BackgroundObserver.java CompositeObserver.java DiffObserver.java
Author: jukka
Date: Sun Nov 10 21:00:36 2013
New Revision: 1540548
URL: http://svn.apache.org/r1540548
Log:
OAK-1143: [scala] Repository init throws "illegal cyclic reference involving class ChangeDispatcher"
Add a few general-purpose Observer classes
Added:
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserver.java
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/CompositeObserver.java
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/DiffObserver.java
Added: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserver.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserver.java?rev=1540548&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserver.java (added)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserver.java Sun Nov 10 21:00:36 2013
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.oak.spi.commit;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.collect.Queues.newArrayBlockingQueue;
+
+import java.lang.Thread.UncaughtExceptionHandler;
+import java.util.concurrent.BlockingQueue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An observer that uses a change queue and a background thread to forward
+ * content changes to another observer. The mechanism is designed so that
+ * the {@link #contentChanged(NodeState, CommitInfo)} method will never block,
+ * regardless of the behavior of the other observer. If that observer blocks
+ * or is too slow to consume all content changes, causing the change queue
+ * to fill up, any further update will automatically be merged into just one
+ * external content change, causing potential loss of local commit information.
+ * To help prevent such cases, any sequential external content changes that
+ * the background observer thread has yet to process are automatically merged
+ * to just one change.
+ */
+public class BackgroundObserver implements Observer {
+
+ private static class ContentChange {
+ private final NodeState root;
+ private final CommitInfo info;
+ ContentChange(NodeState root, CommitInfo info) {
+ this.root = root;
+ this.info = info;
+ }
+ }
+
+ /**
+ * Signal for the background thread to stop processing changes.
+ */
+ private static final ContentChange STOP = new ContentChange(null, null);
+
+ private static Logger getLogger(@Nonnull Observer observer) {
+ return LoggerFactory.getLogger(checkNotNull(observer).getClass());
+ }
+
+ /**
+ * The queue of content changes to be processed.
+ */
+ private final BlockingQueue<ContentChange> queue;
+
+ /**
+ * The content change that was last added to the queue.
+ * Used to compact external changes.
+ */
+ private ContentChange last = null;
+
+ /**
+ * Flag to indicate that some content changes were dropped because
+ * the queue was full.
+ */
+ private boolean full = false;
+
+ /**
+ * The background thread used to process changes.
+ */
+ private final Thread thread;
+
+ public BackgroundObserver(
+ @Nonnull final Observer observer, int queueLength,
+ @Nonnull UncaughtExceptionHandler exceptionHandler) {
+ checkNotNull(observer);
+ checkArgument(queueLength > 0);
+ checkNotNull(exceptionHandler);
+
+ this.queue = newArrayBlockingQueue(queueLength);
+
+ this.thread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ while (true) {
+ ContentChange change = queue.take();
+ if (change == STOP) {
+ return;
+ }
+ observer.contentChanged(change.root, change.info);
+ }
+ } catch (InterruptedException e) {
+ getLogger(observer).warn(
+ "Event processing interrupted for " + observer, e);
+ }
+ }
+ });
+ thread.setName(observer.toString());
+ thread.setPriority(Thread.MIN_PRIORITY);
+ thread.setDaemon(true);
+ thread.setUncaughtExceptionHandler(exceptionHandler);
+ thread.start();
+ }
+
+ public BackgroundObserver(
+ @Nonnull final Observer observer, int queueLength) {
+ this(observer, queueLength, new UncaughtExceptionHandler() {
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ getLogger(observer).error(
+ "Uncaught exception in " + observer, e);
+ }
+ });
+ }
+
+ public BackgroundObserver(@Nonnull Observer observer) {
+ this(observer, 1000);
+ }
+
+ /**
+ * Clears the change queue and signals the background thread to stop
+ * without making any further {@link #contentChanged(NodeState, CommitInfo)}
+ * calls to the background observer. If the thread is currently in the
+ * middle of such a call, then that call is allowed to complete; i.e.
+ * the thread is not forcibly interrupted. This method returns immediately
+ * without blocking to wait for the thread to finish.
+ */
+ public synchronized void stop() {
+ queue.clear();
+ queue.add(STOP);
+ // no need to join the thread; it will stop when encountering the STOP
+ }
+
+ //----------------------------------------------------------< Observer >--
+
+ @Override
+ public synchronized void contentChanged(
+ @Nonnull NodeState root, @Nullable CommitInfo info) {
+ checkNotNull(root);
+
+ if (info == null && last != null && last.info == null) {
+ // This is an external change. If the previous change was
+ // also external, we can drop it from the queue (since external
+ // changes in any case can cover multiple commits) to help
+ // prevent the queue from filling up too fast.
+ queue.remove(last);
+ full = false;
+ }
+
+ ContentChange change;
+ if (full) {
+ // If the queue is full, some commits have already been skipped
+ // so we need to drop the possible local commit information as
+ // only external changes can be merged together to larger chunks.
+ change = new ContentChange(root, null);
+ } else {
+ change = new ContentChange(root, info);
+ }
+
+ // Try to add this change to the queue without blocking, and
+ // mark the queue as full if there wasn't enough space
+ full = !queue.offer(change);
+
+ if (!full) {
+ // Keep track of the last change added, so we can do the
+ // compacting of external changes shown above.
+ last = change;
+ }
+ }
+
+ //------------------------------------------------------------< Object >--
+
+ @Override
+ public String toString() {
+ return thread.getName() + " &";
+ }
+
+}
Added: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/CompositeObserver.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/CompositeObserver.java?rev=1540548&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/CompositeObserver.java (added)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/CompositeObserver.java Sun Nov 10 21:00:36 2013
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.oak.spi.commit;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.Sets.newIdentityHashSet;
+
+import java.util.Set;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+
+/**
+ * Composite observer that delegates all content changes to the set of
+ * currently registered component observers.
+ */
+public class CompositeObserver implements Observer {
+
+ private final Set<Observer> observers = newIdentityHashSet();
+
+ public synchronized void addObserver(@Nonnull Observer observer) {
+ checkState(observers.add(checkNotNull(observer)));
+ }
+
+ public synchronized void removeObserver(@Nonnull Observer observer) {
+ checkState(observers.remove(checkNotNull(observer)));
+ }
+
+ //----------------------------------------------------------< Observer >--
+
+ @Override
+ public synchronized void contentChanged(
+ @Nonnull NodeState root, @Nullable CommitInfo info) {
+ checkNotNull(root);
+ for (Observer observer : observers) {
+ observer.contentChanged(root, info);
+ }
+ }
+
+ //------------------------------------------------------------< Object >--
+
+ @Override
+ public synchronized String toString() {
+ return observers.toString();
+ }
+
+}
Added: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/DiffObserver.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/DiffObserver.java?rev=1540548&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/DiffObserver.java (added)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/DiffObserver.java Sun Nov 10 21:00:36 2013
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.oak.spi.commit;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.jackrabbit.oak.spi.state.NodeStateDiff;
+
+/**
+ * Abstract base class for observers that use a content diff to determine
+ * what changed between two consecutive observed states of the repository.
+ * Subclasses just need to provide the diff handler by implementing the
+ * {@link #getRootDiff(NodeState, NodeState, CommitInfo)} method.
+ */
+public abstract class DiffObserver implements Observer {
+
+ private NodeState before = null;
+
+ /**
+ * Returns the diff handler to be used for the given content change.
+ *
+ * @param before state of the repository before this changes
+ * @param after state of the repository after this changes
+ * @param info local commit information, or {@code null} if not available
+ * @return diff handler for this change
+ */
+ protected abstract NodeStateDiff getRootDiff(
+ @Nonnull NodeState before, @Nonnull NodeState after,
+ @Nullable CommitInfo info);
+
+ //----------------------------------------------------------< Observer >--
+
+ @Override
+ public final synchronized void contentChanged(
+ @Nonnull NodeState root, @Nullable CommitInfo info) {
+ checkNotNull(root);
+ if (before != null) {
+ NodeStateDiff diff = getRootDiff(before, root, info);
+ root.compareAgainstBaseState(before, diff);
+ }
+ before = root;
+ }
+
+}