Move the VertxConnection close reason to the WebSocketConnectionImpl class.

Motivation:

Since only WebSocket use an effective close reason, we should move this generic part of the VertxConnection to WebSocketConnection implementation to simplify unclutter VertxConnection and keep it focus on its responsibitlies.

Author: vietj

vertx-core/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java

@@ -723,8 +723,8 @@ void handleClosed(Throwable err) {
   }
 
   @Override
-  protected void handleShutdown(Object reason, long timeout, TimeUnit unit, ChannelPromise promise) {
-    super.handleShutdown(reason, timeout, unit, promise);
+  protected void handleShutdown(long timeout, TimeUnit unit, ChannelPromise promise) {
+    super.handleShutdown(timeout, unit, promise);
     checkLifecycle();
   }
 
@@ -739,15 +739,15 @@ private boolean checkLifecycle() {
   }
 
   @Override
-  protected void writeClose(Object reason, ChannelPromise promise) {
+  protected void writeClose(ChannelPromise promise) {
     // Maybe move to handleShutdown
     if (!evicted) {
       evicted = true;
       if (evictionHandler != null) {
         evictionHandler.handle(null);
       }
     }
-    super.writeClose(reason, promise);
+    super.writeClose(promise);
   }
 
   private Throwable validateMessage(Object msg) {

vertx-core/src/main/java/io/vertx/core/http/impl/Http1xConnection.java

@@ -28,7 +28,6 @@
 import io.vertx.core.http.Http2Settings;
 import io.vertx.core.http.HttpConnection;
 import io.vertx.core.internal.ContextInternal;
-import io.vertx.core.net.impl.ConnectionBase;
 import io.vertx.core.net.impl.VertxConnection;
 
 import java.util.concurrent.TimeUnit;
@@ -40,7 +39,6 @@ abstract class Http1xConnection extends VertxConnection implements io.vertx.core
 
   protected boolean closeInitiated;
   protected boolean shutdownInitiated;
-  protected Object closeReason;
   protected long shutdownTimeout;
   protected TimeUnit shutdownUnit;
   protected ChannelPromise closePromise;
@@ -61,9 +59,8 @@ public synchronized Http1xConnection shutdownHandler(@Nullable Handler<Void> han
   }
 
   @Override
-  protected void handleShutdown(Object reason, long timeout, TimeUnit unit, ChannelPromise promise) {
+  protected void handleShutdown(long timeout, TimeUnit unit, ChannelPromise promise) {
     shutdownInitiated = true;
-    closeReason = reason;
     shutdownTimeout = timeout;
     shutdownUnit = unit;
     closePromise = promise;
@@ -77,16 +74,16 @@ protected void handleShutdown(Object reason, long timeout, TimeUnit unit, Channe
   }
 
   @Override
-  protected void writeClose(Object reason, ChannelPromise promise) {
+  protected void writeClose(ChannelPromise promise) {
     closeInitiated = true;
-    super.writeClose(reason, promise);
+    super.writeClose(promise);
   }
 
   protected void closeInternal() {
     if (closeInitiated) {
       // Nothing to do
     } else if (shutdownInitiated) {
-      super.handleShutdown(closeReason, shutdownTimeout, shutdownUnit, closePromise);
+      super.handleShutdown(shutdownTimeout, shutdownUnit, closePromise);
     } else {
       chctx.channel().close();
     }

vertx-core/src/main/java/io/vertx/core/http/impl/Http1xServerConnection.java

@@ -107,8 +107,8 @@ public Http1xServerConnection(ThreadingModel threadingModel,
   }
 
   @Override
-  protected void handleShutdown(Object reason, long timeout, TimeUnit unit, ChannelPromise promise) {
-    super.handleShutdown(reason, timeout, unit, promise);
+  protected void handleShutdown(long timeout, TimeUnit unit, ChannelPromise promise) {
+    super.handleShutdown(timeout, unit, promise);
     if (responseInProgress != null) {
     } else {
       closeInternal();

vertx-core/src/main/java/io/vertx/core/http/impl/ServerWebSocketImpl.java

@@ -28,7 +28,7 @@ public class ServerWebSocketImpl extends WebSocketImplBase<ServerWebSocketImpl>
   private final String query;
 
   ServerWebSocketImpl(ContextInternal context,
-                      VertxConnection conn,
+                      WebSocketConnectionImpl conn,
                       boolean supportsContinuation,
                       Http1xServerRequest request,
                       int maxWebSocketFrameSize,

vertx-core/src/main/java/io/vertx/core/http/impl/WebSocketConnectionImpl.java

@@ -16,6 +16,7 @@
 import io.netty.handler.codec.http.websocketx.*;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.ScheduledFuture;
+import io.vertx.core.Future;
 import io.vertx.core.http.WebSocketFrameType;
 import io.vertx.core.http.impl.ws.WebSocketFrameImpl;
 import io.vertx.core.http.impl.ws.WebSocketFrameInternal;
@@ -77,13 +78,24 @@ public NetworkMetrics metrics() {
     return metrics;
   }
 
+  private Object reason;
+
+  public Future<Void> close(Object reason) {
+    return shutdown(reason, 0L, TimeUnit.SECONDS);
+  }
+
+  public Future<Void> shutdown(Object reason, long timeout, TimeUnit unit) {
+    this.reason = reason;
+    return shutdown(timeout, unit);
+  }
+
   @Override
-  protected void handleShutdown(Object reason, long timeout, TimeUnit unit, ChannelPromise promise) {
+  protected void handleShutdown(long timeout, TimeUnit unit, ChannelPromise promise) {
     webSocket.handleShutdown();
   }
 
   @Override
-  protected void writeClose(Object reason, ChannelPromise promise) {
+  protected void writeClose(ChannelPromise promise) {
     assert !closeSent;
     closeSent = true;
     closePromise = promise;
@@ -205,7 +217,7 @@ private void finishClose() {
       closingTimeout = null;
       ChannelPromise p = closePromise;
       closePromise = null;
-      super.writeClose(closeReason, p);
+      super.writeClose(p);
     }
   }
 

vertx-core/src/main/java/io/vertx/core/http/impl/WebSocketImpl.java

@@ -24,7 +24,7 @@ public class WebSocketImpl extends WebSocketImplBase<WebSocketImpl> implements W
   private Handler<Void> evictionHandler;
 
   public WebSocketImpl(ContextInternal context,
-                       VertxConnection conn,
+                       WebSocketConnectionImpl conn,
                        boolean supportsContinuation,
                        int maxWebSocketFrameSize,
                        int maxWebSocketMessageSize,

vertx-core/src/main/java/io/vertx/core/http/impl/WebSocketImplBase.java

@@ -59,7 +59,7 @@ public abstract class WebSocketImplBase<S extends WebSocket> implements WebSocke
   private final String binaryHandlerID;
   private final int maxWebSocketFrameSize;
   private final int maxWebSocketMessageSize;
-  private final VertxConnection conn;
+  private final WebSocketConnectionImpl conn;
   private ChannelHandlerContext chctx;
   protected final ContextInternal context;
   private final InboundMessageQueue<WebSocketFrameInternal> pending;
@@ -82,7 +82,7 @@ public abstract class WebSocketImplBase<S extends WebSocket> implements WebSocke
   private MultiMap headers;
 
   WebSocketImplBase(ContextInternal context,
-                    VertxConnection conn,
+                    WebSocketConnectionImpl conn,
                     MultiMap headers,
                     boolean supportsContinuation,
                     int maxWebSocketFrameSize,

vertx-core/src/main/java/io/vertx/core/net/impl/SocketBase.java

@@ -251,7 +251,7 @@ public synchronized S shutdownHandler(@Nullable Handler<Void> handler) {
   }
 
   @Override
-  protected void handleShutdown(Object reason, long timeout, TimeUnit unit, ChannelPromise promise) {
+  protected void handleShutdown(long timeout, TimeUnit unit, ChannelPromise promise) {
     Handler<Void> handler;
     synchronized (this) {
       handler = shutdownHandler;

vertx-core/src/main/java/io/vertx/core/net/impl/VertxConnection.java

@@ -134,23 +134,21 @@ protected boolean supportsFileRegion() {
     return vertx.transport().supportFileRegion() && !isSsl() &&!isTrafficShaped();
   }
 
-  protected void handleShutdown(Object reason, long timeout, TimeUnit unit, ChannelPromise promise) {
+  protected void handleShutdown(long timeout, TimeUnit unit, ChannelPromise promise) {
     // Assert from event-loop
     ScheduledFuture<?> t = shutdownTimeout;
     if (t != null) {
       shutdownTimeout = null;
       t.cancel(false);
-      terminateClose(reason, promise);
+      terminateClose(promise);
     }
   }
 
   private static class CloseChannelPromise extends DefaultChannelPromise {
-    final Object reason;
     final long timeout;
     final TimeUnit unit;
-    public CloseChannelPromise(Channel channel, Object reason, long timeout, TimeUnit unit) {
+    public CloseChannelPromise(Channel channel, long timeout, TimeUnit unit) {
       super(channel);
-      this.reason = reason;
       this.timeout = timeout;
       this.unit = unit;
     }
@@ -160,23 +158,12 @@ public CloseChannelPromise(Channel channel, Object reason, long timeout, TimeUni
    * Close the connection
    */
   public final Future<Void> close() {
-    return close(null);
-  }
-
-  /**
-   * Close the connection
-   */
-  public final Future<Void> close(Object reason) {
-    return shutdown(reason, 0L, TimeUnit.SECONDS);
+    return shutdown(0L, TimeUnit.SECONDS);
   }
 
   // Is this necessary ?
   public final Future<Void> shutdown(long timeout, TimeUnit unit) {
-    return shutdown(null, timeout, unit);
-  }
-
-  public final Future<Void> shutdown(Object reason, long timeout, TimeUnit unit) {
-    CloseChannelPromise promise = new CloseChannelPromise(channel, reason, timeout, unit);
+    CloseChannelPromise promise = new CloseChannelPromise(channel, timeout, unit);
     shutdown(promise);
     PromiseInternal<Void> p = context.promise();
     promise.addListener(p);
@@ -198,18 +185,15 @@ private void shutdown(CloseChannelPromise promise) {
   final void handleClose(ChannelPromise promise) {
     if (closeInitiated != null) {
       long timeout;
-      Object closeReason;
       if (promise instanceof CloseChannelPromise) {
         timeout = ((CloseChannelPromise)promise).timeout;
-        closeReason = ((CloseChannelPromise)promise).reason;
       } else {
         timeout = 0L;
-        closeReason = null;
       }
       if (timeout == 0L && !closeSent) {
         closeSent = true;
         closeInitiated = promise;
-        writeClose(closeReason, promise);
+        writeClose(promise);
       } else {
         channel
           .closeFuture()
@@ -223,36 +207,33 @@ final void handleClose(ChannelPromise promise) {
       }
     } else {
       closeInitiated = promise;
-      Object reason;
       long timeout;
       TimeUnit unit;
       if (promise instanceof CloseChannelPromise) {
         CloseChannelPromise closeChannelPromise = (CloseChannelPromise) promise;
-        reason = closeChannelPromise.reason;
         timeout = closeChannelPromise.timeout;
         unit = closeChannelPromise.unit;
       } else {
-        reason = null;
         timeout = 0L;
         unit = TimeUnit.SECONDS;
       }
       if (timeout == 0L) {
-        terminateClose(reason, promise);
+        terminateClose(promise);
       } else {
         EventExecutor el = chctx.executor();
         shutdownTimeout = el.schedule(() -> {
           shutdownTimeout = null;
-          terminateClose(reason, promise);
+          terminateClose(promise);
         }, timeout, unit);
-        handleShutdown(reason, timeout, unit, promise);
+        handleShutdown(timeout, unit, promise);
       }
     }
   }
 
-  private void terminateClose(Object reason, ChannelPromise promise) {
+  private void terminateClose(ChannelPromise promise) {
     if (!closeSent) {
       closeSent = true;
-      writeClose(reason, promise);
+      writeClose(promise);
     }
   }
 
@@ -265,8 +246,8 @@ private void terminateClose(Object reason, ChannelPromise promise) {
    * <p/>
    * This method is exclusively called on the event-loop thread and relays a channel user event.
    */
-  protected void writeClose(Object reason, ChannelPromise promise) {
-    writeClose(promise);
+  protected void writeClose(ChannelPromise promise) {
+    doWriteClose(promise);
   }
 
   protected void handleClosed() {
@@ -450,7 +431,7 @@ public final ChannelPromise write(Object msg, boolean forceFlush, ChannelPromise
    *
    * @param promise the promise receiving the completion event
    */
-  private void writeClose(ChannelPromise promise) {
+  private void doWriteClose(ChannelPromise promise) {
     // Make sure everything is flushed out on close
     ChannelPromise channelPromise = chctx
       .newPromise()

vertx-core/src/test/java/io/vertx/tests/net/VertxConnectionTest.java

@@ -559,11 +559,11 @@ protected void handleEvent(Object event) {
         }
       }
       @Override
-      protected void handleShutdown(Object reason, long timeout, TimeUnit unit, ChannelPromise promise) {
+      protected void handleShutdown(long timeout, TimeUnit unit, ChannelPromise promise) {
         shutdown.set(true);
       }
       @Override
-      protected void writeClose(Object reason, ChannelPromise promise) {
+      protected void writeClose(ChannelPromise promise) {
         closed.set(true);
         promise.setSuccess();
       }
@@ -585,11 +585,11 @@ protected void handleEvent(Object event) {
         }
       }
       @Override
-      protected void handleShutdown(Object reason, long timeout, TimeUnit unit, ChannelPromise promise) {
+      protected void handleShutdown(long timeout, TimeUnit unit, ChannelPromise promise) {
         shutdown.set(true);
       }
       @Override
-      protected void writeClose(Object reason, ChannelPromise promise) {
+      protected void writeClose(ChannelPromise promise) {
         closed.set(true);
       }
     });
@@ -611,13 +611,13 @@ protected void handleEvent(Object event) {
         }
       }
       @Override
-      protected void handleShutdown(Object reason, long timeout, TimeUnit unit, ChannelPromise promise) {
+      protected void handleShutdown(long timeout, TimeUnit unit, ChannelPromise promise) {
         shutdown.set(true);
-        close(reason);
+        close();
         assertTrue(closed.get());
       }
       @Override
-      protected void writeClose(Object reason, ChannelPromise promise) {
+      protected void writeClose(ChannelPromise promise) {
         closed.set(true);
       }
     });
@@ -638,7 +638,7 @@ protected void handleEvent(Object event) {
         }
       }
       @Override
-      protected void handleShutdown(Object reason, long timeout, TimeUnit unit, ChannelPromise promise) {
+      protected void handleShutdown(long timeout, TimeUnit unit, ChannelPromise promise) {
         shutdown.getAndIncrement();
         assertEquals(0L, closed.get());
         // Force run tasks at this stage since the task will be cancelled after by embedded channel close
@@ -648,7 +648,7 @@ protected void handleShutdown(Object reason, long timeout, TimeUnit unit, Channe
         assertEquals(1L, closed.get());
       }
       @Override
-      protected void writeClose(Object reason, ChannelPromise promise) {
+      protected void writeClose(ChannelPromise promise) {
         closed.getAndIncrement();
       }
     });