code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.integrations.source.e2e_test;
  
  import com.fasterxml.jackson.databind.JsonNode;
  import io.airbyte.commons.json.Jsons;
  import io.airbyte.commons.resources.MoreResources;
  import io.airbyte.commons.string.Strings;
  import io.airbyte.commons.util.MoreIterators;
  import io.airbyte.protocol.models.AirbyteCatalog;
  import io.airbyte.protocol.models.AirbyteStream;
  import io.airbyte.validation.json.JsonSchemaValidator;
  import io.airbyte.validation.json.JsonValidationException;
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.Set;
  
  @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
  public class ContinuousFeedConfig {
  
    private static final JsonNode JSON_SCHEMA_DRAFT_07;
    private static final JsonSchemaValidator SCHEMA_VALIDATOR = new JsonSchemaValidator();
  
    static {
      try {
        final String jsonSchemaDraft07 = MoreResources.readResource("json_schema_draft_07.json");
        JSON_SCHEMA_DRAFT_07 = Jsons.deserialize(jsonSchemaDraft07);
      } catch (final IOException e) {
        throw new RuntimeException(e);
      }
    }
  
    public enum MockCatalogType {
      SINGLE_STREAM,
      MULTI_STREAM
    }
  
    private final long seed;
    private final AirbyteCatalog mockCatalog;
    private final long maxMessages;
    private final Optional<Long> messageIntervalMs;
  
    public ContinuousFeedConfig(final JsonNode config) throws JsonValidationException {
      this.seed = parseSeed(config);
      this.mockCatalog = parseMockCatalog(config);
      this.maxMessages = parseMaxMessages(config);
      this.messageIntervalMs = parseMessageIntervalMs(config);
    }
  
    static long parseSeed(final JsonNode config) {
      if (!config.has("seed")) {
        return System.currentTimeMillis();
      }
      return config.get("seed").asLong();
    }
  
    static AirbyteCatalog parseMockCatalog(final JsonNode config) throws JsonValidationException {
      final JsonNode mockCatalogConfig = config.get("mock_catalog");
      final MockCatalogType mockCatalogType = MockCatalogType.valueOf(mockCatalogConfig.get("type").asText());
      switch (mockCatalogType) {
        case SINGLE_STREAM -> {
          final String streamName = mockCatalogConfig.get("stream_name").asText();
          final String streamSchemaText = mockCatalogConfig.get("stream_schema").asText();
          final int streamDuplication = mockCatalogConfig.has("stream_duplication")
              ? mockCatalogConfig.get("stream_duplication").asInt()
              : 1;
          final Optional<JsonNode> streamSchema = Jsons.tryDeserialize(streamSchemaText);
          if (streamSchema.isEmpty()) {
            throw new JsonValidationException(String.format("Stream \"%s\" has invalid schema: %s", streamName, streamSchemaText));
          }
          checkSchema(streamName, streamSchema.get());
  
          if (streamDuplication == 1) {
            final AirbyteStream stream = new AirbyteStream().withName(streamName).withJsonSchema(streamSchema.get());
            return new AirbyteCatalog().withStreams(Collections.singletonList(stream));
          } else {
            final List<AirbyteStream> streams = new ArrayList<>(streamDuplication);
            for (int i = 0; i < streamDuplication; ++i) {
              streams.add(new AirbyteStream()
                  .withName(String.join("_", streamName, String.valueOf(i)))
                  .withJsonSchema(streamSchema.get()));
            }
            return new AirbyteCatalog().withStreams(streams);
          }
        }
        case MULTI_STREAM -> {
          final String streamSchemasText = mockCatalogConfig.get("stream_schemas").asText();
          final Optional<JsonNode> streamSchemas = Jsons.tryDeserialize(streamSchemasText);
          if (streamSchemas.isEmpty()) {
            throw new JsonValidationException("Input stream schemas are invalid: %s" + streamSchemasText);
          }
  
          final List<Entry<String, JsonNode>> streamEntries = MoreIterators.toList(streamSchemas.get().fields());
          final List<AirbyteStream> streams = new ArrayList<>(streamEntries.size());
          for (final Map.Entry<String, JsonNode> entry : streamEntries) {
            final String streamName = entry.getKey();
            final JsonNode streamSchema = entry.getValue();
            checkSchema(streamName, streamSchema);
            streams.add(new AirbyteStream().withName(streamName).withJsonSchema(streamSchema));
          }
          return new AirbyteCatalog().withStreams(streams);
        }
        default -> throw new IllegalArgumentException("Unsupported mock catalog type: " + mockCatalogType);
      }
    }
  
    /**
     * Validate the stream schema against Json schema draft 07.
     */
    private static void checkSchema(final String streamName, final JsonNode streamSchema) throws JsonValidationException {
      final Set<String> validationMessages = SCHEMA_VALIDATOR.validate(JSON_SCHEMA_DRAFT_07, streamSchema);
      if (!validationMessages.isEmpty()) {
        throw new JsonValidationException(String.format(
            "Stream \"%s\" has invalid schema.\n- Errors: %s\n- Schema: %s",
            streamName,
            Strings.join(validationMessages, "; "),
            streamSchema.toString()));
      }
    }
  
    static long parseMaxMessages(final JsonNode config) {
      return config.get("max_messages").asLong();
    }
  
    static Optional<Long> parseMessageIntervalMs(final JsonNode config) {
      if (config.has("message_interval_ms")) {
        final long messageIntervalMs = config.get("message_interval_ms").asLong();
        if (messageIntervalMs > 0) {
          return Optional.of(messageIntervalMs);
        }
      }
      return Optional.empty();
    }
  
    public long getSeed() {
      return seed;
    }
  
    public AirbyteCatalog getMockCatalog() {
      return mockCatalog;
    }
  
    public long getMaxMessages() {
      return maxMessages;
    }
  
    public Optional<Long> getMessageIntervalMs() {
      return messageIntervalMs;
    }
  
    @Override
    public String toString() {
      return String.format("%s{maxMessages=%d, seed=%d, messageIntervalMs=%s, mockCatalog=%s}",
          ContinuousFeedConfig.class.getSimpleName(),
          maxMessages,
          seed,
          messageIntervalMs.toString(),
          mockCatalog.toString());
    }
  
    @Override
    public boolean equals(final Object other) {
      if (!(other instanceof final ContinuousFeedConfig that)) {
        return false;
      }
      return this.maxMessages == that.maxMessages
          && this.seed == that.seed
          && this.messageIntervalMs.equals(that.messageIntervalMs)
          && this.mockCatalog.equals(that.mockCatalog);
    }
  
    @Override
    public int hashCode() {
      return Objects.hash(seed, maxMessages, messageIntervalMs, mockCatalog);
    }
  
  }
  