Repository: netty/netty
Issue: shutdownOutput for EpollDomainSocketChannel
Labels: duplicate

I see a protected `shutdownOutput0` method in `AbstractEpollStreamChannel` but there is no public `shutdownOutput` in `EpollDomainSocketChannel`.

I suppose `shutdownOutput` implementation from `EpollSocketChannel` can be moved to  `AbstractEpollStreamChannel`:

``` java
    @Override
    public ChannelFuture shutdownOutput(final ChannelPromise promise) {
        Executor closeExecutor = ((EpollSocketChannelUnsafe) unsafe()).prepareToClose();
        if (closeExecutor != null) {
            closeExecutor.execute(new OneTimeTask() {
                @Override
                public void run() {
                    shutdownOutput0(promise);
                }
            });
        } else {
            EventLoop loop = eventLoop();
            if (loop.inEventLoop()) {
                shutdownOutput0(promise);
            } else {
                loop.execute(new OneTimeTask() {
                    @Override
                    public void run() {
                        shutdownOutput0(promise);
                    }
                });
            }
        }
        return promise;
    }
```

I expect this change to help with https://github.com/docker-java/docker-java/pull/472. If you consider this to be a valid solution then I can create a pull request.


Issue comments:
@rtimush yes!

Duplicates https://github.com/netty/netty/issues/4881

Lets close this in favor of #4881 (reported first).


Pull request review comments:
do we need this change ?

future -> promise

do we need this ?

do we need this ?

found another way, so don't need `public` anymore, fixed

done

no, fixed

no, fixed


Patch that resolved the issue:
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
index 612afce5efb..78aa2ef8009 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
@@ -29,6 +29,7 @@
 import io.netty.channel.DefaultFileRegion;
 import io.netty.channel.EventLoop;
 import io.netty.channel.RecvByteBufAllocator;
+import io.netty.channel.socket.DuplexChannel;
 import io.netty.channel.unix.FileDescriptor;
 import io.netty.channel.unix.Socket;
 import io.netty.util.internal.EmptyArrays;
@@ -44,13 +45,14 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.util.Queue;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
 import static io.netty.channel.unix.FileDescriptor.pipe;
 import static io.netty.util.internal.ObjectUtil.checkNotNull;
 
-public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel {
+public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel implements DuplexChannel {
 
     private static final String EXPECTED_TYPES =
             " (expected: " + StringUtil.simpleClassName(ByteBuf.class) + ", " +
@@ -537,6 +539,47 @@ protected void shutdownOutput0(final ChannelPromise promise) {
         }
     }
 
+    @Override
+    public boolean isInputShutdown() {
+        return fd().isInputShutdown();
+    }
+
+    @Override
+    public boolean isOutputShutdown() {
+        return fd().isOutputShutdown();
+    }
+
+    @Override
+    public ChannelFuture shutdownOutput() {
+        return shutdownOutput(newPromise());
+    }
+
+    @Override
+    public ChannelFuture shutdownOutput(final ChannelPromise promise) {
+        Executor closeExecutor = ((EpollStreamUnsafe) unsafe()).prepareToClose();
+        if (closeExecutor != null) {
+            closeExecutor.execute(new OneTimeTask() {
+                @Override
+                public void run() {
+                    shutdownOutput0(promise);
+                }
+            });
+        } else {
+            EventLoop loop = eventLoop();
+            if (loop.inEventLoop()) {
+                shutdownOutput0(promise);
+            } else {
+                loop.execute(new OneTimeTask() {
+                    @Override
+                    public void run() {
+                        shutdownOutput0(promise);
+                    }
+                });
+            }
+        }
+        return promise;
+    }
+
     @Override
     protected void doClose() throws Exception {
         try {
@@ -610,6 +653,13 @@ private void safeClosePipe(FileDescriptor fd) {
     }
 
     class EpollStreamUnsafe extends AbstractEpollUnsafe {
+
+        // Overridden here just to be able to access this method from AbstractEpollStreamChannel
+        @Override
+        protected Executor prepareToClose() {
+            return super.prepareToClose();
+        }
+
         private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Throwable cause, boolean close) {
             if (byteBuf != null) {
                 if (byteBuf.isReadable()) {
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java
index 4fe6b8a9f45..d0c90ea231c 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java
@@ -16,15 +16,11 @@
 package io.netty.channel.epoll;
 
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelPromise;
-import io.netty.channel.EventLoop;
 import io.netty.channel.socket.ServerSocketChannel;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.unix.FileDescriptor;
 import io.netty.channel.unix.Socket;
 import io.netty.util.concurrent.GlobalEventExecutor;
-import io.netty.util.internal.OneTimeTask;
 import io.netty.util.internal.PlatformDependent;
 
 import java.net.InetAddress;
@@ -143,47 +139,6 @@ public EpollSocketChannelConfig config() {
         return config;
     }
 
-    @Override
-    public boolean isInputShutdown() {
-        return fd().isInputShutdown();
-    }
-
-    @Override
-    public boolean isOutputShutdown() {
-        return fd().isOutputShutdown();
-    }
-
-    @Override
-    public ChannelFuture shutdownOutput() {
-        return shutdownOutput(newPromise());
-    }
-
-    @Override
-    public ChannelFuture shutdownOutput(final ChannelPromise promise) {
-        Executor closeExecutor = ((EpollSocketChannelUnsafe) unsafe()).prepareToClose();
-        if (closeExecutor != null) {
-            closeExecutor.execute(new OneTimeTask() {
-                @Override
-                public void run() {
-                    shutdownOutput0(promise);
-                }
-            });
-        } else {
-            EventLoop loop = eventLoop();
-            if (loop.inEventLoop()) {
-                shutdownOutput0(promise);
-            } else {
-                loop.execute(new OneTimeTask() {
-                    @Override
-                    public void run() {
-                        shutdownOutput0(promise);
-                    }
-                });
-            }
-        }
-        return promise;
-    }
-
     @Override
     public ServerSocketChannel parent() {
         return (ServerSocketChannel) super.parent();
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/unix/DomainSocketChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/unix/DomainSocketChannel.java
index f8163d91342..9a6c08b584b 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/unix/DomainSocketChannel.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/unix/DomainSocketChannel.java
@@ -15,11 +15,13 @@
  */
 package io.netty.channel.unix;
 
+import io.netty.channel.socket.DuplexChannel;
+
 /**
  * A {@link UnixChannel} that supports communication via
  * <a href="http://en.wikipedia.org/wiki/Unix_domain_socket">Unix Domain Socket</a>.
  */
-public interface DomainSocketChannel extends UnixChannel {
+public interface DomainSocketChannel extends UnixChannel, DuplexChannel {
     @Override
     DomainSocketAddress remoteAddress();
 
diff --git a/transport/src/main/java/io/netty/channel/socket/DuplexChannel.java b/transport/src/main/java/io/netty/channel/socket/DuplexChannel.java
new file mode 100644
index 00000000000..d34ec36bff1
--- /dev/null
+++ b/transport/src/main/java/io/netty/channel/socket/DuplexChannel.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2016 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.channel.socket;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelPromise;
+
+import java.net.Socket;
+
+/**
+ * A duplex {@link Channel} that has two sides that can be shutdown independently.
+ */
+public interface DuplexChannel extends Channel {
+    /**
+     * Returns {@code true} if and only if the remote peer shut down its output so that no more
+     * data is received from this channel.  Note that the semantic of this method is different from
+     * that of {@link Socket#shutdownInput()} and {@link Socket#isInputShutdown()}.
+     */
+    boolean isInputShutdown();
+
+    /**
+     * @see Socket#isOutputShutdown()
+     */
+    boolean isOutputShutdown();
+
+    /**
+     * @see Socket#shutdownOutput()
+     */
+    ChannelFuture shutdownOutput();
+
+    /**
+     * @see Socket#shutdownOutput()
+     *
+     * Will notify the given {@link ChannelPromise}
+     */
+    ChannelFuture shutdownOutput(ChannelPromise promise);
+}
diff --git a/transport/src/main/java/io/netty/channel/socket/SocketChannel.java b/transport/src/main/java/io/netty/channel/socket/SocketChannel.java
index ba0962c9fd7..22562c7baba 100644
--- a/transport/src/main/java/io/netty/channel/socket/SocketChannel.java
+++ b/transport/src/main/java/io/netty/channel/socket/SocketChannel.java
@@ -16,16 +16,13 @@
 package io.netty.channel.socket;
 
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelPromise;
 
 import java.net.InetSocketAddress;
-import java.net.Socket;
 
 /**
  * A TCP/IP socket {@link Channel}.
  */
-public interface SocketChannel extends Channel {
+public interface SocketChannel extends DuplexChannel {
     @Override
     ServerSocketChannel parent();
 
@@ -35,28 +32,4 @@ public interface SocketChannel extends Channel {
     InetSocketAddress localAddress();
     @Override
     InetSocketAddress remoteAddress();
-
-    /**
-     * Returns {@code true} if and only if the remote peer shut down its output so that no more
-     * data is received from this channel.  Note that the semantic of this method is different from
-     * that of {@link Socket#shutdownInput()} and {@link Socket#isInputShutdown()}.
-     */
-    boolean isInputShutdown();
-
-    /**
-     * @see Socket#isOutputShutdown()
-     */
-    boolean isOutputShutdown();
-
-    /**
-     * @see Socket#shutdownOutput()
-     */
-    ChannelFuture shutdownOutput();
-
-    /**
-     * @see Socket#shutdownOutput()
-     *
-     * Will notify the given {@link ChannelPromise}
-     */
-    ChannelFuture shutdownOutput(ChannelPromise future);
 }


Test patch: