package org.talend.components.common.runtime;

import com.csvreader.CsvWriter;
import org.apache.avro.Schema;
import org.apache.avro.generic.IndexedRecord;
import org.talend.components.api.component.runtime.*;
import org.talend.components.api.container.RuntimeContainer;
import org.talend.components.common.BulkFileProperties;
import org.talend.daikon.avro.AvroRegistry;
import org.talend.daikon.avro.IndexedRecordAdapterFactory;

import java.io.File;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.List;

/**
 *  Generate bulk file
 */
public class BulkFileWriter implements Writer<WriterResult> {

    protected RuntimeContainer container;

    private WriteOperation<WriterResult> writeOperation;

    private Sink sink;

    protected BulkFileProperties bulkProperties;

    private String uId;

    private CsvWriter csvWriter;

    private char separator = ',';

    private String charset = "UTF-8";

    private boolean isAppend;

    private transient IndexedRecordAdapterFactory<Object, ? extends IndexedRecord> factory;

    protected int dataCount;

    public BulkFileWriter(WriteOperation<WriterResult> writeOperation, BulkFileProperties bulkProperties, RuntimeContainer container) {
        this.writeOperation = writeOperation;
        this.container = container;
        this.sink = writeOperation.getSink();
        this.bulkProperties = bulkProperties;
        this.isAppend = bulkProperties.append.getBooleanValue();
    }

    @Override
    public void open(String uId) throws IOException {
        this.uId = uId;
        File file = new File(bulkProperties.bulkFilePath.getStringValue());
        file.getParentFile().mkdirs();
        csvWriter = new CsvWriter(new OutputStreamWriter(
                new java.io.FileOutputStream(file, isAppend), charset),separator);

        Schema schema = new Schema.Parser().parse(bulkProperties.schema.schema.getStringValue());

        if(!isAppend){
            csvWriter.writeRecord(getHeaders(schema));
        }

    }

    @Override
    public void write(Object datum) throws IOException {
        if (null == datum) {
            return;
        }else {
            List<String> values = getValues(datum);
            csvWriter.writeRecord(values.toArray(new String[values.size()]));
            dataCount++;
        }
    }

    public void flush() throws IOException {
        csvWriter.flush();
    }

    @Override
    public WriterResult close() throws IOException {
        flush();
        csvWriter.close();
        return new WriterResult(uId, dataCount);
    }

    @Override
    public WriteOperation<WriterResult> getWriteOperation() {
        return writeOperation;
    }

    public String[] getHeaders(Schema schema){
        List<String> headers = new ArrayList<String>();
        for(Schema.Field f :schema.getFields()){
            headers.add(f.name());
        }
        return headers.toArray(new String[headers.size()]);
    }

    public List<String> getValues(Object datum){
        IndexedRecord input = getFactory(datum).convertToAvro(datum);
        List<String> values = new ArrayList<String>();
        for (Schema.Field f : input.getSchema().getFields()) {
            if(input.get(f.pos()) != null){
                values.add(String.valueOf(input.get(f.pos())));
            }else{
                values.add("");
            }
        }
        return values;
    }

    public IndexedRecordAdapterFactory<Object, ? extends IndexedRecord> getFactory(Object datum){
        if (null == factory) {
            factory = (IndexedRecordAdapterFactory<Object, ? extends IndexedRecord>) new AvroRegistry()
                    .createAdapterFactory(datum.getClass());
        }
        return factory;
    }
}

@@ -422,16 +422,62 @@
 Factory;
+%0Aimport org.talend.daikon.avro.util.AvroUtils;
 %0A%0Aimport
@@ -1901,16 +1901,256 @@
         
+%0A%0A    %7D%0A%0A    private boolean headerIsReady = false;%0A    %0A    @Override%0A    public void write(Object datum) throws IOException %7B%0A        if (null == datum) %7B%0A            return;%0A        %7D%0A        %0A    %09if(!isAppend && !headerIsReady)%7B%0A    %09%09
 Schema s
@@ -2235,84 +2235,273 @@
 ));%0A
-%0A
     
+%09%09%0A
     
-if(!isAppend)%7B%0A            csvWriter.writeRecord(getHeaders(s
+%09%09if (AvroUtils.isIncludeAllFields(schema) && (datum instanceof org.apache.avro.generic.GenericData.Record)) %7B%0A    %09%09%09org.apache.avro.generic.GenericData.Record record = (org.apache.avro.generic.GenericData.Record)datum;%0A    %09%09%09schema = record.getS
 chema
-)
+(
 );%0A 
@@ -2511,136 +2511,111 @@
     
-%7D%0A%0A
     %7D%0A
-%0A
     
-@Override%0A    public void write(Object datum) throws IOException %7B%0A        if (null == datum) %7B%0A            return
+    %09%0A            csvWriter.writeRecord(getHeaders(schema));%0A            headerIsReady = true
 ;%0A  
@@ -2625,19 +2625,15 @@
    %7D
-else %7B
 %0A    
+%09%0A
     
@@ -2672,28 +2672,24 @@
 ues(datum);%0A
-    
         csvW
@@ -2758,20 +2758,16 @@
         
-    
 dataCoun
@@ -2771,26 +2771,16 @@
 ount++;%0A
-        %7D%0A
     %7D%0A%0A 

fix the dynamic for tsalesforceouputbulk https://jira.talendforge.org/browse/TUP-4480