content: /*
   * 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.camel.kafkaconnector.transforms;
  
  
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  import java.util.stream.Collectors;
  
  import org.apache.kafka.common.config.ConfigDef;
  import org.apache.kafka.connect.connector.ConnectRecord;
  import org.apache.kafka.connect.data.Field;
  import org.apache.kafka.connect.data.Schema;
  import org.apache.kafka.connect.data.Struct;
  import org.apache.kafka.connect.transforms.Transformation;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import static org.apache.kafka.connect.transforms.util.Requirements.requireMapOrNull;
  import static org.apache.kafka.connect.transforms.util.Requirements.requireStructOrNull;
  
  public abstract class FieldsToHeadersTransform<R extends ConnectRecord<R>> implements Transformation<R> {
  
      private static final Logger LOG = LoggerFactory.getLogger(FieldsToHeadersTransform.class);
  
      private static final String PURPOSE = "fields extraction to headers";
      private static final String FIELDS_CONFIG = "fields";
      private static final String HEADERS_CONFIG = "headers";
  
      public static final ConfigDef CONFIG_DEF = new ConfigDef()
              .define(FIELDS_CONFIG, ConfigDef.Type.LIST, new ArrayList<>(), ConfigDef.Importance.MEDIUM, "Fields names to extract and set to headers")
              .define(HEADERS_CONFIG, ConfigDef.Type.LIST, ConfigDef.NO_DEFAULT_VALUE, ConfigDef.Importance.MEDIUM, "Headers names to set with extracted fields");
  
      private List<String> fields;
  
      private List<String> headers;
  
      protected abstract Schema operatingSchema(R record);
  
      protected abstract Object operatingValue(R record);
  
      @Override
      public R apply(R r) {
          RecordValue value = createRecordValue(r);
          Schema currentSchema;
          Object currentValue;
          if (fields.isEmpty()) {
              currentSchema = value.getFieldSchema("");
              currentValue = value.getFieldValue("");
              r.headers().add(headers.get(0), currentValue, currentSchema);
          } else {
              for (int i = 0; i < fields.size(); i++) {
                  currentSchema = value.getFieldSchema(fields.get(i));
                  currentValue = value.getFieldValue(fields.get(i));
                  r.headers().add(headers.get(i), currentValue, currentSchema);
              }
          }
          return r;
      }
  
      @Override
      public ConfigDef config() {
          return CONFIG_DEF;
      }
  
      @Override
      public void close() {
      }
  
      @Override
      public void configure(Map<String, ?> map) {
          Map<String, Object> parsedConfig = CONFIG_DEF.parse(map);
          fields =  (List<String>) parsedConfig.getOrDefault(FIELDS_CONFIG, new ArrayList<>());
          headers = (List<String>) parsedConfig.getOrDefault(HEADERS_CONFIG, new ArrayList<>());
          validateConfig();
      }
  
      private void validateConfig() {
  
          boolean validFields  = fields.stream().allMatch(nef -> nef != null);
          boolean validHeaders = headers.stream().allMatch(nef -> nef != null && !nef.trim().isEmpty());
  
          if (!(validFields && validHeaders)) {
              throw new IllegalArgumentException("fields configuration property cannot be null (can be an empty string if you want the whole value/key), headers configuration property cannot be null or contain empty elements.");
          }
          if (fields.size() != 0 && fields.size() > headers.size()) {
              String fieldsWithoutCorrespondingHeaders = fields.subList(headers.size(), fields.size()).stream().collect(Collectors.joining(","));
              throw new IllegalArgumentException("There is no corresponding header(s) configured for the following field(s): " + fieldsWithoutCorrespondingHeaders);
          }
          if (fields.size() != 0 && headers.size() > fields.size()) {
              String headersWithoutCorrespondingFields = headers.subList(fields.size(), headers.size()).stream().collect(Collectors.joining(","));
              LOG.warn("There is no corresponding header(s) for the following field(s): {} ", headersWithoutCorrespondingFields);
          }
          if (fields.size() == 0 && headers.size() > 1) {
              LOG.warn("Fields are empty and there are more than 1 header it means whole value/key will put in the first header of this list: {} ", headers.stream().collect(Collectors.joining(",")));
          }
      }
  
      private RecordValue createRecordValue(R r) {
          final Schema schema = operatingSchema(r);
          if (fields.isEmpty()) {
              return new WholeRecordValue(operatingValue(r), schema);
          }
          if (schema == null) {
              return new MapRecordValue(requireMapOrNull(operatingValue(r), PURPOSE));
          }
          return new StructRecordValue(requireStructOrNull(operatingValue(r), PURPOSE), schema);
      }
  
      public static class Key<R extends ConnectRecord<R>> extends FieldsToHeadersTransform<R> {
  
          @Override
          protected Schema operatingSchema(R record) {
              return record.keySchema();
          }
  
          @Override
          protected Object operatingValue(R record) {
              return record.key();
          }
      }
  
      public static class Value<R extends ConnectRecord<R>> extends FieldsToHeadersTransform<R> {
  
          @Override
          protected Schema operatingSchema(R record) {
              return record.valueSchema();
          }
  
          @Override
          protected Object operatingValue(R record) {
              return record.value();
          }
      }
  
      public interface RecordValue {
  
          Object getFieldValue(String fieldName);
  
          Schema getFieldSchema(String fieldName);
      }
  
      public class WholeRecordValue implements RecordValue {
          private Object value;
          private Schema schema;
  
          public WholeRecordValue(Object value, Schema schema) {
              this.value = value;
              this.schema = schema;
          }
  
          public Object getFieldValue(String fieldName) {
              return value;
          }
  
          public Schema getFieldSchema(String fieldName) {
              return schema;
          }
      }
  
      public class MapRecordValue implements RecordValue {
  
          private Map<String, Object> map;
  
          public MapRecordValue(Map<String, Object> map) {
              this.map = map;
          }
  
          public Object getFieldValue(String fieldName) {
              return map == null ? null : map.get(fieldName);
          }
  
          public Schema getFieldSchema(String fieldName) {
              return null;
          }
      }
  
      public class StructRecordValue implements RecordValue {
  
          private Struct struct;
  
          private Schema schema;
  
          public StructRecordValue(Struct struct, Schema schema) {
              this.struct = struct;
              this.schema = schema;
          }
  
          public Object getFieldValue(String fieldName) {
              return struct.get(fieldName);
          }
  
          public Schema getFieldSchema(String fieldName) {
              Field field = schema.field(fieldName);
              if (field == null) {
                  throw new IllegalArgumentException("Unknown field: " + fieldName);
              }
              return field.schema();
          }
      }
  }
  