package ch.polgrabia.demos.server;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.http.HttpServer;
import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;

public class WebsocketServerChatVerticle extends AbstractVerticle {
    private static final Logger logger = LoggerFactory.getLogger(WebsocketServerChatVerticle.class);
    private final Producer<String, byte[]> kafkaProducer;
    private HttpServer httpServer;
    private final int port;
    private final String hostname;

    private final Object runningMonitor = new Object();
    private boolean shutdownActivation = false;

    public WebsocketServerChatVerticle(WebsocketServerConfig websocketServerConfig) {
        this.port = websocketServerConfig.port() == null ? 8080 : Integer.parseInt(websocketServerConfig.port());
        this.hostname = websocketServerConfig.host() == null ? "localhost" : websocketServerConfig.host();
        this.kafkaProducer = websocketServerConfig.kafkaProducer();
        logger.info("Host: {}, port {}", hostname, port);
    }

    @Override
    public void start() throws Exception {
        httpServer = getVertx()
                .createHttpServer();
        httpServer.webSocketHandler(new WebsocketServerHandler(kafkaProducer)).listen(port, hostname);
        shutdownActivation = true;
    }

    @Override
    public void stop() throws Exception {
        httpServer.close();
        shutdownActivation = false;
        synchronized (runningMonitor) {
            runningMonitor.notifyAll();
        }
    }

    public CompletableFuture<Boolean> waitUntilFinished() {
        return CompletableFuture.supplyAsync(() -> {
            while (!shutdownActivation) {
                synchronized (runningMonitor) {
                    try {
                        runningMonitor.wait();
                        return true;
                    } catch (InterruptedException e) {
                        logger.warn("We got interrupted, retrying", e);
                    }
                }
            }
            return false;
        });
    }
}
