package org.talend.components.filedelimited.runtime;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.avro.Schema;
import org.apache.avro.generic.IndexedRecord;
import org.apache.commons.lang3.StringUtils;
import org.talend.components.api.exception.DataRejectException;
import org.talend.components.filedelimited.FileDelimitedProperties;
import org.talend.components.filedelimited.tfileinputdelimited.TFileInputDelimitedProperties;
import org.talend.daikon.avro.converter.AvroConverter;
import org.talend.daikon.avro.converter.IndexedRecordConverter;

public class DelimitedAdaptorFactory implements IndexedRecordConverter<String[], IndexedRecord> {

    Schema schema;

    FileDelimitedProperties properties;

    @Override
    public Class<String[]> getDatumClass() {
        return String[].class;
    }

    @Override
    public String[] convertToDatum(IndexedRecord value) {
        return null;
    }

    @Override
    public DelimitedIndexedRecord convertToAvro(String[] values) {
        return new DelimitedIndexedRecord(values);
    }

    @Override
    public Schema getSchema() {
        return schema;
    }

    @Override
    public void setSchema(Schema schema) {
        this.schema = schema;
    }

    public FileDelimitedProperties getProperties() {
        return properties;
    }

    public void setProperties(FileDelimitedProperties properties) {
        this.properties = properties;
    }

    private class DelimitedIndexedRecord implements IndexedRecord {

        String[] values;

        public DelimitedIndexedRecord(String[] values) {
            this.values = values;
        }

        @Override
        public void put(int index, Object json) {
            throw new UnsupportedOperationException();
        }

        private boolean trimValues[];

        AvroConverter[] fieldConverter = null;

        @Override
        public Object get(int index) {
            // Lazy initialization of the cached converter objects.
            Object value = null;
            if (trimValues == null) {
                if (properties == null) {
                    throw new IllegalArgumentException("Runtime properties for converter is not be set!");
                }
                trimValues = new boolean[getSchema().getFields().size()];
                fieldConverter = new AvroConverter[trimValues.length];
                Object trimSelect = ((TFileInputDelimitedProperties) properties).trimColumns.trimTable.trim.getValue();
                boolean trimAll = ((TFileInputDelimitedProperties) properties).trimColumns.trimAll.getValue();
                for (int j = 0; j < trimValues.length; j++) {
                    Schema.Field f = getSchema().getFields().get(j);
                    fieldConverter[j] = new FileDelimitedAvroRegistry().getConverter(f, properties);
                    if (trimAll) {
                        trimValues[j] = true;
                    } else if (trimSelect != null && (trimSelect instanceof List) && (j < ((List<Boolean>) trimSelect).size())) {
                        trimValues[j] = ((List<Boolean>) trimSelect).get(j);
                    }

                }

            }
            if (index < values.length) {
                try {
                    if (trimValues[index] && !StringUtils.isEmpty(values[index])) {
                        values[index] = values[index].trim();
                    }
                    value = fieldConverter[index].convertToAvro(values[index]);
                } catch (Exception e) {
                    if (((TFileInputDelimitedProperties) properties).dieOnError.getValue()) {
                        throw e;
                    } else {
                        Map<String, Object> resultMessage = new HashMap<String, Object>();
                        resultMessage.put("errorMessage", e.getMessage());
                        resultMessage.put("talend_record", this);
                        throw new DataRejectException(resultMessage);
                    }
                }
            }
            return value;
        }

        @Override
        public Schema getSchema() {
            return DelimitedAdaptorFactory.this.getSchema();
        }

    }

}

@@ -743,24 +743,111 @@
 roperties;%0A%0A
+    private boolean trimValues%5B%5D;%0A%0A    private AvroConverter%5B%5D fieldConverter = null;%0A%0A
     @Overrid
@@ -1536,24 +1536,1050 @@
 ies;%0A    %7D%0A%0A
+    private void initializeConverters() %7B%0A        if (properties == null) %7B%0A            throw new IllegalArgumentException(%22Runtime properties for converter is not be set!%22);%0A        %7D%0A        trimValues = new boolean%5BgetSchema().getFields().size()%5D;%0A        fieldConverter = new AvroConverter%5BtrimValues.length%5D;%0A        Object trimSelect = ((TFileInputDelimitedProperties) properties).trimColumns.trimTable.trim.getValue();%0A        boolean trimAll = ((TFileInputDelimitedProperties) properties).trimColumns.trimAll.getValue();%0A        for (int j = 0; j %3C trimValues.length; j++) %7B%0A            Schema.Field f = getSchema().getFields().get(j);%0A            fieldConverter%5Bj%5D = new FileDelimitedAvroRegistry().getConverter(f, properties);%0A            if (trimAll) %7B%0A                trimValues%5Bj%5D = true;%0A            %7D else if (trimSelect != null && (trimSelect instanceof List) && (j %3C ((List%3CBoolean%3E) trimSelect).size())) %7B%0A                trimValues%5Bj%5D = ((List%3CBoolean%3E) trimSelect).get(j);%0A            %7D%0A%0A        %7D%0A    %7D%0A%0A%0A
     private 
@@ -2901,95 +2901,8 @@
  %7D%0A%0A
-        private boolean trimValues%5B%5D;%0A%0A        AvroConverter%5B%5D fieldConverter = null;%0A%0A
     
@@ -3026,41 +3026,8 @@
 ts.%0A
-            Object value = null;%0A
     
@@ -3081,1109 +3081,78 @@
    i
-f (properties == null) %7B%0A                    throw new IllegalArgumentException(%22Runtime properties for converter is not be set!%22);%0A                %7D%0A                trimValues = new boolean%5BgetSchema().getFields().size()%5D;%0A                fieldConverter = new AvroConverter%5BtrimValues.length%5D;%0A                Object trimSelect = ((TFileInputDelimitedProperties) properties).trimColumns.trimTable.trim.getValue();%0A                boolean trimAll = ((TFileInputDelimitedProperties) properties).trimColumns.trimAll.getValue();%0A                for (int j = 0; j %3C trimValues.length; j++) %7B%0A                    Schema.Field f = getSchema().getFields().get(j);%0A                    fieldConverter%5Bj%5D = new FileDelimitedAvroRegistry().getConverter(f, properties);%0A                    if (trimAll) %7B%0A                        trimValues%5Bj%5D = true;%0A                    %7D else if (trimSelect != null && (trimSelect instanceof List) && (j %3C ((List%3CBoolean%3E) trimSelect).size())) %7B%0A                        trimValues%5Bj%5D = ((List%3CBoolean%3E) trimSelect).get(j);%0A                    %7D%0A%0A                %7D%0A%0A            %7D
+nitializeConverters();%0A            %7D%0A%0A            Object value = null;
 %0A   

Fix initialization code for avro converters. (#269)