package io.quarkus.resteasy.runtime.standalone;

import java.io.IOException;
import java.io.InterruptedIOException;

import org.jboss.logging.Logger;

import io.netty.buffer.ByteBuf;
import io.vertx.core.Context;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpServerRequest;

public class VertxBlockingOutput implements VertxOutput {
    private static final Logger log = Logger.getLogger("io.quarkus.resteasy");

    protected boolean waitingForDrain;
    protected boolean drainHandlerRegistered;
    protected final HttpServerRequest request;
    protected boolean first = true;
    protected Throwable throwable;

    public VertxBlockingOutput(HttpServerRequest request) {
        this.request = request;
        request.response().exceptionHandler(new Handler<Throwable>() {
            @Override
            public void handle(Throwable event) {
                throwable = event;
                log.debugf(event, "IO Exception ");
                //TODO: do we need this?
                terminateResponse();
                request.connection().close();
                synchronized (request.connection()) {
                    if (waitingForDrain) {
                        request.connection().notify();
                    }
                }
            }
        });

        request.response().endHandler(new Handler<Void>() {
            @Override
            public void handle(Void event) {
                synchronized (request.connection()) {
                    if (waitingForDrain) {
                        request.connection().notify();
                    }
                }
                terminateResponse();
            }
        });
    }

    public void terminateResponse() {

    }

    Buffer createBuffer(ByteBuf data) {
        return new VertxBufferImpl(data);
    }

    @Override
    public void write(ByteBuf data, boolean last) throws IOException {
        if (last && data == null) {
            request.response().end();
            return;
        }
        try {
            //do all this in the same lock
            synchronized (request.connection()) {
                try {
                    awaitWriteable();
                    if (last) {
                        request.response().end(createBuffer(data));
                    } else {
                        request.response().write(createBuffer(data));
                    }
                } catch (Exception e) {
                    if (data != null && data.refCnt() > 0) {
                        data.release();
                    }
                    throw new IOException("Failed to write", e);
                }
            }
        } finally {
            if (last) {
                terminateResponse();
            }
        }
    }

    private void awaitWriteable() throws IOException {
        if (first) {
            first = false;
            return;
        }
        assert Thread.holdsLock(request.connection());
        while (request.response().writeQueueFull()) {
            if (throwable != null) {
                throw new IOException(throwable);
            }
            if (Context.isOnEventLoopThread()) {
                throw new IOException("Attempting a blocking write on io thread");
            }
            if (request.response().closed()) {
                throw new IOException("Connection has been closed");
            }
            if (!drainHandlerRegistered) {
                drainHandlerRegistered = true;
                Handler<Void> handler = new Handler<Void>() {
                    @Override
                    public void handle(Void event) {
                        if (waitingForDrain) {
                            request.connection().notifyAll();
                        }
                    }
                };
                request.response().drainHandler(handler);
                request.response().closeHandler(handler);
            }
            try {
                waitingForDrain = true;
                request.connection().wait();
            } catch (InterruptedException e) {
                throw new InterruptedIOException(e.getMessage());
            } finally {
                waitingForDrain = false;
            }
        }
    }

}

@@ -268,24 +268,66 @@
 fer.Buffer;%0A
+import io.vertx.core.http.HttpConnection;%0A
 import io.ve
@@ -3773,41 +3773,199 @@
     
-request.connection().notifyAll();
+HttpConnection connection = request.connection();%0A                            synchronized (connection) %7B%0A                                connection.notifyAll();%0A                            %7D
 %0A   

Fix monitor state bug