001    /**
002     * Licensed to the Apache Software Foundation (ASF) under one or more
003     * contributor license agreements.  See the NOTICE file distributed with
004     * this work for additional information regarding copyright ownership.
005     * The ASF licenses this file to You under the Apache License, Version 2.0
006     * (the "License"); you may not use this file except in compliance with
007     * the License.  You may obtain a copy of the License at
008     *
009     *      http://www.apache.org/licenses/LICENSE-2.0
010     *
011     * Unless required by applicable law or agreed to in writing, software
012     * distributed under the License is distributed on an "AS IS" BASIS,
013     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014     * See the License for the specific language governing permissions and
015     * limitations under the License.
016     */
017    package org.apache.camel.util.concurrent;
018    
019    import java.util.concurrent.Callable;
020    import java.util.concurrent.CompletionService;
021    import java.util.concurrent.DelayQueue;
022    import java.util.concurrent.Delayed;
023    import java.util.concurrent.Executor;
024    import java.util.concurrent.Future;
025    import java.util.concurrent.FutureTask;
026    import java.util.concurrent.TimeUnit;
027    import java.util.concurrent.atomic.AtomicInteger;
028    
029    /**
030     * A {@link java.util.concurrent.CompletionService} that orders the completed tasks
031     * in the same order as they where submitted.
032     *
033     * @version $Revision: 16351 $
034     */
035    public class SubmitOrderedCompletionService<V> implements CompletionService<V> {
036        
037        private final Executor executor;
038    
039        // the idea to order the completed task in the same order as they where submitted is to leverage
040        // the delay queue. With the delay queue we can control the order by the getDelay and compareTo methods
041        // where we can order the tasks in the same order as they where submitted.
042        private final DelayQueue<SubmitOrderFutureTask> completionQueue = new DelayQueue<SubmitOrderFutureTask>();
043    
044        // id is the unique id that determines the order in which tasks was submitted (incrementing)
045        private final AtomicInteger id = new AtomicInteger();
046        // index is the index of the next id that should expire and thus be ready to take from the delayed queue
047        private final AtomicInteger index = new AtomicInteger();
048    
049        private class SubmitOrderFutureTask extends FutureTask<V> implements Delayed {
050    
051            // the id this task was assigned
052            private final long id;
053    
054            public SubmitOrderFutureTask(long id, Callable<V> voidCallable) {
055                super(voidCallable);
056                this.id = id;
057            }
058    
059            public SubmitOrderFutureTask(long id, Runnable runnable, V result) {
060                super(runnable, result);
061                this.id = id;
062            }
063    
064            public long getDelay(TimeUnit unit) {
065                // if the answer is 0 then this task is ready to be taken
066                return id - index.get();
067            }
068    
069            @SuppressWarnings("unchecked")
070            public int compareTo(Delayed o) {
071                SubmitOrderFutureTask other = (SubmitOrderFutureTask) o;
072                return (int) (this.id - other.id);
073            }
074    
075            @Override
076            @SuppressWarnings("unchecked")
077            protected void done() {
078                // when we are done add to the completion queue
079                completionQueue.add(this);
080            }
081        }
082    
083        public SubmitOrderedCompletionService(Executor executor) {
084            this.executor = executor;
085        }
086    
087        @SuppressWarnings("unchecked")
088        public Future<V> submit(Callable task) {
089            if (task == null) {
090                throw new IllegalArgumentException("Task must be provided");
091            }
092            SubmitOrderFutureTask f = new SubmitOrderFutureTask(id.incrementAndGet(), task);
093            executor.execute(f);
094            return (Future<V>) f;
095        }
096    
097        @SuppressWarnings("unchecked")
098        public Future<V> submit(Runnable task, Object result) {
099            if (task == null) {
100                throw new IllegalArgumentException("Task must be provided");
101            }
102            SubmitOrderFutureTask f = new SubmitOrderFutureTask(id.incrementAndGet(), task, null);
103            executor.execute(f);
104            return f;
105        }
106    
107        @SuppressWarnings("unchecked")
108        public Future<V> take() throws InterruptedException {
109            index.incrementAndGet();
110            return (Future) completionQueue.take();
111        }
112    
113        @SuppressWarnings("unchecked")
114        public Future<V> poll() {
115            index.incrementAndGet();
116            return (Future) completionQueue.poll();
117        }
118    
119        @SuppressWarnings("unchecked")
120        public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException {
121            index.incrementAndGet();
122            return (Future) completionQueue.poll(timeout, unit);
123        }
124    }