Minor simplification of the pending Future<?> system.

stickfigure · stickfigure · commit 0a47bc0deb7a · 2014-05-06T12:49:51.000-07:00
diff --git a/src/main/java/com/googlecode/objectify/cache/Pending.java b/src/main/java/com/googlecode/objectify/cache/Pending.java
diff --git a/src/main/java/com/googlecode/objectify/cache/PendingFutures.java b/src/main/java/com/googlecode/objectify/cache/PendingFutures.java
@@ -1,80 +1,63 @@
 package com.googlecode.objectify.cache;
 
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 /**
- * <p>This bit of appengine magic hooks into the ApiProxy and does the heavy lifting of
- * making the TriggerFuture<?> work.</p>
- * 
  * <p>This class maintains a thread local list of all the outstanding Future<?> objects
  * that have pending triggers.  When a Future<?> is done and its trigger is executed,
- * it is removed from the list.  At various times (anytime an API call is made) the registered
- * futures are checked for doneness and processed.</p>
- * 
- * <p>The AsyncCacheFilter is necessary to guarantee that any pending triggers are processed
- * at the end of the request.  A future GAE SDK which allows us to hook into the Future<?>
- * creation process might make this extra Filter unnecessary.</p>
+ * it is removed from the list.</p>
  * 
  * @author Jeff Schnitzer <jeff@infohazard.org>
  */
 public class PendingFutures
 {
-	/** The thread local value will be removed (null) if there are none pending */
-	private static ThreadLocal<Pending> pending = new ThreadLocal<>();
-	
+	/** */
+	private static final Logger log = Logger.getLogger(PendingFutures.class.getName());
+
+	/**
+	 * We use ConcurrentHashMap not for concurrency but because it doesn't throw
+	 * ConcurrentModificationException.  We need to be able to iterate while Futures remove
+	 * themselves from the set. A Set is just a Map of key to key.
+	 */
+	private static ThreadLocal<ConcurrentHashMap<Future<?>, Future<?>>> pending = new ThreadLocal<ConcurrentHashMap<Future<?>, Future<?>>>() {
+		@Override
+		protected ConcurrentHashMap<Future<?>, Future<?>> initialValue() {
+			return new ConcurrentHashMap<>(64, 0.75f, 1);
+		}
+	};
 	
 	/**
 	 * Register a pending Future that has a callback.
 	 * @param future must have at least one callback
 	 */
-	public static void addPending(Future<?> future)
-	{
-		Pending pend = pending.get();
-		if (pend == null)
-		{
-			pend = new Pending();
-			pending.set(pend);
-		}
-		
-		pend.add(future);
+	public static void addPending(Future<?> future) {
+		pending.get().put(future, future);
 	}
 	
 	/**
 	 * Deregister a pending Future that had a callback.
 	 */
-	public static void removePending(Future<?> future)
-	{
-		Pending pend = pending.get();
-		if (pend != null)
-		{
-			pend.remove(future);
-			
-			// When the last one is gone, we don't need this thread local anymore
-			if (pend.isEmpty())
-				pending.remove();
-		}
+	public static void removePending(Future<?> future) {
+		pending.get().remove(future);
 	}
 	
-//	/**
-//	 * If any futures are pending, check if they are done.  This will process their callbacks.
-//	 * Don't use this method - see comments on Pending.checkPendingFutures()
-//	 */
-//	@Deprecated
-//	public static void checkPendingFutures()
-//	{
-//		Pending pend = pending.get();
-//		if (pend != null)
-//			pend.checkPendingFutures();
-//	}
-	
 	/**
 	 * Iterate through all pending futures and get() them, forcing any callbacks to be called.
-	 * This is used only by the AsyncCacheFilter because we don't have a proper hook otherwise.
+	 * This is used only by the AsyncCacheFilter (if using cache without Objectify) or ObjectifyFilter
+	 * (if using Objectify normally) because we don't have a proper hook otherwise.
 	 */
-	public static void completeAllPendingFutures()
-	{
-		Pending pend = pending.get();
-		if (pend != null)
-			pend.completeAllPendingFutures();
+	public static void completeAllPendingFutures() {
+		// This will cause done Futures to fire callbacks and remove themselves
+		for (Future<?> fut: pending.get().keySet()) {
+			try {
+				fut.get();
+			}
+			catch (Exception e) {
+				log.log(Level.SEVERE, "Error cleaning up pending Future: " + fut, e);
+			}
+		}
 	}
 }
diff --git a/src/main/java/com/googlecode/objectify/cache/TriggerFuture.java b/src/main/java/com/googlecode/objectify/cache/TriggerFuture.java
@@ -25,6 +25,9 @@
  * 
  * <p>Use the AsyncCacheFilter for normal requests. For situations where a filter is not appropriate
  * (ie, the remote api) be sure to call PendingFutures.completeAllPendingFutures() manually.</p>
+ *
+ * <p>Note that if you are using this with Objectify, you probably want to use ObjectifyFilter.complete()
+ * rather than PendingFutures or AsyncCacheFilter static methods.</p>
  * 
  * @author Jeff Schnitzer <jeff@infohazard.org>
  */
@@ -37,8 +40,7 @@ abstract public class TriggerFuture<T> implements Future<T>
 	boolean triggered = false;
 	
 	/** Wrap a normal Future<?> */
-	public TriggerFuture(Future<T> raw)
-	{
+	public TriggerFuture(Future<T> raw) {
 		this.raw = raw;
 		
 		// We now need to register ourself so that we'll get checked at future API calls
@@ -55,9 +57,7 @@ public TriggerFuture(Future<T> raw)
 	 * @see java.util.concurrent.Future#cancel(boolean)
 	 */
 	@Override
-	public boolean cancel(boolean mayInterruptIfRunning)
-	{
-		//return this.raw.cancel(mayInterruptIfRunning);
+	public boolean cancel(boolean mayInterruptIfRunning) {
 		throw new UnsupportedOperationException("This makes my head spin. Don't do it.");
 	}
 
@@ -77,12 +77,10 @@ public boolean isCancelled()
 	 * @see java.util.concurrent.Future#isDone()
 	 */
 	@Override
-	public boolean isDone()
-	{
+	public boolean isDone() {
 		boolean done = this.raw.isDone();
 		
-		if (!triggered && done)
-		{
+		if (!triggered && done) {
 			this.triggered = true;
 			PendingFutures.removePending(this);
 			
diff --git a/src/main/java/com/googlecode/objectify/cache/TriggerSuccessFuture.java b/src/main/java/com/googlecode/objectify/cache/TriggerSuccessFuture.java
@@ -32,8 +32,7 @@ public TriggerSuccessFuture(Future<T> raw)
 	/* (non-Javadoc)
 	 * @see com.googlecode.objectify.cache.NotifyFuture#execute()
 	 */
-	protected final void trigger()
-	{
+	protected final void trigger() {
 		try {
 			this.success(this.get());
 		} catch (Exception ex) {
