001 /**
002 * Copyright (C) 2011-2012 Barchart, Inc. <http://www.barchart.com/>
003 *
004 * All rights reserved. Licensed under the OSI BSD License.
005 *
006 * http://www.opensource.org/licenses/bsd-license.php
007 */
008 package com.barchart.util.concurrent;
009
010 import java.util.List;
011 import java.util.concurrent.Callable;
012 import java.util.concurrent.CopyOnWriteArrayList;
013 import java.util.concurrent.FutureTask;
014 import java.util.concurrent.locks.Lock;
015 import java.util.concurrent.locks.ReentrantLock;
016
017 import org.slf4j.Logger;
018 import org.slf4j.LoggerFactory;
019
020 /**
021 * <p>
022 * Basic implementation of {@link FutureCallback} based on {@link FutureTask}.
023 * </p>
024 *
025 * @author jeremy
026 * @see FutureTask
027 * @see FutureCallback
028 * @param <E>
029 * The result type
030 */
031 public class FutureCallbackTask<E> extends FutureTask<E> implements
032 FutureCallback<E, FutureCallbackTask<E>> {
033
034 private final static Logger log = LoggerFactory
035 .getLogger(FutureCallbackTask.class);
036
037 private final List<FutureListener<E>> listeners =
038 new CopyOnWriteArrayList<FutureListener<E>>();
039 private final Lock callbackLock = new ReentrantLock();
040
041 /**
042 * Create a future result handler.
043 */
044 public FutureCallbackTask(final Runnable r, final E value) {
045 super(r, value);
046 }
047
048 /**
049 * Create a future result handler.
050 */
051 public FutureCallbackTask(final Callable<E> c) {
052 super(c);
053 }
054
055 @Override
056 public FutureCallbackTask<E> addResultListener(
057 final FutureListener<E> listener) {
058 callbackLock.lock();
059 try {
060 listeners.add(listener);
061 if (isDone()) {
062 try {
063 listener.resultAvailable(this);
064 } catch (final Exception ex) {
065 log.warn("Unhandled exception in callback", ex);
066 }
067 }
068 } finally {
069 callbackLock.unlock();
070 }
071 return this;
072 }
073
074 @Override
075 public FutureCallbackTask<E> succeed(final E result_) {
076 super.set(result_);
077 return this;
078 }
079
080 @Override
081 public FutureCallbackTask<E> fail(final Throwable error_) {
082 super.setException(error_);
083 return this;
084 }
085
086 @Override
087 public E getUnchecked() {
088 try {
089 return get();
090 } catch (final Exception e) {
091 return null;
092 }
093 }
094
095 /**
096 * Notify listeners that a result is available.
097 */
098 @Override
099 protected void done() {
100 callbackLock.lock();
101 try {
102 for (final FutureListener<E> l : listeners) {
103 try {
104 l.resultAvailable(this);
105 } catch (final Exception ex) {
106 log.warn("Unhandled exception in callback", ex);
107 }
108 }
109 } finally {
110 callbackLock.unlock();
111 }
112 }
113
114 /**
115 * Log an error if this FutureCallbackTask is garbage collected before it
116 * has been called.
117 */
118 @Override
119 public void finalize() throws Throwable {
120 try {
121 if (!isDone()) {
122 log.error("finalize() called on an incomplete FutureCallbackTask");
123 }
124 } catch (final Exception e) {
125 } finally {
126 super.finalize();
127 }
128 }
129
130 }