/*
 * Copyright (c) 2011-2017 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 *     The Eclipse Public License is available at
 *     http://www.eclipse.org/legal/epl-v10.html
 *
 *     The Apache License v2.0 is available at
 *     http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */

package io.vertx.micrometer.backend;

import io.micrometer.prometheus.PrometheusMeterRegistry;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.http.HttpClientRequest;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.ext.unit.Async;
import io.vertx.ext.unit.TestContext;
import io.vertx.ext.unit.junit.VertxUnitRunner;
import io.vertx.ext.web.Router;
import io.vertx.micrometer.MetricsDomain;
import io.vertx.micrometer.MicrometerMetricsOptions;
import io.vertx.micrometer.VertxPrometheusOptions;
import io.vertx.micrometer.backends.BackendRegistries;
import org.junit.After;
import org.junit.Test;
import org.junit.runner.RunWith;

import java.util.function.Consumer;

import static org.assertj.core.api.Assertions.assertThat;

@RunWith(VertxUnitRunner.class)
public class PrometheusMetricsITest {

  private Vertx vertx;

  @After
  public void tearDown() {
    BackendRegistries.stop(MicrometerMetricsOptions.DEFAULT_REGISTRY_NAME);
  }

  @Test
  public void shouldStartEmbeddedServer(TestContext context) {
    vertx = Vertx.vertx(new VertxOptions()
      .setMetricsOptions(new MicrometerMetricsOptions()
        .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true)
          .setStartEmbeddedServer(true)
          .setEmbeddedServerOptions(new HttpServerOptions().setPort(9090)))
        .setEnabled(true)));

    Async async = context.async();
    tryConnect(vertx, context, 9090, "localhost", "/metrics", body -> {
      context.verify(v -> assertThat(body)
        .contains("vertx_http_server_connections{local=\"0.0.0.0:9090\",remote=\"_\",} 1.0"));
      async.complete();
    }, 0);
    async.awaitSuccess(10000);
  }

  @Test
  public void shouldBindExistingServer(TestContext context) {
    vertx = Vertx.vertx(new VertxOptions()
      .setMetricsOptions(new MicrometerMetricsOptions()
        .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true))
        .setEnabled(true)));

    Router router = Router.router(vertx);
    router.route("/custom").handler(routingContext -> {
      PrometheusMeterRegistry prometheusRegistry = (PrometheusMeterRegistry) BackendRegistries.getDefaultNow();
      String response = prometheusRegistry.scrape();
      routingContext.response().end(response);
    });
    vertx.createHttpServer().requestHandler(router::accept).listen(8081);

    Async async = context.async();
    HttpClientRequest req = vertx.createHttpClient()
      .get(8081, "localhost", "/custom")
      .handler(res -> {
        context.assertEquals(200, res.statusCode());
        res.bodyHandler(body -> {
          context.verify(v -> assertThat(body.toString())
            .contains("vertx_http_"));
          async.complete();
        });
      });
    req.end();
    async.awaitSuccess(10000);
  }

  @Test
  public void shouldExcludeCategory(TestContext context) {
    vertx = Vertx.vertx(new VertxOptions()
      .setMetricsOptions(new MicrometerMetricsOptions()
        .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true)
          .setStartEmbeddedServer(true)
          .setEmbeddedServerOptions(new HttpServerOptions().setPort(9090)))
        .addDisabledMetricsCategory(MetricsDomain.HTTP_SERVER)
        .setEnabled(true)));

    Async async = context.async();
    tryConnect(vertx, context, 9090, "localhost", "/metrics", body -> {
      context.verify(v -> assertThat(body)
        .contains("vertx_http_client_connections{local=\"?\",remote=\"localhost:9090\",} 1.0")
        .doesNotContain("vertx_http_server_connections{local=\"0.0.0.0:9090\",remote=\"_\",} 1.0"));
      async.complete();
    }, 0);
    async.awaitSuccess(10000);
  }

  @Test
  public void shouldExposeEventBusMetrics(TestContext context) {
    vertx = Vertx.vertx(new VertxOptions()
      .setMetricsOptions(new MicrometerMetricsOptions()
        .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true)
          .setStartEmbeddedServer(true)
          .setEmbeddedServerOptions(new HttpServerOptions().setPort(9090)))
        .setEnabled(true)));

    // Send something on the eventbus and wait til it's received
    Async asyncEB = context.async();
    vertx.eventBus().consumer("test-eb", msg -> asyncEB.complete());
    vertx.eventBus().publish("test-eb", "test message");
    asyncEB.await(2000);

    // Read metrics on HTTP endpoint for eventbus metrics
    Async async = context.async();
    HttpClientRequest req = vertx.createHttpClient()
      .get(9090, "localhost", "/metrics")
      .handler(res -> {
        context.assertEquals(200, res.statusCode());
        res.bodyHandler(body -> {
          String str = body.toString();
          context.verify(v -> assertThat(str)
            .contains("vertx_eventbus_published_total{address=\"test-eb\",side=\"local\",} 1.0",
              "vertx_eventbus_received_total{address=\"test-eb\",side=\"local\",} 1.0",
              "vertx_eventbus_handlers{address=\"test-eb\",} 1.0",
              "vertx_eventbus_delivered_total{address=\"test-eb\",side=\"local\",} 1.0",
              "vertx_eventbus_processingTime_seconds_count{address=\"test-eb\",} 1.0"));
          async.complete();
        });
      });
    req.end();
    async.awaitSuccess(10000);
  }

  private static void tryConnect(Vertx vertx, TestContext context, int port, String host, String requestURI, Consumer<String> bodyReader, int attempt) {
    HttpClientRequest req = vertx.createHttpClient()
      .get(port, host, requestURI)
      .handler(res -> {
        context.assertEquals(200, res.statusCode());
        res.bodyHandler(body -> bodyReader.accept(body.toString()));
      })
      .exceptionHandler(e -> {
        if (attempt < 10) {
          System.out.println(e);
          try {
            Thread.sleep(500);
          } catch (InterruptedException e1) {
            e1.printStackTrace();
          }
          System.out.println("retrying...");
          tryConnect(vertx, context, port, host, requestURI, bodyReader, attempt + 1);
        } else {
          System.out.println("aborting");
        }
      });
    req.end();
  }
}
