Fix request hanging condition

This condition could happen when a TLS secured connection had a lot  of data written to it
very fast from an event loop thread.

The problem was that there was a race condition
on the registration of the drain handler,
which then resulted in the pending data never
being written out.
The fix is to always register the drain handler

Resolves: quarkusio#37807

independent-projects/resteasy-reactive/server/vertx/src/main/java/org/jboss/resteasy/reactive/server/vertx/ResteasyReactiveOutputStream.java

@@ -57,6 +57,9 @@ public void handle(Throwable event) {
                 }
             }
         });
+        Handler<Void> handler = new DrainHandler(this);
+        request.response().drainHandler(handler);
+        request.response().closeHandler(handler);
 
         context.getContext().addEndHandler(new Handler<AsyncResult<Void>>() {
             @Override
@@ -90,7 +93,7 @@ public void write(ByteBuf data, boolean last) throws IOException {
                 boolean bufferRequired = awaitWriteable() || (overflow != null && overflow.size() > 0);
                 if (bufferRequired) {
                     //just buffer everything
-                    registerDrainHandler();
+                    //                    registerDrainHandler();
                     if (overflow == null) {
                         overflow = new ByteArrayOutputStream();
                     }
@@ -135,7 +138,7 @@ private boolean awaitWriteable() throws IOException {
             if (request.response().closed()) {
                 throw new IOException("Connection has been closed");
             }
-            registerDrainHandler();
+            //            registerDrainHandler();
             try {
                 waitingForDrain = true;
                 request.connection().wait();
@@ -148,33 +151,14 @@ private boolean awaitWriteable() throws IOException {
         return false;
     }
 
-    private void registerDrainHandler() {
-        if (!drainHandlerRegistered) {
-            drainHandlerRegistered = true;
-            Handler<Void> handler = new Handler<Void>() {
-                @Override
-                public void handle(Void event) {
-                    synchronized (request.connection()) {
-                        if (waitingForDrain) {
-                            request.connection().notifyAll();
-                        }
-                        if (overflow != null) {
-                            if (overflow.size() > 0) {
-                                if (closed) {
-                                    request.response().end(Buffer.buffer(overflow.toByteArray()), null);
-                                } else {
-                                    request.response().write(Buffer.buffer(overflow.toByteArray()), null);
-                                }
-                                overflow.reset();
-                            }
-                        }
-                    }
-                }
-            };
-            request.response().drainHandler(handler);
-            request.response().closeHandler(handler);
-        }
-    }
+    //    private void registerDrainHandler() {
+    //        if (!drainHandlerRegistered) {
+    //            drainHandlerRegistered = true;
+    //            Handler<Void> handler = new DrainHandler(this);
+    //            request.response().drainHandler(handler);
+    //            request.response().closeHandler(handler);
+    //        }
+    //    }
 
     /**
      * {@inheritDoc}
@@ -299,4 +283,30 @@ public void close() throws IOException {
         }
     }
 
+    private static class DrainHandler implements Handler<Void> {
+        private final ResteasyReactiveOutputStream out;
+
+        public DrainHandler(ResteasyReactiveOutputStream out) {
+            this.out = out;
+        }
+
+        @Override
+        public void handle(Void event) {
+            synchronized (out.request.connection()) {
+                if (out.waitingForDrain) {
+                    out.request.connection().notifyAll();
+                }
+                if (out.overflow != null) {
+                    if (out.overflow.size() > 0) {
+                        if (out.closed) {
+                            out.request.response().end(Buffer.buffer(out.overflow.toByteArray()), null);
+                        } else {
+                            out.request.response().write(Buffer.buffer(out.overflow.toByteArray()), null);
+                        }
+                        out.overflow.reset();
+                    }
+                }
+            }
+        }
+    }
 }