/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.metrics;

import static java.nio.file.Files.createDirectories;
import static java.util.Arrays.asList;
import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPorts;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.compiler.ClassBuilder.writeJarFromClasses;
import static org.assertj.core.api.Assertions.assertThat;

import java.io.IOException;
import java.nio.file.Path;
import java.util.Collection;

import io.micrometer.core.instrument.Counter;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.rules.ServiceJarRule;
import org.apache.geode.test.junit.categories.MetricsTest;
import org.apache.geode.test.junit.rules.FolderRule;
import org.apache.geode.test.junit.rules.gfsh.GfshRule;

@Category(MetricsTest.class)
public class GatewayReceiverMetricsTest {

  private static final String SENDER_LOCATOR_NAME = "sender-locator";
  private static final String RECEIVER_LOCATOR_NAME = "receiver-locator";
  private static final String SENDER_SERVER_NAME = "sender-server";
  private static final String RECEIVER_SERVER_NAME = "receiver-server";
  private static final String REGION_NAME = "region";
  private static final String GFSH_COMMAND_SEPARATOR = " ";

  private Path rootFolder;
  private Path senderLocatorFolder;
  private Path receiverLocatorFolder;
  private Path senderServerFolder;
  private Path receiverServerFolder;
  private int receiverLocatorPort;
  private int senderLocatorPort;

  @Rule(order = 0)
  public FolderRule folderRule = new FolderRule();
  @Rule(order = 1)
  public GfshRule gfshRule = new GfshRule(folderRule::getFolder);
  @Rule
  public ServiceJarRule serviceJarRule = new ServiceJarRule();

  @Before
  public void startClusters() throws IOException {
    rootFolder = folderRule.getFolder().toPath().toAbsolutePath();
    senderLocatorFolder = createDirectories(rootFolder.resolve(SENDER_LOCATOR_NAME));
    receiverLocatorFolder = createDirectories(rootFolder.resolve(RECEIVER_LOCATOR_NAME));
    senderServerFolder = createDirectories(rootFolder.resolve(SENDER_SERVER_NAME));
    receiverServerFolder = createDirectories(rootFolder.resolve(RECEIVER_SERVER_NAME));

    int[] ports = getRandomAvailableTCPPorts(6);

    receiverLocatorPort = ports[0];
    senderLocatorPort = ports[1];
    int senderServerPort = ports[2];
    int receiverServerPort = ports[3];
    int senderLocatorJmxPort = ports[4];
    int receiverLocatorJmxPort = ports[5];

    int senderSystemId = 2;
    int receiverSystemId = 1;

    String startSenderLocatorCommand = String.join(GFSH_COMMAND_SEPARATOR,
        "start locator",
        "--name=" + SENDER_LOCATOR_NAME,
        "--dir=" + senderLocatorFolder,
        "--port=" + senderLocatorPort,
        "--locators=localhost[" + senderLocatorPort + "]",
        "--http-service-port=0",
        "--J=-Dgemfire.remote-locators=localhost[" + receiverLocatorPort + "]",
        "--J=-Dgemfire.distributed-system-id=" + senderSystemId,
        "--J=-Dgemfire.jmx-manager-start=true",
        "--J=-Dgemfire.jmx-manager-port=" + senderLocatorJmxPort);

    String startReceiverLocatorCommand = String.join(GFSH_COMMAND_SEPARATOR,
        "start locator",
        "--name=" + RECEIVER_LOCATOR_NAME,
        "--dir=" + receiverLocatorFolder,
        "--port=" + receiverLocatorPort,
        "--locators=localhost[" + receiverLocatorPort + "]",
        "--http-service-port=0",
        "--J=-Dgemfire.remote-locators=localhost[" + senderLocatorPort + "]",
        "--J=-Dgemfire.distributed-system-id=" + receiverSystemId,
        "--J=-Dgemfire.jmx-manager-start=true ",
        "--J=-Dgemfire.jmx-manager-port=" + receiverLocatorJmxPort);

    String startSenderServerCommand = String.join(GFSH_COMMAND_SEPARATOR,
        "start server",
        "--name=" + SENDER_SERVER_NAME,
        "--dir=" + senderServerFolder,
        "--locators=localhost[" + senderLocatorPort + "]",
        "--server-port=" + senderServerPort,
        "--J=-Dgemfire.distributed-system-id=" + senderSystemId);

    Path serviceJarPath = serviceJarRule.createJarFor("metrics-publishing-service.jar",
        MetricsPublishingService.class, SimpleMetricsPublishingService.class);

    String startReceiverServerCommand = String.join(GFSH_COMMAND_SEPARATOR,
        "start server",
        "--name=" + RECEIVER_SERVER_NAME,
        "--dir=" + receiverServerFolder,
        "--locators=localhost[" + receiverLocatorPort + "]",
        "--server-port=" + receiverServerPort,
        "--classpath=" + serviceJarPath,
        "--J=-Dgemfire.distributed-system-id=" + receiverSystemId);

    gfshRule.execute(startSenderLocatorCommand, startReceiverLocatorCommand,
        startSenderServerCommand, startReceiverServerCommand);

    String gatewaySenderId = "gs";

    String connectToSenderLocatorCommand = "connect --locator=localhost[" + senderLocatorPort + "]";

    String startGatewaySenderCommand = String.join(GFSH_COMMAND_SEPARATOR,
        "create gateway-sender",
        "--id=" + gatewaySenderId,
        "--parallel=false",
        "--remote-distributed-system-id=" + receiverSystemId);

    String createSenderRegionCommand = String.join(GFSH_COMMAND_SEPARATOR,
        "create region",
        "--name=" + REGION_NAME,
        "--type=" + RegionShortcut.REPLICATE.name(),
        "--gateway-sender-id=" + gatewaySenderId);

    gfshRule.execute(connectToSenderLocatorCommand, startGatewaySenderCommand,
        createSenderRegionCommand);

    String connectToReceiverLocatorCommand =
        "connect --locator=localhost[" + receiverLocatorPort + "]";
    String startGatewayReceiverCommand = "create gateway-receiver";
    String createReceiverRegionCommand = String.join(GFSH_COMMAND_SEPARATOR,
        "create region",
        "--name=" + REGION_NAME,
        "--type=" + RegionShortcut.REPLICATE.name());

    gfshRule.execute(connectToReceiverLocatorCommand, startGatewayReceiverCommand,
        createReceiverRegionCommand);

    // Deploy function to members
    Path functionJarPath =
        newJarForFunctionClass(GetEventsReceivedCountFunction.class, "function.jar");
    String deployCommand = "deploy --jar=" + functionJarPath;
    String listFunctionsCommand = "list functions";

    gfshRule.execute(connectToReceiverLocatorCommand, deployCommand, listFunctionsCommand);
  }

  @After
  public void stopClusters() {
    String stopReceiverServerCommand = "stop server --dir=" + receiverServerFolder;
    String stopSenderServerCommand = "stop server --dir=" + senderServerFolder;
    String stopReceiverLocatorCommand = "stop locator --dir=" + receiverLocatorFolder;
    String stopSenderLocatorCommand = "stop locator --dir=" + senderLocatorFolder;

    gfshRule.execute(stopReceiverServerCommand, stopSenderServerCommand,
        stopReceiverLocatorCommand, stopSenderLocatorCommand);
  }

  @Test
  public void whenPerformingOperations_thenGatewayReceiverEventsReceivedIncreases() {
    String connectToSenderLocatorCommand = "connect --locator=localhost[" + senderLocatorPort + "]";

    String doPutCommand = String.join(GFSH_COMMAND_SEPARATOR,
        "put",
        "--region=" + REGION_NAME,
        "--key=foo",
        "--value=bar");

    String doRemoveCommand = String.join(GFSH_COMMAND_SEPARATOR,
        "remove",
        "--region=" + REGION_NAME,
        "--key=foo");

    String doCreateRegionCommand = String.join(GFSH_COMMAND_SEPARATOR,
        "create region",
        "--name=blah",
        "--type=" + RegionShortcut.REPLICATE.name());

    gfshRule.execute(connectToSenderLocatorCommand, doPutCommand, doRemoveCommand,
        doCreateRegionCommand);

    String connectToReceiverLocatorCommand =
        "connect --locator=localhost[" + receiverLocatorPort + "]";
    String executeFunctionCommand = "execute function --id=" + GetEventsReceivedCountFunction.ID;

    Collection<String> gatewayEventsExpectedToReceive = asList(doPutCommand, doRemoveCommand);

    await().untilAsserted(() -> {
      String output = gfshRule
          .execute(connectToReceiverLocatorCommand, executeFunctionCommand)
          .getOutputText();

      assertThat(output.trim())
          .as("Returned count of events received.")
          .endsWith("[" + gatewayEventsExpectedToReceive.size() + ".0]");
    });
  }

  private Path newJarForFunctionClass(Class<?> clazz, String jarName) throws IOException {
    Path jarPath = rootFolder.resolve(jarName);
    writeJarFromClasses(jarPath.toFile(), clazz);
    return jarPath;
  }

  public static class GetEventsReceivedCountFunction implements Function<Void> {
    private static final String ID = "GetEventsReceivedCountFunction";

    @Override
    public void execute(FunctionContext<Void> context) {
      Counter eventsReceivedCounter = SimpleMetricsPublishingService.getRegistry()
          .find("geode.gateway.receiver.events")
          .counter();

      Object result = eventsReceivedCounter == null
          ? "Meter not found."
          : eventsReceivedCounter.count();

      context.getResultSender().lastResult(result);
    }

    @Override
    public String getId() {
      return ID;
    }
  }
}
