ExportMasterListCsvJobConfig

package com.my.pricing.onlinestore.export.job;

import java.io.File;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.TrueFileFilter;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.JobScope;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.classify.BackToBackPatternClassifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementSetter;

import com.myer.pricing.onlinestore.export.DivisionClassifier;
import com.myer.pricing.onlinestore.export.constants.ExecutionContextConstants;
import com.myer.pricing.onlinestore.export.constants.FlatFileConstants;
import com.myer.pricing.onlinestore.export.constants.JobParamConstants;
import com.myer.pricing.onlinestore.export.job.common.CommonExportFromStagingDbJobConfig;
import com.myer.pricing.onlinestore.export.job.writer.MasterListClassifierCompositeItemWriter;
import com.myer.pricing.onlinestore.export.listener.CustomStepExecutionListener;
import com.myer.pricing.onlinestore.export.model.MasterList;
import com.myer.pricing.onlinestore.export.model.PromotionStartDateEndDateGrouping;


/**
 * Master List export job from the online pricing staging db.
 * http://stackoverflow.com/questions/29286699/repeating-a-step-x-times-in-spring-batch
 *
 * @author Richard Riviere
 *
 */
@Configuration
@Import(CommonExportFromStagingDbJobConfig.class)
public class ExportMasterListCsvJobConfig {

    public static final String JOB_NAME = "exportMasterListCsv";

    @Autowired
    private CommonExportFromStagingDbJobConfig commonJobConfig;

    @Bean
    public Job exportMasterListCsvJob(
            @Qualifier("createJobDatesStep") Step createJobDatesStep,
            @Qualifier("readStgDbAndExportMasterListStep") Step readStgDbAndExportMasterListStep,
            @Qualifier("renameFilesStep") Step renameFilesStep) {

        return commonJobConfig.jobBuilderFactory.get(JOB_NAME)
                .incrementer(new RunIdIncrementer())
                .start(createJobDatesStep)
                .next(readStgDbAndExportMasterListStep)
                    .on("CONTINUE").
                        to(readStgDbAndExportMasterListStep).
                    on("FINISHED").end()
                .end()
                .build();

    }

    @Bean
    public Step createJobDatesStep(
            @Qualifier("createJobDatesTasklet")  Tasklet createJobDatesTasklet) {
        return  commonJobConfig.stepBuilderFactory
                    .get("createJobDatesStep")
                    .tasklet(createJobDatesTasklet)
                    .build();
    }


    @Bean
    @JobScope
    public Tasklet createJobDatesTasklet(
            @Value("${exportMasterListCsv.generatePromoStartDateEndDateGroupings.promoStartDateEndDateSql}") String promoStartDateEndDateSql,
            @Value("#{jobParameters['" + JobParamConstants.PARAM_FROM_DATE + "']}") Date jobFromDate,
            @Value("#{jobParameters['" + JobParamConstants.PARAM_TO_DATE + "']}") Date jobToDate) {

        return new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                JdbcTemplate jdbcTemplate = new JdbcTemplate(commonJobConfig.onlineStagingDb);

                List<PromotionStartDateEndDateGrouping> promotionStartDateEndDateGroupings =
                        jdbcTemplate.query(
                                promoStartDateEndDateSql,
                                new PreparedStatementSetter() {
                                    public void setValues(PreparedStatement preparedStatement) throws SQLException {
                                        preparedStatement.setDate(1, new java.sql.Date(jobFromDate.getTime()));
                                    }
                                },
                                BeanPropertyRowMapper.newInstance(PromotionStartDateEndDateGrouping.class));

                System.err.println("createJobDatesTasklet: " + promotionStartDateEndDateGroupings);

                chunkContext
                .getStepContext()
                .getStepExecution()
                .getJobExecution()
                .getExecutionContext()
                .put(ExecutionContextConstants.JOB_EXPORT_STAGING_PROMOTION_START_DATE_END_DATE_GROUPINGS, promotionStartDateEndDateGroupings);
                return RepeatStatus.FINISHED;
            }
        };
    }


    @Bean
    public Step readStgDbAndExportMasterListStep (
            @Qualifier("queryStagingDbReader") JdbcCursorItemReader<MasterList> queryStagingDbReader,
            @Qualifier("masterListOutputProcessor") ItemProcessor<MasterList,MasterList> masterListOutputProcessor,
            @Qualifier("masterListFileWriter") ItemWriter<MasterList> masterListFileWriter,
            @Qualifier("division10MasterListFileWriter") FlatFileItemWriter<MasterList> division10MasterListFileWriter,
            @Qualifier("division15MasterListFileWriter") FlatFileItemWriter<MasterList> division15MasterListFileWriter,
            @Qualifier("division20MasterListFileWriter") FlatFileItemWriter<MasterList> division20MasterListFileWriter,
            @Qualifier("division25MasterListFileWriter") FlatFileItemWriter<MasterList> division25MasterListFileWriter,
            @Qualifier("division30MasterListFileWriter") FlatFileItemWriter<MasterList> division30MasterListFileWriter,
            @Qualifier("division35MasterListFileWriter") FlatFileItemWriter<MasterList> division35MasterListFileWriter,
            @Qualifier("division40MasterListFileWriter") FlatFileItemWriter<MasterList> division40MasterListFileWriter,
            @Qualifier("division45MasterListFileWriter") FlatFileItemWriter<MasterList> division45MasterListFileWriter,
            @Qualifier("division50MasterListFileWriter") FlatFileItemWriter<MasterList> division50MasterListFileWriter,
            @Qualifier("division55MasterListFileWriter") FlatFileItemWriter<MasterList> division55MasterListFileWriter,
            @Qualifier("division60MasterListFileWriter") FlatFileItemWriter<MasterList> division60MasterListFileWriter,
            @Qualifier("division65MasterListFileWriter") FlatFileItemWriter<MasterList> division65MasterListFileWriter,
            @Qualifier("division70MasterListFileWriter") FlatFileItemWriter<MasterList> division70MasterListFileWriter,
            @Qualifier("division75MasterListFileWriter") FlatFileItemWriter<MasterList> division75MasterListFileWriter,
            @Qualifier("division80MasterListFileWriter") FlatFileItemWriter<MasterList> division80MasterListFileWriter,
            @Qualifier("division85MasterListFileWriter") FlatFileItemWriter<MasterList> division85MasterListFileWriter,
            @Qualifier("division90MasterListFileWriter") FlatFileItemWriter<MasterList> division90MasterListFileWriter,
            @Qualifier("division95MasterListFileWriter") FlatFileItemWriter<MasterList> division95MasterListFileWriter,
            @Qualifier("stepListener") CustomStepExecutionListener stepListener) {

        return  commonJobConfig.stepBuilderFactory
                    .get("readStgDbAndExportMasterListStep")
                    .<MasterList,MasterList>chunk(commonJobConfig.chunkSize)
                    .reader(queryStagingDbReader)
                    .processor(masterListOutputProcessor)
                    .writer(masterListFileWriter)
                    .stream(division10MasterListFileWriter)
                    .stream(division15MasterListFileWriter)
                    .stream(division20MasterListFileWriter)
                    .stream(division25MasterListFileWriter)
                    .stream(division30MasterListFileWriter)
                    .stream(division35MasterListFileWriter)
                    .stream(division40MasterListFileWriter)
                    .stream(division45MasterListFileWriter)
                    .stream(division50MasterListFileWriter)
                    .stream(division55MasterListFileWriter)
                    .stream(division60MasterListFileWriter)
                    .stream(division65MasterListFileWriter)
                    .stream(division70MasterListFileWriter)
                    .stream(division75MasterListFileWriter)
                    .stream(division80MasterListFileWriter)
                    .stream(division85MasterListFileWriter)
                    .stream(division90MasterListFileWriter)
                    .stream(division95MasterListFileWriter)
                    .listener(stepListener)
                    .build();
    }


    /**
     * Do some extra processing on master list rows.
     *
     * @return master list row with transformations applied.
     */
    @Bean
    public ItemProcessor<MasterList,MasterList> masterListOutputProcessor() {

        return new ItemProcessor<MasterList, MasterList>() {

            @Override
            public MasterList process(MasterList masterList) throws Exception {
                masterList.setPromoIdPromoCompId(masterList.getPromoId() + " / " + masterList.getPromoCompId());
                return masterList;
            }
        };
    }

    /**
     * Write out the the masterlist.csv
     *
     * @param masterListFile
     * @param executionContext
     * @return
     */
    @Bean
    @StepScope
    public ItemWriter<MasterList> masterListFileWriter(
            @Qualifier("masterListClassifierCompositeItemWriter") MasterListClassifierCompositeItemWriter masterListClassifierCompositeItemWriter,
            @Qualifier("division10MasterListFileWriter") FlatFileItemWriter<MasterList> division10MasterListFileWriter,
            @Qualifier("division15MasterListFileWriter") FlatFileItemWriter<MasterList> division15MasterListFileWriter,
            @Qualifier("division20MasterListFileWriter") FlatFileItemWriter<MasterList> division20MasterListFileWriter,
            @Qualifier("division25MasterListFileWriter") FlatFileItemWriter<MasterList> division25MasterListFileWriter,
            @Qualifier("division30MasterListFileWriter") FlatFileItemWriter<MasterList> division30MasterListFileWriter,
            @Qualifier("division35MasterListFileWriter") FlatFileItemWriter<MasterList> division35MasterListFileWriter,
            @Qualifier("division40MasterListFileWriter") FlatFileItemWriter<MasterList> division40MasterListFileWriter,
            @Qualifier("division45MasterListFileWriter") FlatFileItemWriter<MasterList> division45MasterListFileWriter,
            @Qualifier("division50MasterListFileWriter") FlatFileItemWriter<MasterList> division50MasterListFileWriter,
            @Qualifier("division55MasterListFileWriter") FlatFileItemWriter<MasterList> division55MasterListFileWriter,
            @Qualifier("division60MasterListFileWriter") FlatFileItemWriter<MasterList> division60MasterListFileWriter,
            @Qualifier("division65MasterListFileWriter") FlatFileItemWriter<MasterList> division65MasterListFileWriter,
            @Qualifier("division70MasterListFileWriter") FlatFileItemWriter<MasterList> division70MasterListFileWriter,
            @Qualifier("division75MasterListFileWriter") FlatFileItemWriter<MasterList> division75MasterListFileWriter,
            @Qualifier("division80MasterListFileWriter") FlatFileItemWriter<MasterList> division80MasterListFileWriter,
            @Qualifier("division85MasterListFileWriter") FlatFileItemWriter<MasterList> division85MasterListFileWriter,
            @Qualifier("division90MasterListFileWriter") FlatFileItemWriter<MasterList> division90MasterListFileWriter,
            @Qualifier("division95MasterListFileWriter") FlatFileItemWriter<MasterList> division95MasterListFileWriter) {

        System.err.println("masterListFileWriter:");

        BackToBackPatternClassifier classifier = new BackToBackPatternClassifier();
        classifier.setRouterDelegate(new DivisionClassifier());
        classifier.setMatcherMap(new HashMap<String, ItemWriter<? extends MasterList>>() {{
            put("10", division10MasterListFileWriter);
            put("15", division15MasterListFileWriter);
            put("20", division20MasterListFileWriter);
            put("25", division25MasterListFileWriter);
            put("30", division30MasterListFileWriter);
            put("35", division35MasterListFileWriter);
            put("40", division40MasterListFileWriter);
            put("45", division45MasterListFileWriter);
            put("50", division50MasterListFileWriter);
            put("55", division55MasterListFileWriter);
            put("60", division60MasterListFileWriter);
            put("65", division65MasterListFileWriter);
            put("70", division70MasterListFileWriter);
            put("75", division75MasterListFileWriter);
            put("80", division80MasterListFileWriter);
            put("85", division85MasterListFileWriter);
            put("90", division90MasterListFileWriter);
            put("95", division95MasterListFileWriter);
        }});

        /*ClassifierCompositeItemWriter<MasterList> writer = new ClassifierCompositeItemWriter<MasterList>();
        writer.setClassifier(classifier);*/
        masterListClassifierCompositeItemWriter.setClassifier(classifier);
        return masterListClassifierCompositeItemWriter;
    }


    @Bean
    public Step renameFilesStep(
            @Qualifier("renameFilesTasklet")  Tasklet renameFilesTasklet) {
        return  commonJobConfig.stepBuilderFactory
                    .get("renameFilesStep")
                    .tasklet(renameFilesTasklet)
                    .build();
    }

    @Bean
    @JobScope
    public Tasklet renameFilesTasklet(
            @Value("#{jobExecutionContext['" + ExecutionContextConstants.JOB_EXPORT_STAGING_WORKING_PROMOTION_START_DATE_END_DATE +"']}")
            PromotionStartDateEndDateGrouping promotionStartDateEndDateGrouping) {
        return new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.err.println("commonJobConfig.outDir: " + commonJobConfig.outDir);

                File dir = new File(commonJobConfig.outDir);
                List<File> files = (List<File>) FileUtils.listFiles(dir, TrueFileFilter.INSTANCE, TrueFileFilter.INSTANCE);


                String sep = System.getProperty("file.separator");

                final String dateTimeFormatPattern = "ddMMyyyy";
                final DateFormat format = new SimpleDateFormat(dateTimeFormatPattern);
                String promoCompDetailStartDateStr = format.format(promotionStartDateEndDateGrouping.getRpmPromoCompDetailStartDate());
                String promoCompDetailEndDateStr = format.format(promotionStartDateEndDateGrouping.getRpmPromoCompDetailEndDate());

                for (File file : files) {

                    String absolutePath = file.getAbsolutePath();
                    String filePath = absolutePath.substring(0,absolutePath.lastIndexOf(File.separator));

                    System.err.println("file: " + filePath + sep + file.getName());
                    File destinationFile = new File(
                            filePath
                            + sep
                            + file.getName()
                            + FlatFileConstants.FILENAME_WORD_SEPERATOR
                            + FlatFileConstants.UPLOAD_KEYWORD
                            + FlatFileConstants.FILENAME_WORD_SEPERATOR
                            + promoCompDetailStartDateStr
                            + "-"
                            + promoCompDetailEndDateStr
                            + FlatFileConstants.FILENAME_WORD_SEPERATOR
                            + FlatFileConstants.FILENAME_MSTRLST
                            + FlatFileConstants.EXTENSION_CSV );
                    file.renameTo(destinationFile);

                }
                return RepeatStatus.FINISHED;
            }
        };
    }

    @Bean
    @StepScope
    public MasterListClassifierCompositeItemWriter masterListClassifierCompositeItemWriter(){
           return new MasterListClassifierCompositeItemWriter();
    }

    @Bean
    @JobScope
    public CustomStepExecutionListener stepListener(){
           return new CustomStepExecutionListener();
    }



}