package io.quarkus.smallrye.graphql.runtime;

import java.util.Map;

import org.jboss.logging.Logger;

import io.quarkus.security.identity.CurrentIdentityAssociation;
import io.quarkus.vertx.http.runtime.CurrentVertxRequest;
import io.smallrye.graphql.websocket.GraphQLWebSocketSession;
import io.smallrye.graphql.websocket.GraphQLWebsocketHandler;
import io.smallrye.graphql.websocket.graphqltransportws.GraphQLTransportWSSubprotocolHandler;
import io.smallrye.graphql.websocket.graphqlws.GraphQLWSSubprotocolHandler;
import io.vertx.core.http.HttpHeaders;
import io.vertx.core.http.ServerWebSocket;
import io.vertx.ext.web.RoutingContext;

/**
 * Handler that does the execution of GraphQL Requests
 */
public class SmallRyeGraphQLOverWebSocketHandler extends SmallRyeGraphQLAbstractHandler {
    private static final Logger log = Logger.getLogger(SmallRyeGraphQLOverWebSocketHandler.class);

    public SmallRyeGraphQLOverWebSocketHandler(CurrentIdentityAssociation currentIdentityAssociation,
            CurrentVertxRequest currentVertxRequest, boolean runBlocking) {
        super(currentIdentityAssociation, currentVertxRequest, runBlocking);
    }

    @Override
    protected void doHandle(final RoutingContext ctx) {

        if (ctx.request().headers().contains(HttpHeaders.UPGRADE, HttpHeaders.WEBSOCKET, true) && !ctx.request().isEnded()) {
            Map<String, Object> metaData = getMetaData(ctx);
            ctx.request().toWebSocket(event -> {
                if (event.succeeded()) {
                    ServerWebSocket serverWebSocket = event.result();
                    String subprotocol = serverWebSocket.subProtocol();
                    GraphQLWebsocketHandler handler = null;
                    switch (subprotocol) {
                        case "graphql-transport-ws":
                            handler = new GraphQLTransportWSSubprotocolHandler(
                                    new QuarkusVertxWebSocketSession(serverWebSocket), metaData);
                            break;
                        case "graphql-ws":
                            handler = new GraphQLWSSubprotocolHandler(
                                    new QuarkusVertxWebSocketSession(serverWebSocket), metaData);
                            break;
                        default:
                            log.warn("Unknown graphql-over-websocket protocol: " + subprotocol);
                            serverWebSocket.close();
                            return;
                    }
                    log.debug("Starting websocket with subprotocol = " + subprotocol);
                    GraphQLWebsocketHandler finalHandler = handler;
                    serverWebSocket.closeHandler(v -> finalHandler.onClose());
                    serverWebSocket.endHandler(v -> finalHandler.onEnd());
                    serverWebSocket.exceptionHandler(t -> finalHandler.onThrowable(t));
                    serverWebSocket.textMessageHandler(m -> finalHandler.onMessage(m));
                } else {
                    log.warn("WebSocket failed", event.cause());
                }
            });
        } else {
            ctx.next();
        }
    }

    private static class QuarkusVertxWebSocketSession implements GraphQLWebSocketSession {

        private final ServerWebSocket webSocket;
        private final String peer;

        QuarkusVertxWebSocketSession(ServerWebSocket webSocket) {
            this.webSocket = webSocket;
            if (webSocket.remoteAddress() != null) {
                this.peer = webSocket.remoteAddress().host() + ":" + webSocket.remoteAddress().port();
            } else {
                this.peer = "unknown";
            }
        }

        @Override
        public void sendMessage(String message) {
            if (log.isTraceEnabled()) {
                log.trace(">>> " + message);
            }
            webSocket.writeTextMessage(message);
        }

        @Override
        public void close(short statusCode, String reason) {
            if (log.isDebugEnabled()) {
                log.debug("Closing graphql websocket connection with code=" + statusCode + " and reason " + reason);
            }
            webSocket.close(statusCode, reason);
        }

        @Override
        public boolean isClosed() {
            return webSocket.isClosed();
        }

        @Override
        public String toString() {
            return "{peer=" + peer + "}";
        }
    }

}

@@ -1636,24 +1636,241 @@
 Protocol();%0A
+                    if (subprotocol == null) %7B%0A                        log.warn(%22Websocket subprotocol is null%22);%0A                        serverWebSocket.close();%0A                        return;%0A                    %7D%0A
             
@@ -1912,15 +1912,8 @@
 dler
- = null
 ;%0A  
@@ -2736,16 +2736,17 @@
 og.debug
+f
 (%22Starti
@@ -2777,19 +2777,20 @@
 tocol = 
-%22 +
+%25s%22,
  subprot
@@ -3071,21 +3071,16 @@
 Handler(
-t -%3E 
 finalHan
@@ -3083,17 +3083,18 @@
 lHandler
-.
+::
 onThrowa
@@ -3100,11 +3100,8 @@
 able
-(t)
 );%0A 
@@ -3158,13 +3158,8 @@
 ler(
-m -%3E 
 fina
@@ -3166,17 +3166,18 @@
 lHandler
-.
+::
 onMessag
@@ -3181,11 +3181,8 @@
 sage
-(m)
 );%0A 
@@ -3238,17 +3238,17 @@
 arn(%22Web
-S
+s
 ocket fa

Fix NPE when the subprotocal of websocket is null