diff --git a/lucene/core/src/java/org/apache/lucene/search/NRTManager.java b/lucene/core/src/java/org/apache/lucene/search/NRTManager.java index cfe8028..b708662 100644 --- a/lucene/core/src/java/org/apache/lucene/search/NRTManager.java +++ b/lucene/core/src/java/org/apache/lucene/search/NRTManager.java @@ -70,7 +70,7 @@ import org.apache.lucene.util.ThreadInterruptedException; * @lucene.experimental */ -public class NRTManager extends ReferenceManager { +public final class NRTManager extends ReferenceManager { private static final long MAX_SEARCHER_GEN = Long.MAX_VALUE; private final TrackingIndexWriter writer; private final List waitingListeners = new CopyOnWriteArrayList(); @@ -361,7 +361,7 @@ public class NRTManager extends ReferenceManager { } @Override - protected void afterRefresh() { + protected void afterMaybeRefresh() throws IOException { genLock.lock(); try { if (searchingGen != MAX_SEARCHER_GEN) { @@ -374,6 +374,7 @@ public class NRTManager extends ReferenceManager { } finally { genLock.unlock(); } + super.afterMaybeRefresh(); } @Override @@ -386,6 +387,7 @@ public class NRTManager extends ReferenceManager { } finally { genLock.unlock(); } + super.afterClose(); } /** diff --git a/lucene/core/src/java/org/apache/lucene/search/ReferenceManager.java b/lucene/core/src/java/org/apache/lucene/search/ReferenceManager.java index 21356c7..8ecac06 100755 --- a/lucene/core/src/java/org/apache/lucene/search/ReferenceManager.java +++ b/lucene/core/src/java/org/apache/lucene/search/ReferenceManager.java @@ -19,6 +19,8 @@ package org.apache.lucene.search; import java.io.Closeable; import java.io.IOException; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -44,7 +46,9 @@ public abstract class ReferenceManager implements Closeable { protected volatile G current; private final Lock refreshLock = new ReentrantLock(); - + + private final List listeners = new CopyOnWriteArrayList(); + private void ensureOpen() { if (current == null) { throw new AlreadyClosedException(REFERENCE_MANAGER_IS_CLOSED_MSG); @@ -132,6 +136,9 @@ public abstract class ReferenceManager implements Closeable { * @throws IOException if the after close operation in a sub-class throws an {@link IOException} * */ protected void afterClose() throws IOException { + for (Listener listener : listeners) { + listener.afterClose(); + } } private void doMaybeRefresh() throws IOException { @@ -144,16 +151,16 @@ public abstract class ReferenceManager implements Closeable { refreshLock.lock(); try { final G reference = acquire(); + boolean swapped = false; try { G newReference = refreshIfNeeded(reference); if (newReference != null) { assert newReference != reference : "refreshIfNeeded should return null if refresh wasn't needed"; - boolean success = false; try { swapReference(newReference); - success = true; + swapped = true; } finally { - if (!success) { + if (!swapped) { release(newReference); } } @@ -161,7 +168,8 @@ public abstract class ReferenceManager implements Closeable { } finally { release(reference); } - afterRefresh(); + if (swapped) + afterRefresh(); } finally { refreshLock.unlock(); } @@ -195,6 +203,7 @@ public abstract class ReferenceManager implements Closeable { if (doTryRefresh) { try { doMaybeRefresh(); + afterMaybeRefresh(); } finally { refreshLock.unlock(); } @@ -223,16 +232,27 @@ public abstract class ReferenceManager implements Closeable { refreshLock.lock(); try { doMaybeRefresh(); + afterMaybeRefresh(); } finally { refreshLock.unlock(); } } + /** Called after {@link #maybeRefresh()}, always called even if no new instance was installed. + * @throws IOException if a low level I/O exception occurs + **/ + protected void afterMaybeRefresh() throws IOException { + + } + /** Called after swapReference has installed a new * instance. * @throws IOException if a low level I/O exception occurs **/ protected void afterRefresh() throws IOException { + for (Listener listener : listeners) { + listener.afterRefresh(); + } } /** @@ -245,4 +265,31 @@ public abstract class ReferenceManager implements Closeable { assert reference != null; decRef(reference); } + + /** Adds a listener, to be notified when a reference is refreshed/swapped. */ + public void addReferenceListener(Listener listener) { + if (listener == null) + throw new NullPointerException("Listener cannot be null"); + listeners.add(listener); + } + + /** Remove a listener added with {@link #addReferenceListener}. */ + public void removeReferenceListener(Listener listener) { + if (listener == null) + throw new NullPointerException("Listener cannot be null"); + listeners.remove(listener); + } + + public interface Listener { + + /** + * Called after a successful refresh and a new reference has been installed. When this is called {@link #acquire()} is guaranteed to return a new instance. + */ + void afterRefresh(); + + /** + * Called after the manager was closed, so any resource can be freed + */ + void afterClose(); + } } diff --git a/lucene/core/src/test/org/apache/lucene/search/TestNRTManager.java b/lucene/core/src/test/org/apache/lucene/search/TestNRTManager.java index d864a44..844cf5c 100644 --- a/lucene/core/src/test/org/apache/lucene/search/TestNRTManager.java +++ b/lucene/core/src/test/org/apache/lucene/search/TestNRTManager.java @@ -411,4 +411,34 @@ public class TestNRTManager extends ThreadedIndexingAndSearchingTestCase { other.close(); dir.close(); } + + public void testListenersCalled() throws Exception { + Directory dir = newDirectory(); + IndexWriter iw = new IndexWriter(dir, new IndexWriterConfig(TEST_VERSION_CURRENT, null)); + final AtomicBoolean afterRefreshCalled = new AtomicBoolean(false); + final AtomicBoolean afterCloseCalled = new AtomicBoolean(false); + NRTManager sm = new NRTManager(new NRTManager.TrackingIndexWriter(iw),new SearcherFactory()); + sm.addReferenceListener(new ReferenceManager.Listener() { + @Override + public void afterRefresh() { + afterRefreshCalled.set(true); + } + + @Override + public void afterClose() { + afterCloseCalled.set(true); + } + }); + iw.addDocument(new Document()); + iw.commit(); + assertFalse(afterRefreshCalled.get()); + assertFalse(afterCloseCalled.get()); + sm.maybeRefreshBlocking(); + assertTrue(afterRefreshCalled.get()); + assertFalse(afterCloseCalled.get()); + sm.close(); + assertTrue(afterCloseCalled.get()); + iw.close(); + dir.close(); + } } diff --git a/lucene/core/src/test/org/apache/lucene/search/TestSearcherManager.java b/lucene/core/src/test/org/apache/lucene/search/TestSearcherManager.java index 3e2ac5f..4a41a82 100644 --- a/lucene/core/src/test/org/apache/lucene/search/TestSearcherManager.java +++ b/lucene/core/src/test/org/apache/lucene/search/TestSearcherManager.java @@ -324,6 +324,36 @@ public class TestSearcherManager extends ThreadedIndexingAndSearchingTestCase { dir.close(); } + public void testListenersCalled() throws Exception { + Directory dir = newDirectory(); + IndexWriter iw = new IndexWriter(dir, new IndexWriterConfig(TEST_VERSION_CURRENT, null)); + final AtomicBoolean afterRefreshCalled = new AtomicBoolean(false); + final AtomicBoolean afterCloseCalled = new AtomicBoolean(false); + SearcherManager sm = new SearcherManager(iw, false, new SearcherFactory()); + sm.addReferenceListener(new ReferenceManager.Listener() { + @Override + public void afterRefresh() { + afterRefreshCalled.set(true); + } + + @Override + public void afterClose() { + afterCloseCalled.set(true); + } + }); + iw.addDocument(new Document()); + iw.commit(); + assertFalse(afterRefreshCalled.get()); + assertFalse(afterCloseCalled.get()); + sm.maybeRefreshBlocking(); + assertTrue(afterRefreshCalled.get()); + assertFalse(afterCloseCalled.get()); + sm.close(); + assertTrue(afterCloseCalled.get()); + iw.close(); + dir.close(); + } + public void testEvilSearcherFactory() throws Exception { final Random random = random(); final Directory dir = newDirectory();