util/concurrent/ExecutorCompletionService.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain. Use, modify, and
 * redistribute this code in any way without acknowledgement.
 */

package java.util.concurrent;


/**
 * A {@link CompletionService} that uses a supplied {@link Executor}
 * to execute tasks.
 */
public class ExecutorCompletionService<V> implements CompletionService<V> {
    private final Executor executor;
    private final BlockingQueue<Future<V>> completionQueue;

    /**
     * FutureTask extension to enqueue upon completion
     */
    private class QueueingFuture<T> extends FutureTask<T> {
        QueueingFuture(Callable<T> c) { super(c); }
        QueueingFuture(Runnable t, T r) { super(t, r); }
        protected void done() { completionQueue.add((Future<V>)this); }
    }

    /**
     * Creates an ExecutorCompletionService using the supplied
     * executor for base task execution and a
     * {@link LinkedBlockingQueue} as a completion queue.
     * @param executor the executor to use; normally
     * one dedicated for use by this service
     8 @throws NullPointerException if executor is <tt>null</tt>
     */
    public ExecutorCompletionService(Executor executor) {
        if (executor == null) 
            throw new NullPointerException();
        this.executor = executor;
        this.completionQueue = new LinkedBlockingQueue<Future<V>>();
    }

    /**
     * Creates an ExecutorCompletionService using the supplied
     * executor for base task execution and the supplied queue as its
     * completion queue.
     * @param executor the executor to use; normally
     * one dedicated for use by this service
     * @param completionQueue the queue to use as the completion queue;
     * normally one dedicated for use by this service
     8 @throws NullPointerException if executor or completionQueue are <tt>null</tt>
     */
    public ExecutorCompletionService(Executor executor,
                                     BlockingQueue<Future<V>> completionQueue) {
        if (executor == null || completionQueue == null) 
            throw new NullPointerException();
        this.executor = executor;
        this.completionQueue = completionQueue;
    }

    public Future<V> submit(Callable<V> task) {
        QueueingFuture<V> f = new QueueingFuture<V>(task);
        executor.execute(f);
        return f;
    }

    public Future<V> submit(Runnable task, V result) {
        QueueingFuture<V> f = new QueueingFuture<V>(task, result);
        executor.execute(f);
        return f;
    }

    public Future<V> take() throws InterruptedException {
        return completionQueue.take();
    }

    public Future<V> poll() {
        return completionQueue.poll();
    }

    public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException {
        return completionQueue.poll(timeout, unit);
    }
}


Revision:	1.3
Committed:	Mon Dec 15 12:06:00 2003 UTC (20 years, 5 months ago) by dl
Branch:	MAIN
Changes since 1.2:	+25 -16 lines
Log Message:	Improved documentation; changed constructors and accesors
#	User	Rev	Content
1	dl	1.1	/*
2			* Written by Doug Lea with assistance from members of JCP JSR-166
3			* Expert Group and released to the public domain. Use, modify, and
4			* redistribute this code in any way without acknowledgement.
5			*/
6
7			package java.util.concurrent;
8
9
10			/**
11	jozart	1.2	* A {@link CompletionService} that uses a supplied {@link Executor}
12	dl	1.1	* to execute tasks.
13			*/
14			public class ExecutorCompletionService<V> implements CompletionService<V> {
15	jozart	1.2	private final Executor executor;
16	dl	1.3	private final BlockingQueue<Future<V>> completionQueue;
17	dl	1.1
18			/**
19			* FutureTask extension to enqueue upon completion
20			*/
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	dl	1.3	protected void done() { completionQueue.add((Future<V>)this); }
25	dl	1.1	}
26
27			/**
28			* Creates an ExecutorCompletionService using the supplied
29	dl	1.3	* 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	dl	1.1	*/
35	jozart	1.2	public ExecutorCompletionService(Executor executor) {
36	dl	1.1	if (executor == null)
37			throw new NullPointerException();
38			this.executor = executor;
39	dl	1.3	this.completionQueue = new LinkedBlockingQueue<Future<V>>();
40	dl	1.1	}
41
42			/**
43	dl	1.3	* 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	dl	1.1	*/
52	dl	1.3	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	dl	1.1	}
59
60			public Future<V> submit(Callable<V> task) {
61			QueueingFuture<V> f = new QueueingFuture<V>(task);
62			executor.execute(f);
63			return f;
64			}
65
66			public Future<V> submit(Runnable task, V result) {
67			QueueingFuture<V> f = new QueueingFuture<V>(task, result);
68			executor.execute(f);
69			return f;
70			}
71
72			public Future<V> take() throws InterruptedException {
73	dl	1.3	return completionQueue.take();
74	dl	1.1	}
75
76			public Future<V> poll() {
77	dl	1.3	return completionQueue.poll();
78	dl	1.1	}
79
80			public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException {
81	dl	1.3	return completionQueue.poll(timeout, unit);
82	dl	1.1	}
83			}
84
85