/**
 * 
 */
package org.springblade.etl.connector.source;

import java.util.Map;

import javax.validation.constraints.NotEmpty;
import javax.validation.constraints.NotNull;

import org.springblade.etl.connector.base.SeatunnelConnector;
import org.springblade.etl.connector.source.config.SeatunnelKafkaSourceConfig;
import org.springblade.etl.connector.source.enums.FieldType;
import org.springblade.etl.connector.source.enums.KafkaFormat;
import org.springblade.etl.connector.source.enums.KafkaStartMode;
import org.springblade.etl.util.EtlJsonUtil;
import org.springblade.etl.util.ValidateUtil;
import org.springframework.stereotype.Component;

import io.swagger.annotations.ApiModelProperty;
import lombok.Data;

/**
 * @author wookvn
 *
 */
@Component
public class PlatomixKafkaSource implements SeatunnelConnector<SeatunnelKafkaSourceConfig> {
	
	@Data
	public static class Config {
		@NotEmpty(message = "Kafka服务地址不能为空")
		@ApiModelProperty("kafka服务器地址")
		private String kafkaServers;
		@NotEmpty(message = "Kakfa的Topic不能为空")
		@ApiModelProperty("kafka的topic【多个逗号隔开】")
		private String topic;
		@ApiModelProperty("topic是否正则表达式")
		private Boolean pattern = false;
		@NotNull(message = "Kakfa偏移量模式不能为空")
		@ApiModelProperty("Kakfa偏移量模式")
		private KafkaStartMode offsetMode;
		private Map<String, Long> offsets;
		@NotNull(message = "Kakfa消息格式不能为空")
		@ApiModelProperty("Kakfa消息格式")
		private KafkaFormat format;
		@ApiModelProperty("Kakfa消息文本分隔符")
		private String delimiter;
		@NotEmpty(message = "Kakfa字段不能为空")
		@ApiModelProperty("Kakfa字段信息")
		private Map<String, FieldType> fields;
		@ApiModelProperty("Kakfa配置信息")
		private Map<String, String> config;
	}
	
	@Override
	public SeatunnelKafkaSourceConfig buildConfig(String configJson) {
		Config ptConfig = (Config)validate(configJson);
		SeatunnelKafkaSourceConfig seatunnelConfig = new SeatunnelKafkaSourceConfig();
		seatunnelConfig.setBootstrapServers(ptConfig.getKafkaServers());
		seatunnelConfig.setTopic(ptConfig.getTopic());
		seatunnelConfig.setPattern(ptConfig.getPattern());
		seatunnelConfig.setStartMode(ptConfig.getOffsetMode().name());
		seatunnelConfig.setFieldDelimiter(ptConfig.getDelimiter());
		seatunnelConfig.setFormat(ptConfig.getFormat().name());
		seatunnelConfig.setKafkaConfig(ptConfig.getConfig());
		seatunnelConfig.setSchema(new SeatunnelKafkaSourceConfig.SchemaField(ptConfig.getFields()));
		return seatunnelConfig;
	}
	
	@Override
	public Object validate(String configJson) {
		Config ptConfig = EtlJsonUtil.parseJsonString(configJson, PlatomixKafkaSource.Config.class);
		ValidateUtil.validate(ptConfig);
		return ptConfig;
	}
}
