package org.easybatch.extensions.opencsv;

import com.opencsv.CSVWriter;
import org.easybatch.core.field.BeanFieldExtractor;
import org.easybatch.core.field.FieldExtractor;
import org.easybatch.core.marshaller.RecordMarshaller;
import org.easybatch.core.record.Record;
import org.easybatch.core.record.StringRecord;
import org.easybatch.core.util.Utils;

import java.beans.IntrospectionException;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.List;

/**
 * Marshals a POJO to CSV format using <a href="http://opencsv.sourceforge.net">Open CSV</a>.
 * <p/>
 * <strong>This marshaller does not support recursive marshalling.</strong>
 *
 * @author Mahmoud Ben Hassine (mahmoud.benhassine@icloud.com)
 */
public class OpenCsvRecordMarshaller<P> implements RecordMarshaller<Record<P>, StringRecord> {

    public static final char DEFAULT_DELIMITER = ',';
    public static final char DEFAULT_QUALIFIER = '\"';

    private char delimiter;
    private char qualifier;
    private FieldExtractor<P> fieldExtractor;

    /**
     * Create a open CSV record marshaller.
     *
     * @param type   the type of object to marshal
     * @param fields the list of fields to marshal in order
     * @throws IntrospectionException If the object to marshal cannot be introspected
     */
    public OpenCsvRecordMarshaller(final Class<P> type, final String... fields) throws IntrospectionException {
        this.delimiter = DEFAULT_DELIMITER;
        this.qualifier = DEFAULT_QUALIFIER;
        this.fieldExtractor = new BeanFieldExtractor<>(type, fields);
    }

    @Override
    public StringRecord processRecord(Record<P> record) throws Exception {
        StringWriter stringWriter = null;
        CSVWriter csvWriter = null;
        try {
            stringWriter = new StringWriter();
            csvWriter = new CSVWriter(stringWriter, delimiter, qualifier, ""); // force lineEnd to empty string
            P payload = record.getPayload();
            List<String> fields = extractFields(payload);
            String[] items = fields.toArray(new String[fields.size()]);
            csvWriter.writeNext(items);
            csvWriter.flush();
            return new StringRecord(record.getHeader(), stringWriter.toString());
        } finally {
            if (csvWriter != null) {
                csvWriter.close();
            }
            if (stringWriter != null) {
                stringWriter.close();
            }
        }
    }

    private List<String> extractFields(P payload) throws Exception {
        List<String> tokens = new ArrayList<>();
        Iterable<Object> objects = fieldExtractor.extractFields(payload);
        for (Object object : objects) {
            tokens.add(object.toString());
        }
        return tokens;
    }

    /**
     * Set the delimiter.
     *
     * @param delimiter to use
     */
    public void setDelimiter(char delimiter) {
        this.delimiter = delimiter;
    }

    /**
     * Set the qualifier.
     *
     * @param qualifier to use
     */
    public void setQualifier(char qualifier) {
        this.qualifier = qualifier;
    }

    /**
     * Set field extractor.
     *
     * @param fieldExtractor to use
     */
    public void setFieldExtractor(final FieldExtractor<P> fieldExtractor) {
        Utils.checkNotNull(fieldExtractor, "field extractor");
        this.fieldExtractor = fieldExtractor;
    }
}

@@ -1569,24 +1569,542 @@
 ds);%0A    %7D%0A%0A
+    /**%0A     * Create a open CSV record marshaller.%0A     *%0A     * @param fieldExtractor to use to extract fields%0A     * @throws IntrospectionException If the object to marshal cannot be introspected%0A     */%0A    public OpenCsvRecordMarshaller(final FieldExtractor%3CP%3E fieldExtractor) throws IntrospectionException %7B%0A        Utils.checkNotNull(fieldExtractor, %22field extractor%22);%0A        this.fieldExtractor = fieldExtractor;%0A        this.delimiter = DEFAULT_DELIMITER;%0A        this.qualifier = DEFAULT_QUALIFIER;%0A    %7D%0A%0A
     @Overrid
@@ -3631,286 +3631,7 @@
    %7D
-%0A%0A    /**%0A     * Set field extractor.%0A     *%0A     * @param fieldExtractor to use%0A     */%0A    public void setFieldExtractor(final FieldExtractor%3CP%3E fieldExtractor) %7B%0A        Utils.checkNotNull(fieldExtractor, %22field extractor%22);%0A        this.fieldExtractor = fieldExtractor;%0A    %7D
 %0A%7D%0A

add constructor with FieldExtractor in OpenCsvRecordMarshaller