Allow to use EmbeddedChannel.schedule*(...)

Motivation:

At the moment when EmbeddedChannel is used and a ChannelHandler tries to schedule and task it will throw an UnsupportedOperationException. This makes it impossible to test these handlers or even reuse them with EmbeddedChannel.

Modifications:

- Factor out reusable scheduling code into AbstractSchedulingEventExecutor
- Let EmbeddedEventLoop and SingleThreadEventExecutor extend AbstractSchedulingEventExecutor
- add EmbbededChannel.runScheduledPendingTasks() which allows to run all scheduled tasks that are ready

Result:

Embeddedchannel is now usable even with ChannelHandler that try to schedule tasks.

common/src/main/java/io/netty/util/concurrent/AbstractScheduledEventExecutor.java

@@ -0,0 +1,257 @@
+/*
+ * Copyright 2015 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.util.concurrent;
+
+import io.netty.util.internal.CallableEventExecutorAdapter;
+import io.netty.util.internal.ObjectUtil;
+import io.netty.util.internal.RunnableEventExecutorAdapter;
+
+import java.util.Iterator;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Abstract base class for {@link EventExecutor}s that want to support scheduling.
+ */
+public abstract class AbstractScheduledEventExecutor extends AbstractEventExecutor {
+
+    Queue<ScheduledFutureTask<?>> scheduledTaskQueue;
+
+    protected AbstractScheduledEventExecutor() {
+    }
+
+    protected AbstractScheduledEventExecutor(EventExecutorGroup parent) {
+        super(parent);
+    }
+
+    protected static long nanoTime() {
+        return ScheduledFutureTask.nanoTime();
+    }
+
+    Queue<ScheduledFutureTask<?>> scheduledTaskQueue() {
+        if (scheduledTaskQueue == null) {
+            scheduledTaskQueue = new PriorityQueue<ScheduledFutureTask<?>>();
+        }
+        return scheduledTaskQueue;
+    }
+
+    private static  boolean isNullOrEmpty(Queue<ScheduledFutureTask<?>> queue) {
+        return queue == null || queue.isEmpty();
+    }
+
+    /**
+     * Cancel all scheduled tasks.
+     *
+     * This method MUST be called only when {@link #inEventLoop()} is {@code true}.
+     */
+    protected void cancelScheduledTasks() {
+        assert inEventLoop();
+        Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
+        if (isNullOrEmpty(scheduledTaskQueue)) {
+            return;
+        }
+
+        final ScheduledFutureTask<?>[] scheduledTasks =
+                scheduledTaskQueue.toArray(new ScheduledFutureTask<?>[scheduledTaskQueue.size()]);
+
+        for (ScheduledFutureTask<?> task: scheduledTasks) {
+            task.cancel(false);
+        }
+
+        scheduledTaskQueue.clear();
+    }
+
+    /**
+     * @see {@link #pollScheduledTask(long)}
+     */
+    protected final Runnable pollScheduledTask() {
+        return pollScheduledTask(nanoTime());
+    }
+
+    /**
+     * Return the {@link Runnable} which is ready to be executed with the given {@code nanoTime}.
+     * You should use {@link #nanoTime()} to retrieve the the correct {@code nanoTime}.
+     */
+    protected final Runnable pollScheduledTask(long nanoTime) {
+        assert inEventLoop();
+
+        Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
+        ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
+        if (scheduledTask == null) {
+            return null;
+        }
+
+        if (scheduledTask.deadlineNanos() <= nanoTime) {
+            scheduledTaskQueue.remove();
+            return scheduledTask;
+        }
+        return null;
+    }
+
+    /**
+     * Return the nanoseconds when the next scheduled task is ready to be run or {@code -1} if no task is scheduled.
+     */
+    protected final long nextScheduledTaskNano() {
+        Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
+        ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
+        if (scheduledTask == null) {
+            return -1;
+        }
+        return Math.max(0, scheduledTask.deadlineNanos() - nanoTime());
+    }
+
+    final ScheduledFutureTask<?> peekScheduledTask() {
+        Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
+        if (scheduledTaskQueue == null) {
+            return null;
+        }
+        return scheduledTaskQueue.peek();
+    }
+
+    /**
+     * Returns {@code true} if a scheduled task is ready for processing.
+     */
+    protected final boolean hasScheduledTasks() {
+        Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
+        ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
+        return scheduledTask != null && scheduledTask.deadlineNanos() <= nanoTime();
+    }
+
+    @Override
+    public  ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+        ObjectUtil.checkNotNull(command, "command");
+        ObjectUtil.checkNotNull(unit, "unit");
+        if (delay < 0) {
+            throw new IllegalArgumentException(
+                    String.format("delay: %d (expected: >= 0)", delay));
+        }
+        return schedule(new ScheduledFutureTask<Void>(
+                this, toCallable(command), ScheduledFutureTask.deadlineNanos(unit.toNanos(delay))));
+    }
+
+    @Override
+    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
+        ObjectUtil.checkNotNull(callable, "callable");
+        ObjectUtil.checkNotNull(unit, "unit");
+        if (delay < 0) {
+            throw new IllegalArgumentException(
+                    String.format("delay: %d (expected: >= 0)", delay));
+        }
+        return schedule(new ScheduledFutureTask<V>(
+                this, callable, ScheduledFutureTask.deadlineNanos(unit.toNanos(delay))));
+    }
+
+    @Override
+    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
+        ObjectUtil.checkNotNull(command, "command");
+        ObjectUtil.checkNotNull(unit, "unit");
+        if (initialDelay < 0) {
+            throw new IllegalArgumentException(
+                    String.format("initialDelay: %d (expected: >= 0)", initialDelay));
+        }
+        if (period <= 0) {
+            throw new IllegalArgumentException(
+                    String.format("period: %d (expected: > 0)", period));
+        }
+
+        return schedule(new ScheduledFutureTask<Void>(
+                this, toCallable(command),
+                ScheduledFutureTask.deadlineNanos(unit.toNanos(initialDelay)), unit.toNanos(period)));
+    }
+
+    @Override
+    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
+        ObjectUtil.checkNotNull(command, "command");
+        ObjectUtil.checkNotNull(unit, "unit");
+        if (initialDelay < 0) {
+            throw new IllegalArgumentException(
+                    String.format("initialDelay: %d (expected: >= 0)", initialDelay));
+        }
+        if (delay <= 0) {
+            throw new IllegalArgumentException(
+                    String.format("delay: %d (expected: > 0)", delay));
+        }
+
+        return schedule(new ScheduledFutureTask<Void>(
+                this, toCallable(command),
+                ScheduledFutureTask.deadlineNanos(unit.toNanos(initialDelay)), -unit.toNanos(delay)));
+    }
+
+    <V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) {
+        if (inEventLoop()) {
+            scheduledTaskQueue().add(task);
+        } else {
+            execute(new Runnable() {
+                @Override
+                public void run() {
+                    scheduledTaskQueue().add(task);
+                }
+            });
+        }
+
+        return task;
+    }
+
+    void purgeCancelledScheduledTasks() {
+        Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
+        if (isNullOrEmpty(scheduledTaskQueue)) {
+            return;
+        }
+        Iterator<ScheduledFutureTask<?>> i = scheduledTaskQueue.iterator();
+        while (i.hasNext()) {
+            ScheduledFutureTask<?> task = i.next();
+            if (task.isCancelled()) {
+                i.remove();
+            }
+        }
+    }
+
+    private static Callable<Void> toCallable(final Runnable command) {
+        if (command instanceof RunnableEventExecutorAdapter) {
+            return new RunnableToCallableAdapter((RunnableEventExecutorAdapter) command);
+        } else {
+            return Executors.callable(command, null);
+        }
+    }
+
+    private static class RunnableToCallableAdapter implements CallableEventExecutorAdapter<Void> {
+
+        final RunnableEventExecutorAdapter runnable;
+
+        RunnableToCallableAdapter(RunnableEventExecutorAdapter runnable) {
+            this.runnable = runnable;
+        }
+
+        @Override
+        public EventExecutor executor() {
+            return runnable.executor();
+        }
+
+        @Override
+        public Callable<Void> unwrap() {
+            return null;
+        }
+
+        @Override
+        public Void call() throws Exception {
+            runnable.run();
+            return null;
+        }
+    }
+}