text: "use strict";
  var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
      function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
      return new (P || (P = Promise))(function (resolve, reject) {
          function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
          function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
          function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
          step((generator = generator.apply(thisArg, _arguments || [])).next());
      });
  };
  Object.defineProperty(exports, "__esModule", { value: true });
  exports.StartedKafkaContainer = exports.KafkaContainer = void 0;
  const generic_container_1 = require("../../generic-container");
  const network_1 = require("../../network");
  const port_client_1 = require("../../port-client");
  const uuid_1 = require("../../uuid");
  const logger_1 = require("../../logger");
  const abstract_started_container_1 = require("../abstract-started-container");
  class KafkaContainer extends generic_container_1.GenericContainer {
      constructor(image = "confluentinc/cp-kafka", tag = "latest", host, zooKeeperImage = "confluentinc/cp-zookeeper", zooKeeperTag = "latest") {
          super(image, tag);
          this.image = image;
          this.tag = tag;
          this.host = host;
          this.zooKeeperImage = zooKeeperImage;
          this.zooKeeperTag = zooKeeperTag;
          this.uuid = new uuid_1.RandomUuid();
          this.portClient = new port_client_1.RandomPortClient();
          this.isZooKeeperProvided = false;
          this.host = host === undefined ? this.uuid.nextUuid() : host;
          this.withName(this.host)
              .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,EXTERNAL_LISTENER:PLAINTEXT,PLAINTEXT:PLAINTEXT")
              .withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER")
              .withEnv("KAFKA_BROKER_ID", "1")
              .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")
              .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false");
      }
      withZooKeeper(host, port) {
          this.isZooKeeperProvided = true;
          this.zooKeeperHost = host;
          this.zooKeeperPort = port;
          return this;
      }
      preCreate(dockerClient, boundPorts) {
          return __awaiter(this, void 0, void 0, function* () {
              const kafkaPort = 9093;
              const kafkaInternalPort = boundPorts.getBinding(9093);
              const kafkaBrokerPort = 9092;
              this.withEnv("KAFKA_LISTENERS", `EXTERNAL_LISTENER://0.0.0.0:${kafkaPort},BROKER://0.0.0.0:${kafkaBrokerPort}`);
              this.withEnv("KAFKA_ADVERTISED_LISTENERS", `EXTERNAL_LISTENER://${dockerClient.getHost()}:${kafkaInternalPort},BROKER://${this.host}:${kafkaBrokerPort}`);
              if (this.isZooKeeperProvided) {
                  this.withEnv("KAFKA_ZOOKEEPER_CONNECT", `${this.zooKeeperHost}:${this.zooKeeperPort}`);
              }
              else {
                  const zooKeeperHost = this.uuid.nextUuid();
                  const zooKeeperPort = yield this.portClient.getPort();
                  this.withEnv("KAFKA_ZOOKEEPER_CONNECT", `${zooKeeperHost}:${zooKeeperPort}`);
                  const zookeeperContainer = yield new generic_container_1.GenericContainer(this.zooKeeperImage, this.zooKeeperTag)
                      .withName(zooKeeperHost)
                      .withEnv("ZOOKEEPER_CLIENT_PORT", zooKeeperPort.toString());
                  if (this.networkMode !== undefined) {
                      zookeeperContainer.withNetworkMode(this.networkMode);
                  }
                  else {
                      this.network = yield new network_1.Network().start();
                      this.withNetworkMode(this.network.getName());
                      zookeeperContainer.withNetworkMode(this.network.getName());
                  }
                  this.zooKeeperContainer = yield zookeeperContainer.start();
              }
          });
      }
      start() {
          const _super = Object.create(null, {
              start: { get: () => super.start }
          });
          return __awaiter(this, void 0, void 0, function* () {
              return new StartedKafkaContainer(yield _super.start.call(this), this.network, this.zooKeeperContainer);
          });
      }
  }
  exports.KafkaContainer = KafkaContainer;
  class StartedKafkaContainer extends abstract_started_container_1.AbstractStartedContainer {
      constructor(startedTestContainer, network, zooKeeperContainer) {
          super(startedTestContainer);
          this.network = network;
          this.zooKeeperContainer = zooKeeperContainer;
      }
      stop(options) {
          const _super = Object.create(null, {
              stop: { get: () => super.stop }
          });
          return __awaiter(this, void 0, void 0, function* () {
              logger_1.log.debug("Stopping Kafka container");
              const stoppedContainer = yield _super.stop.call(this, options);
              if (this.zooKeeperContainer) {
                  logger_1.log.debug("Stopping ZooKeeper container");
                  yield this.zooKeeperContainer.stop(options);
              }
              if (this.network) {
                  logger_1.log.debug("Stopping Kafka network");
                  yield this.network.stop();
              }
              return stoppedContainer;
          });
      }
  }
  exports.StartedKafkaContainer = StartedKafkaContainer;
  