Introduces a type for asynchronous indexer operation

fcrepo4-archive · Dec 8, 2013 · 002b1c4 · 002b1c4
1 parent 833e68e
commit 002b1c4
Show file tree

Hide file tree

Showing 2 changed files with 120 additions and 47 deletions.
diff --git a/fcrepo-jms-indexer-core/src/main/java/org/fcrepo/indexer/AsynchIndexer.java b/fcrepo-jms-indexer-core/src/main/java/org/fcrepo/indexer/AsynchIndexer.java
@@ -0,0 +1,83 @@
+/**
+ * Copyright 2013 DuraSpace, Inc.
+ *
+ * Licensed 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.fcrepo.indexer;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.io.IOException;
+import java.io.Reader;
+
+import org.slf4j.Logger;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListenableFutureTask;
+import com.google.common.util.concurrent.ListeningExecutorService;
+
+/**
+ * An {@link Indexer} that executes its operation asynchronously.
+ *
+ * @author ajs6f
+ * @date Dec 8, 2013
+ * @param <T>
+ */
+public abstract class AsynchIndexer<T> implements Indexer {
+
+    private static final Logger LOGGER = getLogger(AsynchIndexer.class);
+
+    abstract ListeningExecutorService executorService();
+
+    @Override
+    public ListenableFuture<T> update(final String identifier,
+        final Reader content) throws IOException {
+        LOGGER.debug("Received update for identifier: {}", identifier);
+
+        final ListenableFutureTask<T> task = updateSynch(identifier, content);
+        task.addListener(new Runnable() {
+            @Override
+            public void run() {
+                synchronized (this) {
+                    notifyAll();
+                }
+            }
+        }, executorService());
+        executorService().submit(task);
+        return task;
+    }
+
+    @Override
+    public ListenableFuture<T> remove(final String identifier)
+        throws IOException {
+        LOGGER.debug("Received remove for identifier: {}", identifier);
+        final ListenableFutureTask<T> task = removeSynch(identifier);
+        task.addListener(new Runnable() {
+            @Override
+            public void run() {
+                synchronized (this) {
+                    notifyAll();
+                }
+            }
+        }, executorService());
+        executorService().submit(task);
+        return task;
+    }
+
+    abstract ListenableFutureTask<T> removeSynch(final String identifier);
+
+    abstract ListenableFutureTask<T> updateSynch(final String identifier,
+            final Reader content);
+
+}
diff --git a/fcrepo-jms-indexer-core/src/test/java/org/fcrepo/indexer/TestIndexer.java b/fcrepo-jms-indexer-core/src/test/java/org/fcrepo/indexer/TestIndexer.java
@@ -16,19 +16,19 @@
 
 package org.fcrepo.indexer;
 
+import static com.google.common.util.concurrent.MoreExecutors.sameThreadExecutor;
 import static org.fcrepo.indexer.Indexer.IndexerType.NO_CONTENT;
 import static org.slf4j.LoggerFactory.getLogger;
 
-import java.io.IOException;
 import java.io.Reader;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.Callable;
 
 import org.slf4j.Logger;
 
-import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListenableFutureTask;
+import com.google.common.util.concurrent.ListeningExecutorService;
 
 
 /**
@@ -38,84 +38,74 @@
  * @author Esmé Cowles
  * @date Nov 25, 2013
 **/
-public class TestIndexer implements Indexer {
+public class TestIndexer extends  AsynchIndexer<Boolean> {
 
     private static final Logger LOGGER = getLogger(TestIndexer.class);
 
     private final Set<String> updates = new HashSet<>();
     private final Set<String> removes = new HashSet<>();
 
+    private final ListeningExecutorService executorService =
+        sameThreadExecutor();
+
 
-    /**
-     * Create or update an index entry for the object.
-     *
-     * @return
-     **/
     @Override
-    public ListenableFuture<Boolean> update(final String pid,
-            final Reader content) throws IOException {
-        LOGGER.debug("Received update for identifier: {}", pid);
-        final ListenableFutureTask<Boolean> result = ListenableFutureTask.create(new Callable<Boolean>() {
+    public ListenableFutureTask<Boolean> updateSynch(final String identifier,
+        final Reader content) {
+        LOGGER.debug("Received update for identifier: {}", identifier);
+        return ListenableFutureTask.create(new Callable<Boolean>() {
 
             @Override
             public Boolean call() throws Exception {
-                final Boolean success = updates.add(pid);
+                final Boolean success = updates.add(identifier);
                 LOGGER.debug("Current recorded updates include: {}", updates);
                 return success;
             }
         });
-        result.run();
-        synchronized(this) {
-            notifyAll();
-            return result;
-        }
     }
 
-    /**
-     * Remove the object from the index.
-     * @return
-    **/
     @Override
-    public ListenableFuture<Boolean> remove(final String pid)
-        throws IOException {
-        LOGGER.debug("Received remove for identifier: {}", pid);
-        final ListenableFutureTask<Boolean> result = ListenableFutureTask.create(new Callable<Boolean>() {
-
-            @Override
-            public Boolean call() throws Exception {
-                final Boolean success = removes.add(pid);
-                LOGGER.debug("Current recorded removes include: {}", removes);
-                return success;
-            }
-        });
-        result.run();
-        synchronized (this) {
-            notifyAll();
-            return result;
-        }
+    public ListenableFutureTask<Boolean>
+    removeSynch(final String identifier) {
+        LOGGER.debug("Received remove for identifier: {}", identifier);
+        return ListenableFutureTask.create(new Callable<Boolean>() {
+
+                @Override
+                public Boolean call() throws Exception {
+                    final Boolean success = removes.add(identifier);
+                    LOGGER.debug("Current recorded removes include: {}",
+                            removes);
+                    return success;
+                }
+            });
 
     }
 
     /**
      * Test whether an update message has been received for a PID.
     **/
-    public boolean receivedUpdate(final String pid) {
-        LOGGER.debug("Checked whether we received an update for: {}, {}", pid,
-                updates.contains(pid));
-        return updates.contains(pid);
+    public boolean receivedUpdate(final String id) {
+        LOGGER.debug("Checked whether we received an update for: {}, {}", id,
+                updates.contains(id));
+        return updates.contains(id);
     }
 
     /**
      * Test whether a remove message has been received for a PID.
     **/
-    public boolean receivedRemove(final String pid) {
-        LOGGER.debug("Checked whether we received a remove for: {}, {}", pid,
-                removes.contains(pid));
-        return removes.contains(pid);
+    public boolean receivedRemove(final String id) {
+        LOGGER.debug("Checked whether we received a remove for: {}, {}", id,
+                removes.contains(id));
+        return removes.contains(id);
     }
 
     @Override
     public IndexerType getIndexerType() {
         return NO_CONTENT;
     }
+
+    @Override
+    ListeningExecutorService executorService() {
+        return executorService;
+    }
 }