| 13 |
|
*/ |
| 14 |
|
public class ExecutorCompletionService<V> implements CompletionService<V> { |
| 15 |
|
private final Executor executor; |
| 16 |
< |
private final LinkedBlockingQueue<Future<V>> cq = |
| 17 |
< |
new LinkedBlockingQueue<Future<V>>(); |
| 16 |
> |
private final BlockingQueue<Future<V>> completionQueue; |
| 17 |
|
|
| 18 |
|
/** |
| 19 |
|
* FutureTask extension to enqueue upon completion |
| 21 |
|
private class QueueingFuture<T> extends FutureTask<T> { |
| 22 |
|
QueueingFuture(Callable<T> c) { super(c); } |
| 23 |
|
QueueingFuture(Runnable t, T r) { super(t, r); } |
| 24 |
< |
protected void done() { cq.add((Future<V>)this); } |
| 24 |
> |
protected void done() { completionQueue.add((Future<V>)this); } |
| 25 |
|
} |
| 26 |
|
|
| 27 |
|
/** |
| 28 |
|
* Creates an ExecutorCompletionService using the supplied |
| 29 |
< |
* executor for base task execution. Normally, this |
| 30 |
< |
* executor should be dedicated for use by this service |
| 31 |
< |
8 @throws NullPointerException if executor is null |
| 29 |
> |
* executor for base task execution and a |
| 30 |
> |
* {@link LinkedBlockingQueue} as a completion queue. |
| 31 |
> |
* @param executor the executor to use; normally |
| 32 |
> |
* one dedicated for use by this service |
| 33 |
> |
8 @throws NullPointerException if executor is <tt>null</tt> |
| 34 |
|
*/ |
| 35 |
|
public ExecutorCompletionService(Executor executor) { |
| 36 |
|
if (executor == null) |
| 37 |
|
throw new NullPointerException(); |
| 38 |
|
this.executor = executor; |
| 39 |
+ |
this.completionQueue = new LinkedBlockingQueue<Future<V>>(); |
| 40 |
|
} |
| 41 |
|
|
| 40 |
– |
|
| 42 |
|
/** |
| 43 |
< |
* Return the {@link Executor} used for base |
| 44 |
< |
* task execution. This may for example be used to shut |
| 45 |
< |
* down the service. |
| 46 |
< |
* @return the executor |
| 43 |
> |
* Creates an ExecutorCompletionService using the supplied |
| 44 |
> |
* executor for base task execution and the supplied queue as its |
| 45 |
> |
* completion queue. |
| 46 |
> |
* @param executor the executor to use; normally |
| 47 |
> |
* one dedicated for use by this service |
| 48 |
> |
* @param completionQueue the queue to use as the completion queue; |
| 49 |
> |
* normally one dedicated for use by this service |
| 50 |
> |
8 @throws NullPointerException if executor or completionQueue are <tt>null</tt> |
| 51 |
|
*/ |
| 52 |
< |
public Executor getExecutor() { |
| 53 |
< |
return executor; |
| 52 |
> |
public ExecutorCompletionService(Executor executor, |
| 53 |
> |
BlockingQueue<Future<V>> completionQueue) { |
| 54 |
> |
if (executor == null || completionQueue == null) |
| 55 |
> |
throw new NullPointerException(); |
| 56 |
> |
this.executor = executor; |
| 57 |
> |
this.completionQueue = completionQueue; |
| 58 |
|
} |
| 59 |
|
|
| 60 |
|
public Future<V> submit(Callable<V> task) { |
| 70 |
|
} |
| 71 |
|
|
| 72 |
|
public Future<V> take() throws InterruptedException { |
| 73 |
< |
return cq.take(); |
| 73 |
> |
return completionQueue.take(); |
| 74 |
|
} |
| 75 |
|
|
| 76 |
|
public Future<V> poll() { |
| 77 |
< |
return cq.poll(); |
| 77 |
> |
return completionQueue.poll(); |
| 78 |
|
} |
| 79 |
|
|
| 80 |
|
public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException { |
| 81 |
< |
return cq.poll(timeout, unit); |
| 81 |
> |
return completionQueue.poll(timeout, unit); |
| 82 |
|
} |
| 83 |
|
} |
| 84 |
|
|
