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.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class WebsocketMessageHandler implements Handler<Buffer> {
  private static final Logger logger = LoggerFactory.getLogger(WebsocketMessageHandler.class);
  private final ServerWebSocket serverWebSocket;
  private final Producer<String, byte[]> kafkaProducer;

  public WebsocketMessageHandler(Producer<String, byte[]> kafkaProducer, ServerWebSocket serverWebSocket) {
    this.kafkaProducer = kafkaProducer;
    this.serverWebSocket = serverWebSocket;
    this.serverWebSocket.setWriteQueueMaxSize(0);
  }


  @Override
  public void handle(Buffer s) {
    String channel = TopicUtil.convertTopicName(serverWebSocket.path());
    logger.info("[{}] Got message {}", channel, s);
    serverWebSocket.writeBinaryMessage(Buffer.buffer("Pong: " + s.toString() + "\n")); // TBD send to kafka
    if (kafkaProducer != null) {
      kafkaProducer.send(new ProducerRecord<>(channel, channel, s.getBytes())); // kafka setup needs to be checked
    }
  }
}
