Handling AUTO_READ should not be the responsibility of DefaultChannel… (#8650)

* Handling AUTO_READ should not be the responsibility of DefaultChannelPipeline but the Channel itself.

Motivation:

At the moment we do automatically call read() in the DefaultChannelPipeline when fireChannelReadComplete() / fireChannelActive() is called and the Channel is using auto read. This is nice in terms of sharing code but imho is not the responsibility of the ChannelPipeline implementation but the responsibility of the Channel implementation.

Modifications:

Move handing of auto read from DefaultChannelPipeline to Channel implementations.

Result:

More clear responsibiliy and not depending on implemention details of the ChannelPipeline.

codec-http2/src/main/java/io/netty/handler/codec/http2/Http2MultiplexCodec.java

@@ -850,6 +850,9 @@ public class Http2MultiplexCodec extends Http2FrameCodec {
                 pipeline().fireChannelRegistered();
                 if (isActive()) {
                     pipeline().fireChannelActive();
+                    if (config().isAutoRead()) {
+                        read();
+                    }
                 }
             }
 
@@ -1047,6 +1050,10 @@ public class Http2MultiplexCodec extends Http2FrameCodec {
                 }
                 allocHandle.readComplete();
                 pipeline().fireChannelReadComplete();
+                if (config().isAutoRead()) {
+                    read();
+                }
+
                 // Reading data may result in frames being written (e.g. WINDOW_UPDATE, RST, etc..). If the parent
                 // channel is not currently reading we need to force a flush at the child channel, because we cannot
                 // rely upon flush occurring in channelReadComplete on the parent channel.

transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollChannel.java

@@ -611,6 +611,7 @@ abstract class AbstractEpollChannel extends AbstractChannel implements UnixChann
             // because what happened is what happened.
             if (!wasActive && active) {
                 pipeline().fireChannelActive();
+                readIfIsAutoRead();
             }
 
             // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().

transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollServerChannel.java

@@ -129,6 +129,7 @@ public abstract class AbstractEpollServerChannel extends AbstractEpollChannel im
                 if (exception != null) {
                     pipeline.fireExceptionCaught(exception);
                 }
+                readIfIsAutoRead();
             } finally {
                 epollInFinally(config);
             }

transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java

@@ -736,6 +736,8 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel im
             pipeline.fireExceptionCaught(cause);
             if (close || cause instanceof IOException) {
                 shutdownInput(false);
+            } else {
+                readIfIsAutoRead();
             }
         }
 
@@ -820,6 +822,8 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel im
 
                 if (close) {
                     shutdownInput(false);
+                } else {
+                    readIfIsAutoRead();
                 }
             } catch (Throwable t) {
                 handleReadException(pipeline, byteBuf, t, close, allocHandle);

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java

@@ -499,6 +499,7 @@ public final class EpollDatagramChannel extends AbstractEpollChannel implements
                 if (exception != null) {
                     pipeline.fireExceptionCaught(exception);
                 }
+                readIfIsAutoRead();
             } finally {
                 epollInFinally(config);
             }

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainSocketChannel.java

@@ -185,6 +185,7 @@ public final class EpollDomainSocketChannel extends AbstractEpollStreamChannel i
                 pipeline.fireChannelReadComplete();
                 pipeline.fireExceptionCaught(t);
             } finally {
+                readIfIsAutoRead();
                 epollInFinally(config);
             }
         }

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueChannel.java

@@ -617,6 +617,7 @@ abstract class AbstractKQueueChannel extends AbstractChannel implements UnixChan
             // because what happened is what happened.
             if (!wasActive && active) {
                 pipeline().fireChannelActive();
+                readIfIsAutoRead();
             }
 
             // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueServerChannel.java

@@ -121,6 +121,7 @@ public abstract class AbstractKQueueServerChannel extends AbstractKQueueChannel
                 if (exception != null) {
                     pipeline.fireExceptionCaught(exception);
                 }
+                readIfIsAutoRead();
             } finally {
                 readReadyFinally(config);
             }

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueStreamChannel.java

@@ -562,6 +562,8 @@ public abstract class AbstractKQueueStreamChannel extends AbstractKQueueChannel
 
                 if (close) {
                     shutdownInput(false);
+                } else {
+                    readIfIsAutoRead();
                 }
             } catch (Throwable t) {
                 handleReadException(pipeline, byteBuf, t, close, allocHandle);
@@ -586,6 +588,8 @@ public abstract class AbstractKQueueStreamChannel extends AbstractKQueueChannel
                 pipeline.fireExceptionCaught(cause);
                 if (close || cause instanceof IOException) {
                     shutdownInput(false);
+                } else {
+                    readIfIsAutoRead();
                 }
             }
         }

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDatagramChannel.java

@@ -464,6 +464,8 @@ public final class KQueueDatagramChannel extends AbstractKQueueChannel implement
 
                 if (exception != null) {
                     pipeline.fireExceptionCaught(exception);
+                } else {
+                    readIfIsAutoRead();
                 }
             } finally {
                 readReadyFinally(config);

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainSocketChannel.java

@@ -180,6 +180,7 @@ public final class KQueueDomainSocketChannel extends AbstractKQueueStreamChannel
                 pipeline.fireChannelReadComplete();
                 pipeline.fireExceptionCaught(t);
             } finally {
+                readIfIsAutoRead();
                 readReadyFinally(config);
             }
         }

transport/src/main/java/io/netty/channel/AbstractChannel.java

@@ -421,6 +421,12 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
         return pipeline.voidPromise();
     }
 
+    protected final void readIfIsAutoRead() {
+        if (config().isAutoRead()) {
+            read();
+        }
+    }
+
     /**
      * {@link Unsafe} implementation which sub-classes must extend and use.
      */
@@ -520,13 +526,8 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
                 if (isActive()) {
                     if (firstRegistration) {
                         pipeline.fireChannelActive();
-                    } else if (config().isAutoRead()) {
-                        // This channel was registered before and autoRead() is set. This means we need to begin read
-                        // again so that we process inbound data.
-                        //
-                        // See https://github.com/netty/netty/issues/4805
-                        beginRead();
                     }
+                    readIfIsAutoRead();
                 }
             } catch (Throwable t) {
                 // Close the channel directly to avoid FD leak.
@@ -571,6 +572,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
                     @Override
                     public void run() {
                         pipeline.fireChannelActive();
+                        readIfIsAutoRead();
                     }
                 });
             }

transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java

@@ -1420,8 +1420,6 @@ public class DefaultChannelPipeline implements ChannelPipeline {
         @Override
         public void channelActive(ChannelHandlerContext ctx) throws Exception {
             ctx.fireChannelActive();
-
-            readIfIsAutoRead();
         }
 
         @Override
@@ -1437,14 +1435,6 @@ public class DefaultChannelPipeline implements ChannelPipeline {
         @Override
         public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
             ctx.fireChannelReadComplete();
-
-            readIfIsAutoRead();
-        }
-
-        private void readIfIsAutoRead() {
-            if (channel.config().isAutoRead()) {
-                channel.read();
-            }
         }
 
         @Override

transport/src/main/java/io/netty/channel/embedded/EmbeddedChannel.java

@@ -365,6 +365,7 @@ public class EmbeddedChannel extends AbstractChannel {
     private ChannelFuture flushInbound(boolean recordException, ChannelPromise promise) {
       if (checkOpen(recordException)) {
           pipeline().fireChannelReadComplete();
+          readIfIsAutoRead();
           runPendingTasks();
       }
 

transport/src/main/java/io/netty/channel/local/LocalChannel.java

@@ -188,6 +188,7 @@ public class LocalChannel extends AbstractChannel {
                     // connectPromise may be set to null if doClose() was called in the meantime.
                     if (promise != null && promise.trySuccess()) {
                         peer.pipeline().fireChannelActive();
+                        peer.readIfIsAutoRead();
                     }
                 }
             });
@@ -305,6 +306,7 @@ public class LocalChannel extends AbstractChannel {
         } while (handle.continueReading());
 
         pipeline.fireChannelReadComplete();
+        readIfIsAutoRead();
     }
 
     @Override

transport/src/main/java/io/netty/channel/local/LocalServerChannel.java

@@ -158,6 +158,7 @@ public class LocalServerChannel extends AbstractServerChannel {
         } while (handle.continueReading());
 
         pipeline.fireChannelReadComplete();
+        readIfIsAutoRead();
     }
 
     /**

transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java

@@ -125,6 +125,8 @@ public abstract class AbstractNioByteChannel extends AbstractNioChannel {
             pipeline.fireExceptionCaught(cause);
             if (close || cause instanceof IOException) {
                 closeOnRead(pipeline);
+            } else {
+                readIfIsAutoRead();
             }
         }
 
@@ -169,6 +171,8 @@ public abstract class AbstractNioByteChannel extends AbstractNioChannel {
 
                 if (close) {
                     closeOnRead(pipeline);
+                } else {
+                    readIfIsAutoRead();
                 }
             } catch (Throwable t) {
                 handleReadException(pipeline, byteBuf, t, close, allocHandle);

transport/src/main/java/io/netty/channel/nio/AbstractNioChannel.java

@@ -309,6 +309,7 @@ public abstract class AbstractNioChannel extends AbstractChannel {
             // because what happened is what happened.
             if (!wasActive && active) {
                 pipeline().fireChannelActive();
+                readIfIsAutoRead();
             }
 
             // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().

transport/src/main/java/io/netty/channel/nio/AbstractNioMessageChannel.java

@@ -107,6 +107,8 @@ public abstract class AbstractNioMessageChannel extends AbstractNioChannel {
                     if (isOpen()) {
                         close(voidPromise());
                     }
+                } else {
+                    readIfIsAutoRead();
                 }
             } finally {
                 // Check if there is a readPending which was not processed yet.

transport/src/test/java/io/netty/channel/DefaultChannelPipelineTailTest.java

@@ -357,6 +357,7 @@ public class DefaultChannelPipelineTailTest {
                 if (!active) {
                     active = true;
                     pipeline().fireChannelActive();
+                    readIfIsAutoRead();
                 }
 
                 promise.setSuccess();