package ch.polgrabia.demos.client;

import ch.polgrabia.demos.utils.TopicUtil;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.AsyncResult;
import io.vertx.core.http.HttpClient;
import io.vertx.core.http.HttpClientOptions;
import io.vertx.core.http.WebSocket;
import io.vertx.core.http.impl.WebSocketClientImpl;
import io.vertx.core.impl.CloseFuture;
import io.vertx.core.impl.VertxInternal;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;

public class WebsocketClientParticle extends AbstractVerticle {
  private static final Logger logger = LoggerFactory.getLogger(WebsocketClientParticle.class);
  private final String webSocketHostname;
  private final int webSocketPort;
  // path maps to channel
  private final String webSocketPath;
  private final String channel;
  private final ClientPeriodicStdinScannerParticle clientPeriodicStdinScannerParticle = new ClientPeriodicStdinScannerParticle(
    System.in
  );
  private final WebApiParticle webApiParticle;
  private HttpClient client;
  private boolean isShuttingDownActivated = false;
  private final Object runningMonitor = new Object();
  private WebSocketSenderParticle webSocketSenderParticle;
  private WebSocketReceiverParticle webSocketReceiverParticle;

  public WebsocketClientParticle(
    String webSocketHostname,
    int webSocketPort,
    String webSocketPath,
    int clientApiPort) {
    this.webSocketHostname = webSocketHostname;
    this.webSocketPort = webSocketPort;
    this.webSocketPath = webSocketPath;
    this.channel = TopicUtil.convertTopicName(webSocketPath);
    this.webApiParticle = new WebApiParticle(clientApiPort);
  }

  @Override
  public void start() {
    this.client = getVertx()
      .createHttpClient();
    logger.info("Starting websocket client command server");
    new WebSocketClientImpl((VertxInternal) getVertx(), new HttpClientOptions(), new CloseFuture())
      .webSocket(webSocketPort, webSocketHostname, webSocketPath, this::handleMessage);
    getVertx().deployVerticle(clientPeriodicStdinScannerParticle);
    getVertx().deployVerticle(webApiParticle);
  }

  private void handleMessage(AsyncResult<WebSocket> webSocketAsyncResult) {
    logger.info("Handling websocket");
    if (!webSocketAsyncResult.succeeded()) {
      logger.warn("Websocket handler connection failed", webSocketAsyncResult.cause());
      try {
        stop();
      } catch (Exception e) {
        logger.error("Failed to stop particle", e);
      }
      return;
    }
    WebSocket webSocket = webSocketAsyncResult.result();
    webSocketReceiverParticle = new WebSocketReceiverParticle(webSocket, channel, this::stop);
    getVertx().deployVerticle(webSocketReceiverParticle);
    webSocketSenderParticle = new WebSocketSenderParticle(webSocket);
    getVertx().deployVerticle(webSocketSenderParticle);
    vertx.setPeriodic(Constants.CLIENT_CONSOLE_INPUT_SCANNING_PERIOD_MS, clientPeriodicStdinScannerParticle);
  }

  @Override
  public void stop() {
    isShuttingDownActivated = true;
    getVertx().undeploy(webApiParticle.deploymentID());
    getVertx().undeploy(webSocketSenderParticle.deploymentID());
    getVertx().undeploy(webSocketReceiverParticle.deploymentID());
    getVertx().undeploy(clientPeriodicStdinScannerParticle.deploymentID());
    client.close();
    synchronized (runningMonitor) {
      runningMonitor.notifyAll();
    }
  }

  public CompletableFuture<Boolean> waitingUntilFinished() {
    return CompletableFuture.supplyAsync(() -> {
      while (!isShuttingDownActivated) {
        synchronized (runningMonitor) {
          try {
            runningMonitor.wait();
            return true;
          } catch (InterruptedException e) {
            logger.warn("Got interrupted", e);
          }
        }
      }
      return false;
    });
  }

}
