WebSVN - OpenConcerto - Diff - Rev 156 and 177 - /trunk/OpenConcerto/src/org/openconcerto/utils/SleepingQueue.java

 import java.util.Collection;
 import java.util.Deque;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Future;
 import java.util.concurrent.FutureTask;
+import java.util.concurrent.RunnableFuture;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 import net.jcip.annotations.GuardedBy;
 /**
- * A queue that can be put to sleep. Submitted runnables are converted to FutureTask, that can later
+ * A queue that can be put to sleep. Submitted runnables are converted to RunnableFuture, that can
- * be cancelled.
+ * later be cancelled.
+ *
  * @author Sylvain
  */
-public class SleepingQueue {
+public class SleepingQueue implements Executor {
     public static enum RunningState {
         NEW, RUNNING, WILL_DIE, DYING, DEAD
+    }
             @Override
             public void run() {
                 final boolean isDone;
                 final RunningState runningState;
-                final FutureTask<?> beingRun;
+                final RunnableFuture<?> beingRun;
                 final SleepingQueue q = lethalFuture.getQueue();
                 synchronized (q) {
                     runningState = q.getRunningState();
                     beingRun = q.getBeingRun();
                     isDone = lethalFuture.isDone();
     @GuardedBy("this")
     private RunningState state;
     private final PropertyChangeSupport support;
     @GuardedBy("this")
-    private FutureTask<?> beingRun;
+    private RunnableFuture<?> beingRun;
     private final SingleThreadedExecutor tasksQueue;
     @GuardedBy("this")
     private boolean canceling;
     @GuardedBy("this")
-    private IPredicate<FutureTask<?>> cancelPredicate;
+    private IPredicate<? super RunnableFuture<?>> cancelPredicate;
     public SleepingQueue() {
         this(SleepingQueue.class.getName() + System.currentTimeMillis());
+    }
      */
     protected void customizeThread(Thread thr) {
         thr.setPriority(Thread.MIN_PRIORITY);
+    }
-    protected final <T> FutureTask<T> newTaskFor(final Runnable task) {
+    protected final <T> RunnableFuture<T> newTaskFor(final Runnable task) {
         return this.newTaskFor(task, null);
+    }
-    protected <T> FutureTask<T> newTaskFor(final Runnable task, T value) {
+    protected <T> RunnableFuture<T> newTaskFor(final Runnable task, T value) {
         return new IFutureTask<T>(task, value, " for {" + this.name + "}");
+    }
-    public final FutureTask<?> put(Runnable workRunnable) {
+    public final RunnableFuture<?> put(Runnable workRunnable) {
+        /*
-        // otherwise if passing a FutureTask, it will itself be wrapped in another FutureTask. The
+         * Otherwise if passing a RunnableFuture, it will itself be wrapped in another
-        // outer FutureTask will call the inner one's run(), which just record any exception. So the
+         * RunnableFuture. The outer RunnableFuture will call the inner one's run(), which just
-        // outer one's get() won't throw it and the exception will effectively be swallowed.
+         * record any exception. So the outer one's get() won't throw it and the exception will
+         * effectively be swallowed.
+         */
-        final FutureTask<?> t;
+        final RunnableFuture<?> t;
-        if (workRunnable instanceof FutureTask) {
+        if (workRunnable instanceof RunnableFuture) {
-            t = ((FutureTask<?>) workRunnable);
+            t = ((RunnableFuture<?>) workRunnable);
         } else {
             t = this.newTaskFor(workRunnable);
+        }
-        return this.execute(t);
+        return this.add(t);
+    }
-    public final <F extends FutureTask<?>> F execute(F t) {
-        if (this.shallAdd(t)) {
-            this.add(t);
+    @Override
-            return t;
+    public final void execute(Runnable command) {
-        } else
-            return null;
+        this.put(command);
+    }
-    private void add(FutureTask<?> t) {
+    public final <F extends RunnableFuture<?>> F add(F t) {
+        if (this.shallAdd(t)) {
-        // no need to synchronize, if die() is called after our test, t won't be executed anyway
+            // no need to synchronize, if die() is called after our test, t won't be executed anyway
-        if (this.dieCalled())
+            if (this.dieCalled())
-            throw new IllegalStateException("Already dead, cannot exec " + t);
+                throw new IllegalStateException("Already dead, cannot exec " + t);
-        this.tasksQueue.put(t);
+            this.tasksQueue.put(t);
+            return t;
+        } else {
+            return null;
+        }
+    }
-    private final boolean shallAdd(FutureTask<?> runnable) {
+    private final boolean shallAdd(RunnableFuture<?> runnable) {
         if (runnable == null)
             throw new NullPointerException("null runnable");
         try {
             this.willPut(runnable);
             return true;
      * Give subclass the ability to reject runnables.
+     *
      * @param r the runnable that is being added.
      * @throws InterruptedException if r should not be added to this queue.
      */
-    protected void willPut(FutureTask<?> r) throws InterruptedException {
+    protected void willPut(RunnableFuture<?> r) throws InterruptedException {
+    }
     /**
      * An exception was thrown by a task. This implementation uses
      * {@link Thread#getUncaughtExceptionHandler()} or
     /**
      * Cancel only tasks for which pred is <code>true</code>.
+     *
      * @param pred a predicate to know which tasks to cancel.
      */
-    protected final void cancel(final IPredicate<FutureTask<?>> pred) {
+    protected final void cancel(final IPredicate<? super RunnableFuture<?>> pred) {
-        this.tasksDo(new IClosure<Collection<FutureTask<?>>>() {
+        this.tasksDo(new IClosure<Collection<RunnableFuture<?>>>() {
             @Override
-            public void executeChecked(Collection<FutureTask<?>> tasks) {
+            public void executeChecked(Collection<RunnableFuture<?>> tasks) {
                 cancel(pred, tasks);
+            }
         });
+    }
-    private final void cancel(IPredicate<FutureTask<?>> pred, Collection<FutureTask<?>> tasks) {
+    private final void cancel(IPredicate<? super RunnableFuture<?>> pred, Collection<RunnableFuture<?>> tasks) {
         try {
             synchronized (this) {
                 this.canceling = true;
                 this.cancelPredicate = pred;
                 this.cancelCheck(this.getBeingRun());
+            }
-            for (final FutureTask<?> t : tasks) {
+            for (final RunnableFuture<?> t : tasks) {
                 this.cancelCheck(t);
+            }
         } finally {
             synchronized (this) {
                 this.canceling = false;
                 this.cancelPredicate = null;
+            }
+        }
+    }
-    public final void tasksDo(IClosure<? super Deque<FutureTask<?>>> c) {
+    public final void tasksDo(IClosure<? super Deque<RunnableFuture<?>>> c) {
         this.tasksQueue.itemsDo(c);
+    }
-    private void cancelCheck(FutureTask<?> t) {
+    private void cancelCheck(RunnableFuture<?> t) {
         if (t != null)
             synchronized (this) {
                 if (this.canceling && (this.cancelPredicate == null || this.cancelPredicate.evaluateChecked(t)))
                     t.cancel(true);
+            }
+    }
-    private void setBeingRun(final FutureTask<?> beingRun) {
+    private void setBeingRun(final RunnableFuture<?> beingRun) {
         final Future<?> old;
         synchronized (this) {
             old = this.beingRun;
             this.beingRun = beingRun;
+        }
         this.support.firePropertyChange("beingRun", old, beingRun);
+    }
-    public final synchronized FutureTask<?> getBeingRun() {
+    public final synchronized RunnableFuture<?> getBeingRun() {
         return this.beingRun;
+    }
     public boolean isSleeping() {
         return this.tasksQueue.isSleeping();
                         SleepingQueue.this.tasksQueue.setSleeping(true);
+                }
+            }
         });
         // die as soon as possible not after all currently queued tasks
-        this.tasksQueue.itemsDo(new IClosure<Deque<FutureTask<?>>>() {
+        this.tasksQueue.itemsDo(new IClosure<Deque<RunnableFuture<?>>>() {
             @Override
-            public void executeChecked(Deque<FutureTask<?>> input) {
+            public void executeChecked(Deque<RunnableFuture<?>> input) {
                 // since we cancel the current task, we might as well remove all of them since they
                 // might depend on the cancelled one
                 // cancel removed tasks so that callers of get() don't wait forever
-                for (final FutureTask<?> ft : input) {
+                for (final RunnableFuture<?> ft : input) {
                     // by definition tasks in the queue aren't executing, so interrupt parameter is
                     // useless. On the other hand cancel() might return false if already cancelled.
                     ft.cancel(false);
+                }
                 input.clear();
                 input.addFirst(res);
                 // die as soon as possible, even if there's a long task already running
-                final FutureTask<?> beingRun = getBeingRun();
+                final RunnableFuture<?> beingRun = getBeingRun();
                 // since we hold the lock on items
                 assert beingRun != res : "beingRun: " + beingRun + " ; res: " + res;
                 if (beingRun != null)
                     beingRun.cancel(true);
+            }
     public void rmPropertyChangeListener(PropertyChangeListener l) {
         this.support.removePropertyChangeListener(l);
+    }
-    private final class SingleThreadedExecutor extends DropperQueue<FutureTask<?>> {
+    private final class SingleThreadedExecutor extends DropperQueue<RunnableFuture<?>> {
         private SingleThreadedExecutor() {
             super(SleepingQueue.this.name + System.currentTimeMillis());
+        }
         @Override
-        protected void process(FutureTask<?> task) {
+        protected void process(RunnableFuture<?> task) {
             if (!task.isDone()) {
                 /*
                  * From ThreadPoolExecutor : Track execution state to ensure that afterExecute is
                  * called only if task completed or threw exception. Otherwise, the caught runtime
                  * exception will have been thrown by afterExecute itself, in which case we don't
                     // this queue)
+                }
+            }
+        }
-        protected void beforeExecute(final FutureTask<?> f) {
+        protected void beforeExecute(final RunnableFuture<?> f) {
             cancelCheck(f);
             setBeingRun(f);
+        }
-        protected void afterExecute(final FutureTask<?> f, final Throwable t) {
+        protected void afterExecute(final RunnableFuture<?> f, final Throwable t) {
             setBeingRun(null);
             try {
                 f.get();
             } catch (CancellationException e) {

Dépôt officiel du code source de l'ERP OpenConcerto

svn://code.openconcerto.org/openconcerto

(root)/trunk/OpenConcerto/src/org/openconcerto/utils/SleepingQueue.java @ 180 - Rev 156 → 177