Add default methods to EventExecutor / EventExecutorGroup / EventLoop…

… / EventLoopGroup (#11649)

Motivation:

We can remove some classes and duplication if we add default methods

Modifications:

- Add default methods to EventExecutor / EventExecutorGroup / EventLoop / EventLoopGroup
- Remove code duplication
- Remove AbstractEventExecutorGroup as it is not needed anymore

Result:

Cleanup and removal of code-duplication. Also makes it easier for people to implement their custom executors / groups

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

@@ -18,14 +18,9 @@
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
 import java.util.concurrent.AbstractExecutorService;
 import java.util.concurrent.Callable;
 import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 
 /**
  * Abstract base class for {@link EventExecutor} implementations.
@@ -35,46 +30,8 @@ public abstract class AbstractEventExecutor extends AbstractExecutorService impl
     static final long DEFAULT_SHUTDOWN_QUIET_PERIOD = 2;
     static final long DEFAULT_SHUTDOWN_TIMEOUT = 15;
 
-    private final Collection<EventExecutor> selfCollection = Collections.singleton(this);
     private final Future<?> successfulVoidFuture = DefaultPromise.newSuccessfulPromise(this, null).asFuture();
 
-    @Override
-    public EventExecutor next() {
-        return this;
-    }
-
-    @Override
-    public final boolean inEventLoop() {
-        return inEventLoop(Thread.currentThread());
-    }
-
-    @Override
-    public final Iterator<EventExecutor> iterator() {
-        return selfCollection.iterator();
-    }
-
-    @Override
-    public final Future<?> shutdownGracefully() {
-        return shutdownGracefully(DEFAULT_SHUTDOWN_QUIET_PERIOD, DEFAULT_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS);
-    }
-
-    /**
-     * @deprecated {@link #shutdownGracefully(long, long, TimeUnit)} or {@link #shutdownGracefully()} instead.
-     */
-    @Override
-    @Deprecated
-    public abstract void shutdown();
-
-    /**
-     * @deprecated {@link #shutdownGracefully(long, long, TimeUnit)} or {@link #shutdownGracefully()} instead.
-     */
-    @Override
-    @Deprecated
-    public List<Runnable> shutdownNow() {
-        shutdown();
-        return Collections.emptyList();
-    }
-
     @Override
     public <V> Future<V> newSucceededFuture(V result) {
         if (result == null) {

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

@@ -15,6 +15,9 @@
  */
 package io.netty.util.concurrent;
 
+import java.util.Collections;
+import java.util.Iterator;
+
 /**
  * The {@link EventExecutor} is a special {@link EventExecutorGroup} which comes
  * with some handy methods to see if a {@link Thread} is executed in a event loop.
@@ -28,12 +31,21 @@ public interface EventExecutor extends EventExecutorGroup {
      * Returns a reference to itself.
      */
     @Override
-    EventExecutor next();
+    default EventExecutor next() {
+        return this;
+    }
+
+    @Override
+    default Iterator<EventExecutor> iterator() {
+        return Collections.singleton(this).iterator();
+    }
 
     /**
      * Calls {@link #inEventLoop(Thread)} with {@link Thread#currentThread()} as argument
      */
-    boolean inEventLoop();
+    default boolean inEventLoop() {
+        return inEventLoop(Thread.currentThread());
+    }
 
     /**
      * Return {@code true} if the given {@link Thread} is executed in the event loop,

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

@@ -15,11 +15,18 @@
  */
 package io.netty.util.concurrent;
 
+import java.util.Collection;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static io.netty.util.concurrent.AbstractEventExecutor.DEFAULT_SHUTDOWN_QUIET_PERIOD;
+import static io.netty.util.concurrent.AbstractEventExecutor.DEFAULT_SHUTDOWN_TIMEOUT;
 
 /**
  * The {@link EventExecutorGroup} is responsible for providing the {@link EventExecutor}'s to use
@@ -40,7 +47,9 @@ public interface EventExecutorGroup extends ScheduledExecutorService, Iterable<E
      *
      * @return the {@link #terminationFuture()}
      */
-    Future<?> shutdownGracefully();
+    default Future<?> shutdownGracefully() {
+        return shutdownGracefully(DEFAULT_SHUTDOWN_QUIET_PERIOD, DEFAULT_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS);
+    }
 
     /**
      * Signals this executor that the caller wants the executor to be shut down.  Once this method is called,
@@ -76,7 +85,10 @@ public interface EventExecutorGroup extends ScheduledExecutorService, Iterable<E
      */
     @Override
     @Deprecated
-    List<Runnable> shutdownNow();
+    default List<Runnable> shutdownNow() {
+        shutdown();
+        return Collections.emptyList();
+    }
 
     /**
      * Returns one of the {@link EventExecutor}s managed by this {@link EventExecutorGroup}.
@@ -87,23 +99,65 @@ public interface EventExecutorGroup extends ScheduledExecutorService, Iterable<E
     Iterator<EventExecutor> iterator();
 
     @Override
-    Future<?> submit(Runnable task);
+    default Future<?> submit(Runnable task) {
+        return next().submit(task);
+    }
+
+    @Override
+    default <T> Future<T> submit(Runnable task, T result) {
+        return next().submit(task, result);
+    }
+
+    @Override
+    default <T> Future<T> submit(Callable<T> task) {
+        return next().submit(task);
+    }
+
+    @Override
+    default ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+        return next().schedule(command, delay, unit);
+    }
+
+    @Override
+    default <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
+        return next().schedule(callable, delay, unit);
+    }
+
+    @Override
+    default ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
+        return next().scheduleAtFixedRate(command, initialDelay, period, unit);
+    }
 
     @Override
-    <T> Future<T> submit(Runnable task, T result);
+    default ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
+        return next().scheduleWithFixedDelay(command, initialDelay, delay, unit);
+    }
 
     @Override
-    <T> Future<T> submit(Callable<T> task);
+    default <T> List<java.util.concurrent.Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
+            throws InterruptedException {
+        return next().invokeAll(tasks);
+    }
 
     @Override
-    ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
+    default <T> List<java.util.concurrent.Future<T>> invokeAll(
+            Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
+        return next().invokeAll(tasks, timeout, unit);
+    }
 
     @Override
-    <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
+    default <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
+        return next().invokeAny(tasks);
+    }
 
     @Override
-    ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
+    default <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+            throws InterruptedException, ExecutionException, TimeoutException {
+        return next().invokeAny(tasks, timeout, unit);
+    }
 
     @Override
-    ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
+    default void execute(Runnable command) {
+        next().execute(command);
+    }
 }

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

@@ -25,7 +25,7 @@
 public final class ImmediateExecutor implements Executor {
     public static final ImmediateExecutor INSTANCE = new ImmediateExecutor();
 
-    private  ImmediateExecutor() {
+    private ImmediateExecutor() {
         // use static instance
     }
 

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

@@ -33,12 +33,12 @@
  * {@link EventExecutorGroup} implementation that handles their tasks with multiple threads at
  * the same time.
  */
-public class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
+public class MultithreadEventExecutorGroup implements EventExecutorGroup {
 
     private final EventExecutor[] children;
     private final List<EventExecutor> readonlyChildren;
     private final AtomicInteger terminatedChildren = new AtomicInteger();
-    private final Promise<?> terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE);
+    private final Promise<?> terminationFuture = GlobalEventExecutor.INSTANCE.newPromise();
     private final boolean powerOfTwo;
 
     /**