package com.marklogic.hub.job;

import com.marklogic.client.document.DocumentWriteOperation;
import com.marklogic.client.helper.DatabaseClientProvider;
import com.marklogic.spring.batch.columnmap.ColumnMapSerializer;
import com.marklogic.spring.batch.columnmap.DefaultStaxColumnMapSerializer;
import com.marklogic.spring.batch.columnmap.XmlStringColumnMapSerializer;
import com.marklogic.spring.batch.item.processor.ColumnMapProcessor;
import com.marklogic.spring.batch.item.reader.AllTablesItemReader;
import com.marklogic.spring.batch.item.writer.MarkLogicItemWriter;
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.JobScope;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;

import javax.sql.DataSource;
import java.util.Map;

@EnableBatchProcessing
@Import(value = {com.marklogic.spring.batch.config.MarkLogicBatchConfiguration.class })
public class SqlDbToHubJobConfig {

    private final String JOB_NAME = "sqlDbToHubJob";

    @Bean
    public Job job(JobBuilderFactory jobBuilderFactory, Step step) {
        return jobBuilderFactory.get(JOB_NAME).start(step).build();
    }

    @Bean
    @JobScope
    public Step step(
        StepBuilderFactory stepBuilderFactory,
        DataSource dataSource,
        DatabaseClientProvider databaseClientProvider,
        @Value("#{jobParameters['output_collections']}") String[] collections) {

        AllTablesItemReader reader = new AllTablesItemReader(dataSource);

        // Processor - this is a very basic implementation for converting a column map to an XML string
        ColumnMapSerializer serializer = new XmlStringColumnMapSerializer();
        ItemProcessor processor = new AllTablesColumnMapProcessor(serializer);

        ItemWriter<DocumentWriteOperation> writer = new MarkLogicItemWriter(databaseClientProvider.getDatabaseClient());

        return stepBuilderFactory.get("step1")
            .<Map<String, Object>, DocumentWriteOperation>chunk(10)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .build();
    }

}

@@ -25,16 +25,60 @@
 b.job;%0A%0A
+import com.marklogic.client.DatabaseClient;%0A
 import c
@@ -130,16 +130,16 @@
 ration;%0A
-
 import c
@@ -536,24 +536,89 @@
 ItemReader;%0A
+import com.marklogic.spring.batch.item.writer.DataHubItemWriter;%0A
 import com.m
@@ -1975,52 +1975,148 @@
 rs%5B'
-output_collections'%5D%7D%22) String%5B%5D collections
+entity'%5D%7D%22) String entity,%0A        @Value(%22#%7BjobParameters%5B'flow'%5D%7D%22) String flow,%0A        @Value(%22#%7BjobParameters%5B'job_id'%5D%7D%22) String jobId
 ) %7B%0A
@@ -2456,24 +2456,114 @@
 );%0A%0A        
+// Uncomment to push data directly into MarkLogic without going through DataHub%0A        //
 ItemWriter%3CD
@@ -2660,24 +2660,223 @@
 Client());%0A%0A
+        ItemWriter writer = new DataHubItemWriter(databaseClientProvider.getDatabaseClient(),%0A            entity,%0A            DataHubItemWriter.FlowType.INPUT,%0A            flow,%0A            jobId);%0A%0A
         retu

Add DataHubItemWriter to batch job