Added ListenableFuture interface

Added extension to Future with capabilities for registering callbacks
when the future is complete.

- Added ListenableFuture, ListenableFutureCallback,
  ListenableFutureCallbackRegistry, and ListenableFutureTask.
- Using ListenableFuture in AsyncRestOperations/AsyncRestTemplate.
- Added AsyncListenableTaskExecutor, implemented in
  SimpleAsyncTaskExecutor.
- Added FutureAdapter and ListenableFutureAdapter.

spring-core/src/main/java/org/springframework/core/task/AsyncListenableTaskExecutor.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * 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.springframework.core.task;
+
+import java.util.concurrent.Callable;
+
+import org.springframework.util.concurrent.ListenableFuture;
+
+/**
+ * Extension of the {@link AsyncTaskExecutor} interface, adding the capability to submit
+ * tasks for {@link ListenableFuture}s.
+ *
+ * @author Arjen Poutsma
+ * @since 4.0
+ * @see ListenableFuture
+ */
+public interface AsyncListenableTaskExecutor extends AsyncTaskExecutor {
+
+	/**
+	 * Submit a {@code Runnable} task for execution, receiving a {@code ListenableFuture}
+	 * representing that task. The Future will return a {@code null} result upon completion.
+	 * @param task the {@code Runnable} to execute (never {@code null})
+	 * @return a {@code ListenableFuture} representing pending completion of the task
+	 * @throws TaskRejectedException if the given task was not accepted
+	 */
+	ListenableFuture<?> submitListenable(Runnable task);
+
+	/**
+	 * Submit a {@code Callable} task for execution, receiving a {@code ListenableFuture}
+	 * representing that task. The Future will return the Callable's result upon
+	 * completion.
+	 * @param task the {@code Callable} to execute (never {@code null})
+	 * @return a {@code ListenableFuture} representing pending completion of the task
+	 * @throws TaskRejectedException if the given task was not accepted
+	 */
+	<T> ListenableFuture<T> submitListenable(Callable<T> task);
+
+}

spring-core/src/main/java/org/springframework/core/task/SimpleAsyncTaskExecutor.java

@@ -25,6 +25,8 @@ import java.util.concurrent.ThreadFactory;
 import org.springframework.util.Assert;
 import org.springframework.util.ConcurrencyThrottleSupport;
 import org.springframework.util.CustomizableThreadCreator;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.ListenableFutureTask;
 
 /**
  * {@link TaskExecutor} implementation that fires up a new Thread for each task,
@@ -45,7 +47,7 @@ import org.springframework.util.CustomizableThreadCreator;
  * @see org.springframework.scheduling.commonj.WorkManagerTaskExecutor
  */
 @SuppressWarnings("serial")
-public class SimpleAsyncTaskExecutor extends CustomizableThreadCreator implements AsyncTaskExecutor, Serializable {
+public class SimpleAsyncTaskExecutor extends CustomizableThreadCreator implements AsyncListenableTaskExecutor, Serializable {
 
 	/**
 	 * Permit any number of concurrent invocations: that is, don't throttle concurrency.
@@ -184,6 +186,20 @@ public class SimpleAsyncTaskExecutor extends CustomizableThreadCreator implement
 		return future;
 	}
 
+	@Override
+	public ListenableFuture<?> submitListenable(Runnable task) {
+		ListenableFutureTask<Object> future = new ListenableFutureTask<Object>(task, null);
+		execute(future, TIMEOUT_INDEFINITE);
+		return future;
+	}
+
+	@Override
+	public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
+		ListenableFutureTask<T> future = new ListenableFutureTask<T>(task);
+		execute(future, TIMEOUT_INDEFINITE);
+		return future;
+	}
+
 	/**
 	 * Template method for the actual execution of a task.
 	 * <p>The default implementation creates a new Thread and starts it.

spring-core/src/main/java/org/springframework/util/concurrent/FutureAdapter.java

@@ -0,0 +1,122 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * 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.springframework.util.concurrent;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.springframework.util.Assert;
+
+/**
+ * Abstract class that adapts a {@link Future} parameterized over S into a {@code
+ * Future} parameterized over T. All methods are delegated to the adaptee, where {@link
+ * #get()} and {@link #get(long, TimeUnit)} call {@@link #adapt(Object)} on the adaptee's
+ * result.
+ *
+ * @param <T> the type of this {@code Future}
+ * @param <S> the type of the adaptee's {@code Future}
+ * @author Arjen Poutsma
+ * @since 4.0
+ */
+public abstract class FutureAdapter<T, S> implements Future<T> {
+
+	private final Future<S> adaptee;
+
+	private Object result = null;
+
+	private State state = State.NEW;
+
+	private final Object mutex = new Object();
+
+	/**
+	 * Constructs a new {@code FutureAdapter} with the given adaptee.
+	 * @param adaptee the future to delegate to
+	 */
+	protected FutureAdapter(Future<S> adaptee) {
+		Assert.notNull(adaptee, "'delegate' must not be null");
+		this.adaptee = adaptee;
+	}
+
+	/**
+	 * Returns the adaptee.
+	 */
+	protected Future<S> getAdaptee() {
+		return adaptee;
+	}
+
+	@Override
+	public boolean cancel(boolean mayInterruptIfRunning) {
+		return adaptee.cancel(mayInterruptIfRunning);
+	}
+
+	@Override
+	public boolean isCancelled() {
+		return adaptee.isCancelled();
+	}
+
+	@Override
+	public boolean isDone() {
+		return adaptee.isDone();
+	}
+
+	@Override
+	public T get() throws InterruptedException, ExecutionException {
+		return adaptInternal(adaptee.get());
+	}
+
+	@Override
+	public T get(long timeout, TimeUnit unit)
+			throws InterruptedException, ExecutionException, TimeoutException {
+		return adaptInternal(adaptee.get(timeout, unit));
+	}
+
+	@SuppressWarnings("unchecked")
+	final T adaptInternal(S adapteeResult) throws ExecutionException {
+		synchronized (mutex) {
+			switch (state) {
+				case SUCCESS:
+					return (T) result;
+				case FAILURE:
+					throw (ExecutionException) result;
+				case NEW:
+					try {
+						T adapted = adapt(adapteeResult);
+						result = adapted;
+						state = State.SUCCESS;
+						return adapted;
+					} catch (ExecutionException ex) {
+						result = ex;
+						state = State.FAILURE;
+						throw ex;
+					}
+				default:
+					throw new IllegalStateException();
+			}
+		}
+	}
+
+	/**
+	 * Adapts the given adaptee's result into T.
+	 * @return the adapted result
+	 */
+	protected abstract T adapt(S adapteeResult) throws ExecutionException;
+
+	private enum State {NEW, SUCCESS, FAILURE}
+
+}

spring-core/src/main/java/org/springframework/util/concurrent/ListenableFuture.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * 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.springframework.util.concurrent;
+
+import java.util.concurrent.Future;
+
+/**
+ * Extends the {@link Future} interface with the capability to accept completion
+ * callbacks. If the future has already completed when the callback is added, the
+ * callback will be triggered immediately.
+ * <p>Inspired by {@link com.google.common.util.concurrent.ListenableFuture}.
+
+ * @author Arjen Poutsma
+ * @since 4.0
+ */
+public interface ListenableFuture<T> extends Future<T> {
+
+	/**
+	 * Registers the given callback to this {@code ListenableFuture}. The callback will
+	 * be triggered when this {@code Future} is complete or, if it is already complete,
+	 * immediately.
+	 *
+	 * @param callback the callback to register
+	 */
+	void addCallback(ListenableFutureCallback<? super T> callback);
+
+}

spring-core/src/main/java/org/springframework/util/concurrent/ListenableFutureAdapter.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * 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.springframework.util.concurrent;
+
+import java.util.concurrent.ExecutionException;
+
+/**
+ * Abstract class that adapts a {@link ListenableFuture} parameterized over S into a
+ * {@code ListenableFuture} parameterized over T. All methods are delegated to the
+ * adaptee, where {@link #get()}, {@link #get(long, java.util.concurrent.TimeUnit)}, and
+ * {@link ListenableFutureCallback#onSuccess(Object)} call {@@link #adapt(Object)} on the
+ * adaptee's result.
+ *
+ * @param <T> the type of this {@code Future}
+ * @param <S> the type of the adaptee's {@code Future}
+ * @author Arjen Poutsma
+ * @since 4.0
+ */
+public abstract class ListenableFutureAdapter<T, S> extends FutureAdapter<T, S>
+		implements ListenableFuture<T> {
+
+	/**
+	 * Constructs a new {@code ListenableFutureAdapter} with the given adaptee.
+	 * @param adaptee the future to adaptee to
+	 */
+	protected ListenableFutureAdapter(ListenableFuture<S> adaptee) {
+		super(adaptee);
+	}
+
+
+	@Override
+	public void addCallback(final ListenableFutureCallback<? super T> callback) {
+		ListenableFuture<S> listenableAdaptee = (ListenableFuture<S>) getAdaptee();
+		listenableAdaptee.addCallback(new ListenableFutureCallback<S>() {
+			@Override
+			public void onSuccess(S result) {
+				try {
+					callback.onSuccess(adaptInternal(result));
+				}
+				catch (ExecutionException ex) {
+					Throwable cause = ex.getCause();
+					onFailure(cause != null ? cause : ex);
+				}
+				catch (Throwable t) {
+					onFailure(t);
+				}
+			}
+
+			@Override
+			public void onFailure(Throwable t) {
+				callback.onFailure(t);
+			}
+		});
+	}
+}

spring-core/src/main/java/org/springframework/util/concurrent/ListenableFutureCallback.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * 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.springframework.util.concurrent;
+
+/**
+ * Defines the contract for callbacks that accept the result of a
+ * {@link ListenableFuture}.
+ *
+ * @author Arjen Poutsma
+ * @since 4.0
+ */
+public interface ListenableFutureCallback<T> {
+
+	/**
+	 * Called when the {@link ListenableFuture} successfully completes.
+	 * @param result the result
+	 */
+	void onSuccess(T result);
+
+	/**
+	 * Called when the {@link ListenableFuture} fails to complete.
+	 * @param t the exception that triggered the failure
+	 */
+	void onFailure(Throwable t);
+
+}

spring-core/src/main/java/org/springframework/util/concurrent/ListenableFutureCallbackRegistry.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * 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.springframework.util.concurrent;
+
+import java.util.LinkedList;
+import java.util.Queue;
+
+import org.springframework.util.Assert;
+
+/**
+ * Registry for {@link ListenableFutureCallback} instances.
+ * <p>Inspired by {@link com.google.common.util.concurrent.ExecutionList}.
+ * @author Arjen Poutsma
+ * @since 4.0
+ */
+public class ListenableFutureCallbackRegistry<T> {
+
+	private final Queue<ListenableFutureCallback<? super T>> callbacks =
+			new LinkedList<ListenableFutureCallback<? super T>>();
+
+	private State state = State.NEW;
+
+	private Object result = null;
+
+	private final Object mutex = new Object();
+
+
+	/**
+	 * Adds the given callback to this registry.
+	 * @param callback the callback to add
+	 */
+	@SuppressWarnings("unchecked")
+	public void addCallback(ListenableFutureCallback<? super T> callback) {
+		Assert.notNull(callback, "'callback' must not be null");
+
+		synchronized (mutex) {
+			switch (state) {
+				case NEW:
+					callbacks.add(callback);
+					break;
+				case SUCCESS:
+					callback.onSuccess((T)result);
+					break;
+				case FAILURE:
+					callback.onFailure((Throwable) result);
+					break;
+			}
+		}
+	}
+
+	/**
+	 * Triggers a {@link ListenableFutureCallback#onSuccess(Object)} call on all added
+	 * callbacks with the given result
+	 * @param result the result to trigger the callbacks with
+	 */
+	public void success(T result) {
+		synchronized (mutex) {
+			state = State.SUCCESS;
+			this.result = result;
+
+			while (!callbacks.isEmpty()) {
+				callbacks.poll().onSuccess(result);
+			}
+		}
+	}
+
+	/**
+	 * Triggers a {@link ListenableFutureCallback#onFailure(Throwable)} call on all added
+	 * callbacks with the given {@code Throwable}.
+	 * @param t the exception to trigger the callbacks with
+	 */
+	public void failure(Throwable t) {
+		synchronized (mutex) {
+			state = State.FAILURE;
+			this.result = t;
+
+			while (!callbacks.isEmpty()) {
+				callbacks.poll().onFailure(t);
+			}
+		}
+	}
+
+	private enum State {NEW, SUCCESS, FAILURE}
+
+}

spring-core/src/main/java/org/springframework/util/concurrent/ListenableFutureTask.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * 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.springframework.util.concurrent;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
+
+/**
+ * Extension of {@link FutureTask} that implements {@link ListenableFuture}.
+ *
+ * @author Arjen Poutsma
+ * @since 4.0
+ */
+public class ListenableFutureTask<T> extends FutureTask<T>
+		implements ListenableFuture<T> {
+
+	private final ListenableFutureCallbackRegistry<T> callbacks =
+			new ListenableFutureCallbackRegistry<T>();
+
+	/**
+	 * Creates a new {@code ListenableFutureTask} that will, upon running, execute the given
+	 * {@link Callable}.
+	 * @param callable the callable task
+	 */
+	public ListenableFutureTask(Callable<T> callable) {
+		super(callable);
+	}
+
+	/**
+	 * Creates a {@code ListenableFutureTask} that will, upon running, execute the given
+	 * {@link Runnable}, and arrange that {@link #get()} will return the given result on
+	 * successful completion.
+	 * @param runnable the runnable task
+	 * @param result the result to return on successful completion
+	 */
+	public ListenableFutureTask(Runnable runnable, T result) {
+		super(runnable, result);
+	}
+
+	@Override
+	public void addCallback(ListenableFutureCallback<? super T> callback) {
+		callbacks.addCallback(callback);
+	}
+
+	@Override
+	protected final void done() {
+		Throwable cause;
+		try {
+			T result = get();
+			callbacks.success(result);
+			return;
+		}
+		catch (InterruptedException ex) {
+			Thread.currentThread().interrupt();
+			return;
+		}
+		catch (ExecutionException ex) {
+			cause = ex.getCause();
+			if (cause == null) {
+				cause = ex;
+			}
+		}
+		catch (Throwable t) {
+			cause = t;
+		}
+		callbacks.failure(cause);
+	}
+}

spring-core/src/main/java/org/springframework/util/concurrent/package-info.java

@@ -0,0 +1,7 @@
+
+/**
+ *
+ * Useful generic {@code java.util.concurrent.Future} extension.
+ */
+package org.springframework.util.concurrent;
+