text: /*******************************************************************************
   * Copyright (c) 2019, 2020 Eurotech and/or its affiliates and others
   *
   * All rights reserved. This program and the accompanying materials
   * are made available under the terms of the Eclipse Public License v1.0
   * which accompanies this distribution, and is available at
   * http://www.eclipse.org/legal/epl-v10.html
   *
   * Contributors:
   *     Eurotech
   *******************************************************************************/
  package org.eclipse.kapua.qa.integration.steps;
  
  import com.google.common.base.Strings;
  import com.google.common.collect.Lists;
  import com.spotify.docker.client.DefaultDockerClient;
  import com.spotify.docker.client.DockerClient;
  import com.spotify.docker.client.exceptions.DockerCertificateException;
  import com.spotify.docker.client.exceptions.DockerException;
  import com.spotify.docker.client.messages.ContainerConfig;
  import com.spotify.docker.client.messages.ContainerCreation;
  import com.spotify.docker.client.messages.HostConfig;
  import com.spotify.docker.client.messages.Image;
  import com.spotify.docker.client.messages.NetworkConfig;
  import com.spotify.docker.client.messages.NetworkCreation;
  import com.spotify.docker.client.messages.PortBinding;
  import cucumber.api.java.Before;
  import cucumber.api.java.en.And;
  import cucumber.api.java.en.Given;
  import cucumber.api.java.en.Then;
  import cucumber.runtime.java.guice.ScenarioScoped;
  import org.apache.activemq.command.BrokerInfo;
  import org.eclipse.kapua.qa.common.StepData;
  import org.eclipse.paho.client.mqttv3.MqttException;
  import org.junit.Assert;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import javax.inject.Inject;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  
  @ScenarioScoped
  public class DockerSteps {
  
      private static final Logger logger = LoggerFactory.getLogger(DockerSteps.class);
  
      private static final String NETWORK_PREFIX = "kapua-net";
  
      private DockerClient docker;
  
      private NetworkConfig networkConfig;
  
      private String networkId;
  
      private boolean debug;
  
      private List<String> envVar;
  
      private Map<String, String> containerMap;
  
      public Map<String, Integer> portMap;
  
      public Map<String, BrokerInfo> brokerMap;
  
      private ContainerConfig dbContainerConfig;
  
      private StepData stepData;
  
      private static final String ALL_IP = "0.0.0.0";
  
      @Inject
      public DockerSteps(StepData stepData) {
  
          this.stepData = stepData;
          containerMap = new HashMap<>();
      }
  
      @Given("^Enable debug$")
      public void enableDebug() {
          this.debug = true;
      }
  
      @Given("^Disable debug$")
      public void disableDebug() {
          this.debug = false;
      }
  
      @Before
      public void setupDockerClient() {
          logger.info("Creating docker client.");
          try {
              docker = DefaultDockerClient.fromEnv().build();
          } catch (DockerCertificateException e) {
              logger.error("Could not connect to docker.");
              throw new RuntimeException("Cannot initialize docker client!", e);
          }
      }
  
      @Given("^Create mqtt \"(.*)\" client for broker \"(.*)\" on port (\\d+) with user \"(.*)\" and pass \"(.*)\"$")
      public void createMqttClient(String clientId, String broker, int port, String user, String pass) {
          try {
              BrokerClient client = new BrokerClient(broker, port, clientId, user, pass);
              stepData.put(clientId, client);
          } catch (MqttException e) {
              logger.error("Error creating mqtt client with id " + clientId, e);
          }
      }
  
      @Given("^Connect to mqtt client \"(.*)\"$")
      public void connectMqttClient(String clientId) {
          BrokerClient client = (BrokerClient) stepData.get(clientId);
          try {
              client.connect();
          } catch (MqttException e) {
              logger.error("Unable to connect to mqtt broker with client " + clientId, e);
              e.printStackTrace();
          }
      }
  
      @Given("^Disconnect mqtt client \"(.*)\"$")
      public void disconnectMqttClient(String clientId) {
          BrokerClient client = (BrokerClient) stepData.get(clientId);
          try {
              client.disconnect();
          } catch (MqttException e) {
              logger.error("Unable to disconnect from mqtt broker with client " + clientId, e);
          }
      }
  
      @Given("^Subscribe mqtt client \"(.*)\" to topic \"(.*)\"$")
      public void subscribeMqttClient(String clientId, String topic) {
          BrokerClient client = (BrokerClient) stepData.get(clientId);
          try {
              client.subscribe(topic, 1);
          } catch (MqttException e) {
              logger.error("Can not subscribe with client " + clientId);
          }
      }
  
      @Then("^Client \"(.*)\" has (\\d+) messages?.*$")
      public void clientCountMsg(String clientId, int numMsgs) {
          BrokerClient client = (BrokerClient) stepData.get(clientId);
          int receivedMsgs = client.getRecivedMsgCnt();
          Assert.assertEquals(numMsgs, receivedMsgs);
      }
  
      @Given("^Publish string \"(.*)\" to topic \"(.*)\" as client \"(.*)\"")
      public void publishMqttClient(String message, String topic, String clientId) {
          BrokerClient client = (BrokerClient) stepData.get(clientId);
          try {
              client.publish(topic, 1, message);
          } catch (MqttException e) {
              logger.error("Can not publish to topic " + topic);
          }
      }
  
      @Given("^Create network$")
      public void createNetwork() throws DockerException, InterruptedException {
          networkConfig = NetworkConfig.builder().name(NETWORK_PREFIX).build();
          NetworkCreation networkCreation = docker.createNetwork(networkConfig);
          networkId = networkCreation.id();
      }
  
      @Given("^Remove network$")
      public void removeNetwork() throws DockerException, InterruptedException {
          docker.removeNetwork(networkId);
      }
  
      @Given("^Pull image \"(.*)\"$")
      public void pullImage(String image) throws DockerException, InterruptedException {
          docker.pull(image);
      }
  
      @Given("^List images by name \"(.*)\"$")
      public void listImages(String imageName) throws Exception {
          List<Image> images = docker.listImages(DockerClient.ListImagesParam.byName(imageName));
          if ((images != null) && (images.size() > 0)) {
              for (Image image : images) {
                  logger.info("Image: " + image);
              }
          } else {
              logger.info("No docker images found.");
          }
      }
  
      @And("^Start DB container with name \"(.*)\"$")
      public void startDBContainer(String name) throws DockerException, InterruptedException {
          logger.info("Starting DB container...");
          ContainerConfig dbConfig = getDbContainerConfig();
          ContainerCreation dbContainerCreation = docker.createContainer(dbConfig, name);
          String containerId = dbContainerCreation.id();
  
          docker.startContainer(containerId);
          docker.connectToNetwork(containerId, networkId);
          containerMap.put("db", containerId);
          logger.info("DB container started: {}", containerId);
      }
  
      @And("^Start ES container with name \"(.*)\"$")
      public void startESContainer(String name) throws DockerException, InterruptedException {
          logger.info("Starting ES container...");
          ContainerConfig esConfig = getEsContainerConfig();
          ContainerCreation esContainerCreation = docker.createContainer(esConfig, name);
          String containerId = esContainerCreation.id();
  
          docker.startContainer(containerId);
          docker.connectToNetwork(containerId, networkId);
          containerMap.put("es", containerId);
          logger.info("ES container started: {}", containerId);
      }
  
      @And("^Start EventBroker container with name \"(.*)\"$")
      public void startEBContainer(String name) throws DockerException, InterruptedException {
          logger.info("Starting EventBroker container...");
          ContainerConfig ebConfig = getEventBrokerContainerConfig();
          ContainerCreation ebContainerCreation = docker.createContainer(ebConfig, name);
          String containerId = ebContainerCreation.id();
  
          docker.startContainer(containerId);
          docker.connectToNetwork(containerId, networkId);
          containerMap.put(name, containerId);
          logger.info("EventBroker container started: {}", containerId);
      }
  
      @And("^Start Message Broker container$")
      public void startEBContainer(List<BrokerConfigData> brokerConfigDataList) throws DockerException, InterruptedException {
          BrokerConfigData bcData = brokerConfigDataList.get(0);
          logger.info("Starting Message Broker container {}...", bcData.getName());
          ContainerConfig mbConfig = getBrokerContainerConfig(
                  bcData.getBrokerAddress(), bcData.getBrokerIp(), bcData.getClusterName(), null,
                  bcData.getMqttPort(), bcData.getMqttHostPort(), bcData.getMqttsPort(), bcData.getMqttsHostPort(),
                  bcData.getWebPort(), bcData.getWebHostPort(), bcData.getDebugPort(), bcData.getDebugHostPort(),
                  bcData.getBrokerInternalDebugPort(), bcData.getDockerImage());
          ContainerCreation mbContainerCreation = docker.createContainer(mbConfig, bcData.getName());
          String containerId = mbContainerCreation.id();
  
          docker.startContainer(containerId);
          docker.connectToNetwork(containerId, networkId);
          containerMap.put(bcData.getName(), containerId);
          logger.info("Message Broker {} container started: {}", bcData.getName(), containerId);
      }
  
      @Then("^Stop container with name \"(.*)\"$")
      public void stopContainer(String name) throws DockerException, InterruptedException {
          logger.info("Stopping container {}...", name);
          String containerId = containerMap.get(name);
          docker.stopContainer(containerId, 3);
          logger.info("Container {} stopped.", name);
      }
  
      @Then("^Remove container with name \"(.*)\"$")
      public void removeContainer(String name) throws DockerException, InterruptedException {
          logger.info("Removing container {}...", name);
          String containerId = containerMap.get(name);
          docker.removeContainer(containerId);
          logger.info("Container {} removed.", name);
      }
  
      /**
       * Creation of docker container configuration for broker.
       *
       * @param brokerAddr
       * @param brokerIp
       * @param clusterName
       * @param controlMessageForwarding
       * @param mqttPort                 mqtt port on docker
       * @param mqttHostPort             mqtt port on docker host
       * @param mqttsPort                mqtts port on docker
       * @param mqttsHostPort            mqtts port on docker host
       * @param webPort                  web port on docker
       * @param webHostPort              web port on docker host
       * @param debugPort                debug port on docker
       * @param debugHostPort            debug port on docker host
       * @param brokerInternalDebugPort
       * @param dockerImage              full name of image (e.g. "kapua/kapua-broker:1.3.0-SNAPSHOT")
       * @return Container configuration for specific boroker instance
       */
      private ContainerConfig getBrokerContainerConfig(String brokerAddr, String brokerIp,
              String clusterName,
              String controlMessageForwarding,
              int mqttPort, int mqttHostPort,
              int mqttsPort, int mqttsHostPort,
              int webPort, int webHostPort,
              int debugPort, int debugHostPort,
              int brokerInternalDebugPort,
              String dockerImage) {
  
          final Map<String, List<PortBinding>> portBindings = new HashMap<>();
          addHostPort(ALL_IP, portBindings, mqttPort, mqttHostPort);
          addHostPort(ALL_IP, portBindings, mqttsPort, mqttsHostPort);
          addHostPort(ALL_IP, portBindings, webPort, webHostPort);
          addHostPort(ALL_IP, portBindings, debugPort, debugHostPort);
  
          final HostConfig hostConfig = HostConfig.builder().portBindings(portBindings).build();
  
          List<String> envVars = Lists.newArrayList("commons.db.schema.update=true",
                  "commons.db.connection.host=db",
                  "commons.db.connection.port=3306",
                  "datastore.elasticsearch.nodes=es",
                  "datastore.elasticsearch.port=9200",
                  "datastore.client.class=org.eclipse.kapua.service.datastore.client.rest.RestDatastoreClient",
                  "commons.eventbus.url=failover:(amqp://events-broker:5672)?jms.sendTimeout=1000",
                  "certificate.jwt.private.key=file:///var/opt/activemq/key.pk8",
                  "certificate.jwt.certificate=file:///var/opt/activemq/cert.pem",
                  String.format("broker.ip=%s", brokerIp));
          if (envVar != null) {
              envVars.addAll(envVar);
          }
  
          if (debug) {
              envVars.add(String.format("ACTIVEMQ_DEBUG_OPTS=-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=%s", brokerInternalDebugPort));
          }
  
          if (!Strings.isNullOrEmpty(clusterName)) {
              envVars.add(String.format("cluster.name=%s", clusterName));
          }
  
          if (!Strings.isNullOrEmpty(controlMessageForwarding)) {
              envVars.add(String.format("cluster.control_message_forwarding=%s", controlMessageForwarding));
          }
  
          String[] ports = {
                  String.valueOf(mqttPort),
                  String.valueOf(mqttsPort),
                  String.valueOf(webPort),
                  String.valueOf(debugPort)
          };
  
          return ContainerConfig.builder()
                  .hostConfig(hostConfig)
                  .exposedPorts(ports)
                  .env(envVars)
                  .image(dockerImage)
                  .build();
      }
  
      /**
       * Creation of docker container configuration for H2 database.
       *
       * @return Container configuration for database instance.
       */
      private ContainerConfig getDbContainerConfig() {
          final int dbPort = 3306;
          final Map<String, List<PortBinding>> portBindings = new HashMap<>();
          addHostPort(ALL_IP, portBindings, dbPort, dbPort);
          final HostConfig hostConfig = HostConfig.builder().portBindings(portBindings).build();
  
          return ContainerConfig.builder()
                  .hostConfig(hostConfig)
                  .exposedPorts(String.valueOf(dbPort))
                  .env(
                          "DATABASE=kapuadb",
                          "DB_USER=kapua",
                          "DB_PASSWORD=kapua",
                          "DB_PORT_3306_TCP_PORT=3306"
                  )
                  .image("kapua/kapua-sql:1.3.0-SNAPSHOT")
                  .build();
      }
  
      /**
       * Creation of docker container configuration for Elasticsearch.
       *
       * @return Container configuration for Elasticsearch instance.
       */
      private ContainerConfig getEsContainerConfig() {
          final int esPortRest = 9200;
          final int esPortNodes = 9300;
          final Map<String, List<PortBinding>> portBindings = new HashMap<>();
          addHostPort(ALL_IP, portBindings, esPortRest, esPortRest);
          addHostPort(ALL_IP, portBindings, esPortNodes, esPortNodes);
          final HostConfig hostConfig = HostConfig.builder().portBindings(portBindings).build();
  
          return ContainerConfig.builder()
                  .hostConfig(hostConfig)
                  .exposedPorts(String.valueOf(esPortRest), String.valueOf(esPortNodes))
                  .image("elasticsearch:5.4.0")
                  .cmd(
                          "-Ecluster.name=kapua-datastore",
                          "-Ediscovery.type=single-node",
                          "-Etransport.host=0.0.0.0 ",
                          "-Etransport.ping_schedule=-1 ",
                          "-Etransport.tcp.connect_timeout=30s"
                  )
                  .build();
      }
  
      /**
       * Creation of docker container configuration for event broker.
       *
       * @return Container configuration for event broker instance.
       */
      private ContainerConfig getEventBrokerContainerConfig() {
          final int brokerPort = 5672;
          final Map<String, List<PortBinding>> portBindings = new HashMap<>();
          addHostPort(ALL_IP, portBindings, brokerPort, brokerPort);
          final HostConfig hostConfig = HostConfig.builder().portBindings(portBindings).build();
  
          return ContainerConfig.builder()
                  .hostConfig(hostConfig)
                  .exposedPorts(String.valueOf(brokerPort))
                  .image("kapua/kapua-events-broker:1.3.0-SNAPSHOT")
                  .build();
      }
  
      /**
       * Add docker port to host port mapping.
       *
       * @param host         ip address of host
       * @param portBindings list ob bindings that gets updated
       * @param port         docker port
       * @param hostPort     port on host
       */
      private void addHostPort(String host, Map<String, List<PortBinding>> portBindings,
              int port, int hostPort) {
  
          List<PortBinding> hostPorts = new ArrayList<>();
          hostPorts.add(PortBinding.of(host, hostPort));
          portBindings.put(String.valueOf(port), hostPorts);
      }
  
  }
  