package ch.polgrabia.demos.server;

import io.vertx.core.Vertx;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class WebsocketServerApp {
    private static final Logger logger = LoggerFactory.getLogger(WebsocketServerApp.class);

    public static void main(String[] args) {
        var vertx = Vertx.vertx();
        Producer<String, byte[]> kafkaProducer = null;
        try {
            CommandLineParser parser = new DefaultParser();
            var options = new Options()
                    .addOption("h", "host", true, "Host")
                    .addOption("p", "port", true, "Port")
                    .addOption("ke", "kafka-enabled", true, "Enable Kafka server")
                    .addOption("bs", "bootstrap-servers", true, "Bootstrap server address");
            CommandLine result = parser.parse(options, args);
            logger.info("Got options: {}", result);
            boolean isKafkaEnabled = "true".equals(result.getOptionValue("kafka-enabled"));
            String bootstrapServers = result.getOptionValue("bootstrap-servers");
            logger.info("Got options kafka.enabled: {}", isKafkaEnabled);
            if (isKafkaEnabled) {
                logger.info("Bootstrap servers {}", bootstrapServers);
                Properties props = new Properties();
                props.put("bootstrap.servers", bootstrapServers);
                props.put("linger.ms", 1);
                props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
                props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

                kafkaProducer = new KafkaProducer<>(props);
            }

            var config = new WebsocketServerConfig(
                    result.getOptionValue("host"),
                    result.getOptionValue("port"),
                    isKafkaEnabled,
                    bootstrapServers,
                    kafkaProducer);

            logger.info("Configuration {}", config);

            WebsocketServerChatVerticle websocketServerChatVerticle = new WebsocketServerChatVerticle(config); // kafka producer
            vertx.deployVerticle(websocketServerChatVerticle);
            websocketServerChatVerticle
                    .waitUntilFinished()
                    .join();
        } catch (ParseException e) {
            logger.error("Invalid use of configuration", e);
        } finally {
            vertx.close();
            if (kafkaProducer != null) {
                kafkaProducer.close();
            }
        }

    }
}