Quic stream reset handling improvements.

Motivation:

The initial implementation of stream reset handling closes the stream upon reception of a reset frame when a stream handler is present.

It is the desired default behavior, however a stream having sent or received a reset frame can still process or read data on the other direction. Setting a handler on the stream makes the application aware of the reset and when it happens the stream should not be closed (that is writing a final stream frame) and instead the application should take care of it at the most appropriate time the protocol requires it.

Changes:

Setting a quic stream handler inhibates exception handling and stream closing.

Author: vietj

vertx-core/src/main/java/io/vertx/core/datagram/impl/DatagramSocketImpl.java

@@ -348,7 +348,7 @@ public NetworkMetrics metrics() {
     }
 
     @Override
-    protected void handleException(Throwable t) {
+    protected boolean handleException(Throwable t) {
       super.handleException(t);
       Handler<Throwable> handler;
       synchronized (DatagramSocketImpl.this) {
@@ -357,6 +357,7 @@ protected void handleException(Throwable t) {
       if (handler != null) {
         handler.handle(t);
       }
+      return true;
     }
 
     @Override

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

@@ -1222,8 +1222,8 @@ protected void handleIdle(IdleStateEvent event) {
   }
 
   @Override
-  public void handleException(Throwable e) {
-    super.handleException(e);
+  public boolean handleException(Throwable e) {
+    boolean ret = super.handleException(e);
     LinkedHashSet<Stream> allStreams = new LinkedHashSet<>();
     synchronized (this) {
       allStreams.addAll(requests);
@@ -1232,6 +1232,7 @@ public void handleException(Throwable e) {
     for (Stream stream : allStreams) {
       stream.handleException(e);
     }
+    return ret;
   }
 
   @Override

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

@@ -472,8 +472,8 @@ protected void handleClosed() {
   }
 
   @Override
-  public void handleException(Throwable t) {
-    super.handleException(t);
+  public boolean handleException(Throwable t) {
+    boolean ret = super.handleException(t);
     Http1xServerRequest responseInProgress = this.responseInProgress;
     Http1xServerRequest requestInProgress = this.requestInProgress;
     if (requestInProgress != null) {
@@ -484,6 +484,7 @@ public void handleException(Throwable t) {
       responseInProgress.reportMetricsFailed = true;
       responseInProgress.handleException(t);
     }
+    return ret;
   }
 
   @Override

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

@@ -136,11 +136,12 @@ private CloseWebSocketFrame closeFrame(short statusCode, String reason) {
   }
 
   @Override
-  public void handleException(Throwable t) {
+  public boolean handleException(Throwable t) {
     WebSocketImplBase<?> ws = webSocket;
     if (ws != null) {
       ws.context().execute(t, ws::handleException);
     }
+    return true;
   }
 
   @Override

vertx-core/src/main/java/io/vertx/core/net/QuicStream.java

@@ -42,7 +42,13 @@ public interface QuicStream extends Socket {
   QuicConnection connection();
 
   /**
-   * Set a handler called upon stream reset.
+   * <p>Set a handler called upon stream reset: when a stream receives a reset frame from its peer,
+   * this handler is called.</p>
+   *
+   * <p>When no such handler is set, the stream exception handler is called and then the stream is automatically
+   * closed. Setting this handler changes this behavior: the handler processes the reset event and the handler
+   * has the full responsibility of managing the stream. That means the sending part of this stream is left
+   * untouched and the application can continue sending data.</p>
    *
    * @param handler the handler
    * @return this instance of a stream

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

@@ -160,7 +160,7 @@ public NetworkMetrics metrics() {
     return null;
   }
 
-  protected void handleException(Throwable t) {
+  protected boolean handleException(Throwable t) {
     NetworkMetrics metrics = metrics();
     if (metrics != null) {
       metrics.exceptionOccurred(metric, remoteAddress(), t);
@@ -180,6 +180,7 @@ protected void handleException(Throwable t) {
         }
       }
     });
+    return true;
   }
 
   protected void handleClosed() {

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

@@ -144,10 +144,15 @@ public void channelWritabilityChanged(ChannelHandlerContext ctx) {
   @Override
   public void exceptionCaught(ChannelHandlerContext chctx, final Throwable t) {
     C connection = getConnection();
+    boolean close;
     if (connection != null) {
-      connection.handleException(t);
+      close = connection.handleException(t);
+    } else {
+      close = true;
+    }
+    if (close) {
+      chctx.close();
     }
-    chctx.close();
   }
 
   @Override

vertx-core/src/main/java/io/vertx/core/net/impl/quic/QuicConnectionHandler.java

@@ -104,8 +104,9 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
 
   @Override
   public void exceptionCaught(ChannelHandlerContext chctx, final Throwable t) {
-    connection.handleException(t);
-    chctx.close();
+    if (connection.handleException(t)) {
+      chctx.close();
+    }
   }
 
   @Override

vertx-core/src/main/java/io/vertx/core/net/impl/quic/QuicConnectionImpl.java

@@ -131,8 +131,8 @@ public QuicConnectionImpl closeHandler(Handler<Void> handler) {
   }
 
   @Override
-  protected void handleException(Throwable t) {
-    super.handleException(t);
+  protected boolean handleException(Throwable t) {
+    return super.handleException(t);
   }
 
   void handleClosed(QuicConnectionClose payload) {

vertx-core/src/main/java/io/vertx/core/net/impl/quic/QuicStreamImpl.java

@@ -115,18 +115,18 @@ protected void handleEvent(Object event) {
   }
 
   @Override
-  protected void handleException(Throwable t) {
+  protected boolean handleException(Throwable t) {
     if (t instanceof QuicException) {
       QuicException quicException = (QuicException) t;
       if (quicException.error() == null && "STREAM_RESET".equals(quicException.getMessage())) {
         Handler<Integer> handler = resetHandler;
         if (handler != null) {
           context.emit(0, handler);
+          return false;
         }
-        return;
       }
     }
-    super.handleException(t);
+    return super.handleException(t);
   }
 
   @Override