package me.snikit.config;

import javax.sql.DataSource;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.transform.LineAggregator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;
import org.springframework.jdbc.core.BeanPropertyRowMapper;

import me.snikit.domains.Customer;

@Configuration
@EnableBatchProcessing
@EnableAutoConfiguration
public class BatchConfig {

	@Autowired
	public JobBuilderFactory jobBuilderFactory;

	@Autowired
	public StepBuilderFactory stepBuilderFactory;

	@Autowired
	DataSource dataSource;

	@Bean
	ItemWriter<Customer> writer() {
		FlatFileItemWriter<Customer> csvFileWriter = new FlatFileItemWriter<>();

		String exportFileHeader = "customerNumber|addressLine1|addressLine2|city|contactFirstName|contactLastName|country|creditLimit|customerName|phone|postalCode|state";
		StringHeaderWriter headerWriter = new StringHeaderWriter(exportFileHeader);
		csvFileWriter.setHeaderCallback(headerWriter);

		String exportFilePath = "D:/a/students.csv";
		csvFileWriter.setResource(new FileSystemResource(exportFilePath));

		LineAggregator<Customer> lineAggregator = WriterConfig.createLineAggregator();
		csvFileWriter.setLineAggregator(lineAggregator);

		return csvFileWriter;
	}

	private static final String SQL_QUERY = "SELECT * FROM customers";

	@Bean
	ItemReader<Customer> reader(DataSource dataSource) {
		JdbcCursorItemReader<Customer> databaseReader = new JdbcCursorItemReader<>();
		databaseReader.setDataSource(dataSource);
		databaseReader.setSql(SQL_QUERY);
		databaseReader.setRowMapper(new BeanPropertyRowMapper<>(Customer.class));
		databaseReader.setFetchSize(100);
		return databaseReader;
	}

	@Bean
	public Step step1(ItemWriter<Customer> writer, ItemReader<Customer> reader) {
		return stepBuilderFactory.get("step1").<Customer, Customer> chunk(10).reader(reader).writer(writer).build();
	}

	@Bean
	public Job job(Step step1) {
		return jobBuilderFactory.get("job").incrementer(new RunIdIncrementer()).flow(step1).end().build();
	}

}
spring.datasource.url=jdbc:mysql://localhost/classicmodels
spring.datasource.username=snikit
spring.datasource.password=maxpain
spring.jpa.hibernate.ddl-auto=update
spring.jpa.show-sql=true
spring.datasource.validationQuery = SELECT 1
spring.jpa.properties.hibernate.dialect = org.hibernate.dialect.MySQL5Dialect