package ch.polgrabia.demos.server;

import ch.polgrabia.demos.utils.TopicUtil;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.ServerWebSocket;
import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class WebsocketServerHandler implements Handler<ServerWebSocket> {
    private static final Logger logger = LoggerFactory.getLogger(WebsocketServerHandler.class);

    private static final ConcurrentHashMap<String, Set<ServerWebSocket>>
            channelSocketMapping = new ConcurrentHashMap<>();
    private final Producer<String, byte[]> kafkaProducer;

    public WebsocketServerHandler(Producer<String, byte[]> kafkaProducer) {
        this.kafkaProducer = kafkaProducer;
    }

    @Override
    public void handle(ServerWebSocket serverWebSocket) {
        serverWebSocket.writeBinaryMessage(Buffer.buffer("Hello"));
        serverWebSocket.handler(new WebsocketMessageHandler(kafkaProducer, serverWebSocket));
        String channel = TopicUtil.convertTopicName(serverWebSocket.path());
        Set<ServerWebSocket> sockets = channelSocketMapping.computeIfAbsent(channel, (k) -> {
            Set<ServerWebSocket> s = new HashSet<>();
            s.add(serverWebSocket);
            return s;
        });


        serverWebSocket.closeHandler(aVoid -> {
            int id = System.identityHashCode(serverWebSocket);
            logger.info("[{}] Closing web socket with id {}", channel, id);
            Set<ServerWebSocket> serverWebSockets = channelSocketMapping.get(channel);
            if (serverWebSockets != null) {
                boolean removed = serverWebSockets
                        .remove(serverWebSocket);
                if (!removed) {
                    logger.warn("Server websocket channel {} was not found in the channel set of sockets." +
                            " It should never happen", channel);
                }
            } else {
                logger.warn("Channel {} not found. It should never happen",channel);
            }
        });

        sockets.add(serverWebSocket);
    }

}
