/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.camel.dataformat.csv;

import java.io.InputStream;
import java.io.OutputStream;
import java.util.StringJoiner;

import org.apache.camel.Exchange;
import org.apache.camel.spi.DataFormat;
import org.apache.camel.spi.DataFormatName;
import org.apache.camel.spi.annotations.Dataformat;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.QuoteMode;

/**
 * CSV Data format.
 * <p/>
 * By default, columns are autogenerated in the resulting CSV. Subsequent messages use the previously created columns
 * with new fields being added at the end of the line. Thus, field order is the same from message to message.
 * Autogeneration can be disabled. In this case, only the fields defined in csvConfig are written on the output.
 */
@Dataformat("csv")
public class CsvDataFormat extends ServiceSupport implements DataFormat, DataFormatName {
    // CSV format options
    private CSVFormat csvFormat = CSVFormat.DEFAULT;
    private String format;
    private boolean commentMarkerDisabled;
    private Character commentMarker;
    private Character delimiter;
    private boolean escapeDisabled;
    private Character escape;
    private boolean headerDisabled;
    private String header;
    private Boolean allowMissingColumnNames;
    private Boolean ignoreEmptyLines;
    private Boolean ignoreSurroundingSpaces;
    private boolean nullStringDisabled;
    private String nullString;
    private boolean quoteDisabled;
    private Character quote;
    private QuoteMode quoteMode;
    private boolean recordSeparatorDisabled;
    private String recordSeparator;
    private Boolean skipHeaderRecord;
    private Boolean trim;
    private Boolean ignoreHeaderCase;
    private Boolean trailingDelimiter;

    // Unmarshal options
    private boolean captureHeaderRecord;
    private boolean lazyLoad;
    private boolean useMaps;
    private boolean useOrderedMaps;
    private CsvRecordConverter<?> recordConverter;

    private CsvMarshallerFactory marshallerFactory = CsvMarshallerFactory.DEFAULT;

    private volatile CsvMarshaller marshaller;
    private volatile CsvUnmarshaller unmarshaller;

    public CsvDataFormat() {
    }

    public CsvDataFormat(CSVFormat format) {
        csvFormat = format;
    }

    @Override
    public String getDataFormatName() {
        return "csv";
    }

    @Override
    public void marshal(Exchange exchange, Object object, OutputStream outputStream) throws Exception {
        marshaller.marshal(exchange, object, outputStream);
    }

    @Override
    public Object unmarshal(Exchange exchange, InputStream inputStream) throws Exception {
        return unmarshal(exchange, (Object) inputStream);
    }

    @Override
    public Object unmarshal(Exchange exchange, Object body) throws Exception {
        return unmarshaller.unmarshal(exchange, body);
    }

    @Override
    protected void doInit() throws Exception {
        super.doInit();

        if (csvFormat == null && format != null) {
            csvFormat = CSVFormat.valueOf(format);
        }
        if (csvFormat == null) {
            csvFormat = CSVFormat.DEFAULT;
        }

        marshaller = marshallerFactory.create(getActiveFormat(), this);
        unmarshaller = CsvUnmarshaller.create(getActiveFormat(), this);
    }

    @Override
    protected void doStop() throws Exception {
        // noop
    }

    CSVFormat getActiveFormat() {
        CSVFormat answer = csvFormat;

        if (commentMarkerDisabled) {
            answer = answer.builder().setCommentMarker(null).get(); // null disables the comment marker
        } else if (commentMarker != null) {
            answer = answer.builder().setCommentMarker(commentMarker).get();
        }

        if (delimiter != null) {
            answer = answer.builder().setDelimiter(delimiter).get();
        }

        if (escapeDisabled) {
            answer = answer.builder().setEscape(null).get(); // null disables the escape
        } else if (escape != null) {
            answer = answer.builder().setEscape(escape).get();
        }

        if (headerDisabled) {
            answer = answer.builder().setHeader((String[]) null).get(); // null disables the header
        } else if (header != null) {
            if (header.indexOf(',') != -1) {
                answer = answer.builder().setHeader(header.split(",")).get();
            } else {
                answer = answer.builder().setHeader(header).get();
            }
        }

        if (allowMissingColumnNames != null) {
            answer = answer.builder().setAllowMissingColumnNames(allowMissingColumnNames).get();
        }

        if (ignoreEmptyLines != null) {
            answer = answer.builder().setIgnoreEmptyLines(ignoreEmptyLines).get();
        }

        if (ignoreSurroundingSpaces != null) {
            answer = answer.builder().setIgnoreSurroundingSpaces(ignoreSurroundingSpaces).get();
        }

        if (nullStringDisabled) {
            answer = answer.builder().setNullString(null).get(); // null disables the null string replacement
        } else if (nullString != null) {
            answer = answer.builder().setNullString(nullString).get();
        }

        if (quoteDisabled) {
            answer = answer.builder().setQuote(null).get(); // null disables quotes
        } else if (quote != null) {
            answer = answer.builder().setQuote(quote).get();
        }

        if (quoteMode != null) {
            answer = answer.builder().setQuoteMode(quoteMode).get();
        }

        if (recordSeparatorDisabled) {
            answer = answer.builder().setRecordSeparator(null).get(); // null disables the record separator
        } else if (recordSeparator != null) {
            answer = answer.builder().setRecordSeparator(recordSeparator).get();
        }

        if (skipHeaderRecord != null) {
            answer = answer.builder().setSkipHeaderRecord(skipHeaderRecord).get();
        }

        if (trim != null) {
            answer = answer.builder().setTrim(trim).get();
        }

        if (ignoreHeaderCase != null) {
            answer = answer.builder().setIgnoreHeaderCase(ignoreHeaderCase).get();
        }

        if (trailingDelimiter != null) {
            answer = answer.builder().setTrailingDelimiter(trailingDelimiter).get();
        }

        return answer;
    }

    //region Getters/Setters

    /**
     * Sets the {@link CsvMarshaller} factory. If {@code null}, then {@link CsvMarshallerFactory#DEFAULT} is used
     * instead.
     *
     * @param  marshallerFactory
     * @return                   Current {@code CsvDataFormat}, fluent API
     */
    public CsvDataFormat setMarshallerFactory(CsvMarshallerFactory marshallerFactory) {
        this.marshallerFactory = (marshallerFactory == null) ? CsvMarshallerFactory.DEFAULT : marshallerFactory;
        return this;
    }

    /**
     * Returns the used {@link CsvMarshallerFactory}.
     *
     * @return never {@code null}.
     */
    public CsvMarshallerFactory getMarshallerFactory() {
        return marshallerFactory;
    }

    public CSVFormat getCsvFormat() {
        return csvFormat;
    }

    public void setCsvFormat(CSVFormat csvFormat) {
        this.csvFormat = csvFormat;
    }

    public String getFormat() {
        return format;
    }

    public void setFormat(String format) {
        this.format = format;
    }

    /**
     * Indicates whether the comment markers are disabled.
     *
     * @return {@code true} if the comment markers are disabled, {@code false} otherwise
     */
    public boolean isCommentMarkerDisabled() {
        return commentMarkerDisabled;
    }

    /**
     * Sets whether the comment markers are disabled.
     *
     * @param  commentMarkerDisabled {@code true} if the comment markers are disabled, {@code false} otherwise
     * @return                       Current {@code CsvDataFormat}, fluent API
     * @see                          org.apache.commons.csv.CSVFormat#withCommentMarker(java.lang.Character)
     */
    public CsvDataFormat setCommentMarkerDisabled(boolean commentMarkerDisabled) {
        this.commentMarkerDisabled = commentMarkerDisabled;
        return this;
    }

    /**
     * Gets the comment marker. If {@code null} then the default one of the format used.
     *
     * @return Comment marker
     */
    public Character getCommentMarker() {
        return commentMarker;
    }

    /**
     * Sets the comment marker to use. If {@code null} then the default one of the format used.
     *
     * @param  commentMarker Comment marker
     * @return               Current {@code CsvDataFormat}, fluent API
     * @see                  org.apache.commons.csv.CSVFormat#withCommentMarker(Character)
     */
    public CsvDataFormat setCommentMarker(Character commentMarker) {
        this.commentMarker = commentMarker;
        return this;
    }

    /**
     * Gets the delimiter. If {@code null} then the default one of the format used.
     *
     * @return Delimiter
     */
    public Character getDelimiter() {
        return delimiter;
    }

    /**
     * Sets the delimiter. If {@code null} then the default one of the format used.
     *
     * @param  delimiter Delimiter
     * @return           Current {@code CsvDataFormat}, fluent API
     * @see              org.apache.commons.csv.CSVFormat#withDelimiter(char)
     */
    public CsvDataFormat setDelimiter(Character delimiter) {
        this.delimiter = delimiter;
        return this;
    }

    /**
     * Indicates whether the escaping is disabled.
     *
     * @return {@code true} if the escaping is disabled, {@code false} otherwise
     */
    public boolean isEscapeDisabled() {
        return escapeDisabled;
    }

    /**
     * Sets whether the escaping is disabled.
     *
     * @param  escapeDisabled {@code true} if the escaping is disabled, {@code false} otherwise
     * @return                Current {@code CsvDataFormat}, fluent API
     * @see                   org.apache.commons.csv.CSVFormat#withEscape(Character)
     */
    public CsvDataFormat setEscapeDisabled(boolean escapeDisabled) {
        this.escapeDisabled = escapeDisabled;
        return this;
    }

    /**
     * Gets the escape character. If {@code null} then the default one of the format used.
     *
     * @return Escape character
     */
    public Character getEscape() {
        return escape;
    }

    /**
     * Sets the escape character. If {@code null} then the default one of the format used.
     *
     * @param  escape Escape character
     * @return        Current {@code CsvDataFormat}, fluent API
     * @see           org.apache.commons.csv.CSVFormat#withEscape(Character)
     */
    public CsvDataFormat setEscape(Character escape) {
        this.escape = escape;
        return this;
    }

    /**
     * Indicates whether the headers are disabled.
     *
     * @return {@code true} if the headers are disabled, {@code false} otherwise
     */
    public boolean isHeaderDisabled() {
        return headerDisabled;
    }

    /**
     * Sets whether the headers are disabled.
     *
     * @param  headerDisabled {@code true} if the headers are disabled, {@code false} otherwise
     * @return                Current {@code CsvDataFormat}, fluent API
     * @see                   org.apache.commons.csv.CSVFormat#withHeader(String...)
     */
    public CsvDataFormat setHeaderDisabled(boolean headerDisabled) {
        this.headerDisabled = headerDisabled;
        return this;
    }

    /**
     * Gets the header. Multiple values can be separated by comma.
     *
     * If {@code null} then the default one of the format used. If empty, then it will be automatically handled.
     *
     * @return Header
     */
    public String getHeader() {
        return header;
    }

    /**
     * Gets the header. Multiple values can be separated by comma.
     *
     * If {@code null} then the default one of the format used. If empty, then it will be automatically handled.
     *
     * @param  header Header
     * @return        Current {@code CsvDataFormat}, fluent API
     * @see           org.apache.commons.csv.CSVFormat#withHeader(String...)
     */
    public CsvDataFormat setHeader(String header) {
        this.header = header;
        return this;
    }

    public CsvDataFormat setHeader(String[] header) {
        StringJoiner sj = new StringJoiner(",");
        for (String s : header) {
            sj.add(s);
        }
        this.header = sj.toString();
        return this;
    }

    /**
     * Indicates whether missing column names are allowed. If {@code null} then the default value of the format used.
     *
     * @return whether missing column names are allowed
     */
    public Boolean getAllowMissingColumnNames() {
        return allowMissingColumnNames;
    }

    /**
     * Sets whether missing column names are allowed. If {@code null} then the default value of the format used.
     *
     * @param  allowMissingColumnNames whether missing column names are allowed
     * @return                         Current {@code CsvDataFormat}, fluent API
     * @see                            org.apache.commons.csv.CSVFormat#withAllowMissingColumnNames(boolean)
     */
    public CsvDataFormat setAllowMissingColumnNames(Boolean allowMissingColumnNames) {
        this.allowMissingColumnNames = allowMissingColumnNames;
        return this;
    }

    /**
     * Indicates whether empty lines must be ignored. If {@code null} then the default value of the format used.
     *
     * @return whether empty lines must be ignored
     */
    public Boolean getIgnoreEmptyLines() {
        return ignoreEmptyLines;
    }

    /**
     * Sets whether empty lines must be ignored. If {@code null} then the default value of the format used.
     *
     * @param  ignoreEmptyLines whether empty lines must be ignored
     * @return                  Current {@code CsvDataFormat}, fluent API
     * @see                     org.apache.commons.csv.CSVFormat#withIgnoreEmptyLines(boolean)
     */
    public CsvDataFormat setIgnoreEmptyLines(Boolean ignoreEmptyLines) {
        this.ignoreEmptyLines = ignoreEmptyLines;
        return this;
    }

    /**
     * Indicates whether surrounding spaces must be ignored. If {@code null} then the default value of the format used.
     *
     * @return whether surrounding spaces must be ignored
     */
    public Boolean getIgnoreSurroundingSpaces() {
        return ignoreSurroundingSpaces;
    }

    /**
     * Sets whether surrounding spaces must be ignored. If {@code null} then the default value of the format used.
     *
     * @param  ignoreSurroundingSpaces whether surrounding spaces must be ignored
     * @return                         Current {@code CsvDataFormat}, fluent API
     * @see                            org.apache.commons.csv.CSVFormat#withIgnoreSurroundingSpaces(boolean)
     */
    public CsvDataFormat setIgnoreSurroundingSpaces(Boolean ignoreSurroundingSpaces) {
        this.ignoreSurroundingSpaces = ignoreSurroundingSpaces;
        return this;
    }

    /**
     * Indicates whether the null string replacement is disabled.
     *
     * @return {@code true} if the null string replacement is disabled, {@code false} otherwise
     */
    public boolean isNullStringDisabled() {
        return nullStringDisabled;
    }

    /**
     * Sets whether the null string replacement is disabled.
     *
     * @param  nullStringDisabled {@code true} if the null string replacement is disabled, {@code false} otherwise
     * @return                    Current {@code CsvDataFormat}, fluent API
     * @see                       org.apache.commons.csv.CSVFormat#withNullString(String)
     */
    public CsvDataFormat setNullStringDisabled(boolean nullStringDisabled) {
        this.nullStringDisabled = nullStringDisabled;
        return this;
    }

    /**
     * Gets the null string replacement. If {@code null} then the default one of the format used.
     *
     * @return Null string replacement
     */
    public String getNullString() {
        return nullString;
    }

    /**
     * Sets the null string replacement. If {@code null} then the default one of the format used.
     *
     * @param  nullString Null string replacement
     * @return            Current {@code CsvDataFormat}, fluent API
     * @see               org.apache.commons.csv.CSVFormat#withNullString(String)
     */
    public CsvDataFormat setNullString(String nullString) {
        this.nullString = nullString;
        return this;
    }

    /**
     * Indicates whether quotes are disabled.
     *
     * @return {@code true} if quotes are disabled, {@code false} otherwise
     */
    public boolean isQuoteDisabled() {
        return quoteDisabled;
    }

    /**
     * Sets whether quotes are disabled
     *
     * @param  quoteDisabled {@code true} if quotes are disabled, {@code false} otherwise
     * @return               Current {@code CsvDataFormat}, fluent API
     * @see                  org.apache.commons.csv.CSVFormat#withQuote(Character)
     */
    public CsvDataFormat setQuoteDisabled(boolean quoteDisabled) {
        this.quoteDisabled = quoteDisabled;
        return this;
    }

    /**
     * Gets the quote character. If {@code null} then the default one of the format used.
     *
     * @return Quote character
     */
    public Character getQuote() {
        return quote;
    }

    /**
     * Sets the quote character. If {@code null} then the default one of the format used.
     *
     * @param  quote Quote character
     * @return       Current {@code CsvDataFormat}, fluent API
     * @see          org.apache.commons.csv.CSVFormat#withQuote(Character)
     */
    public CsvDataFormat setQuote(Character quote) {
        this.quote = quote;
        return this;
    }

    /**
     * Gets the quote mode. If {@code null} then the default one of the format used.
     *
     * @return Quote mode
     */
    public QuoteMode getQuoteMode() {
        return quoteMode;
    }

    /**
     * Sets the quote mode. If {@code null} then the default one of the format used.
     *
     * @param  quoteMode Quote mode
     * @return           Current {@code CsvDataFormat}, fluent API
     * @see              org.apache.commons.csv.CSVFormat#withQuoteMode(org.apache.commons.csv.QuoteMode)
     */
    public CsvDataFormat setQuoteMode(QuoteMode quoteMode) {
        this.quoteMode = quoteMode;
        return this;
    }

    /**
     * Indicates whether the record separator is disabled.
     *
     * @return {@code true} if the record separator disabled, {@code false} otherwise
     */
    public boolean isRecordSeparatorDisabled() {
        return recordSeparatorDisabled;
    }

    /**
     * Sets whether the record separator is disabled.
     *
     * @param  recordSeparatorDisabled {@code true} if the record separator disabled, {@code false} otherwise
     * @return                         Current {@code CsvDataFormat}, fluent API
     * @see                            org.apache.commons.csv.CSVFormat#withRecordSeparator(String)
     */
    public CsvDataFormat setRecordSeparatorDisabled(boolean recordSeparatorDisabled) {
        this.recordSeparatorDisabled = recordSeparatorDisabled;
        return this;
    }

    /**
     * Gets the record separator. If {@code null} then the default one of the format used.
     *
     * @return Record separator
     */
    public String getRecordSeparator() {
        return recordSeparator;
    }

    /**
     * Sets the record separator. If {@code null} then the default one of the format used.
     *
     * @param  recordSeparator Record separator
     * @return                 Current {@code CsvDataFormat}, fluent API
     * @see                    org.apache.commons.csv.CSVFormat#withRecordSeparator(String)
     */
    public CsvDataFormat setRecordSeparator(String recordSeparator) {
        this.recordSeparator = recordSeparator;
        return this;
    }

    /**
     * Indicates whether header record must be skipped. If {@code null} then the default value of the format used.
     *
     * @return whether header record must be skipped
     */
    public Boolean getSkipHeaderRecord() {
        return skipHeaderRecord;
    }

    /**
     * Sets whether header record must be skipped. If {@code null} then the default value of the format used.
     *
     * @param  skipHeaderRecord whether header record must be skipped
     * @return                  Current {@code CsvDataFormat}, fluent API
     * @see                     org.apache.commons.csv.CSVFormat#withSkipHeaderRecord(boolean)
     */
    public CsvDataFormat setSkipHeaderRecord(Boolean skipHeaderRecord) {
        this.skipHeaderRecord = skipHeaderRecord;
        return this;
    }

    /**
     * Indicates whether the unmarshalling should capture the header record.
     *
     * @return {@code true} for capture header record, {@code false} otherwise
     */
    public boolean isCaptureHeaderRecord() {
        return captureHeaderRecord;
    }

    /**
     * Indicates whether the unmarshalling should capture the header record.
     *
     * @param  captureHeaderRecord {@code true} for capture header record, {@code false} otherwise
     * @return                     Current {@code CsvDataFormat}, fluent API
     */
    public CsvDataFormat setCaptureHeaderRecord(boolean captureHeaderRecord) {
        this.captureHeaderRecord = captureHeaderRecord;
        return this;
    }

    /**
     * Indicates whether the unmarshalling should lazily load the records.
     *
     * @return {@code true} for lazy loading, {@code false} otherwise
     */
    public boolean isLazyLoad() {
        return lazyLoad;
    }

    /**
     * Indicates whether the unmarshalling should lazily load the records.
     *
     * @param  lazyLoad {@code true} for lazy loading, {@code false} otherwise
     * @return          Current {@code CsvDataFormat}, fluent API
     */
    public CsvDataFormat setLazyLoad(boolean lazyLoad) {
        this.lazyLoad = lazyLoad;
        return this;
    }

    /**
     * Indicates whether the unmarshalling should produce maps instead of lists.
     *
     * @return {@code true} for maps, {@code false} for lists
     */
    public boolean isUseMaps() {
        return useMaps;
    }

    /**
     * Sets whether the unmarshalling should produce maps instead of lists.
     *
     * @param  useMaps {@code true} for maps, {@code false} for lists
     * @return         Current {@code CsvDataFormat}, fluent API
     */
    public CsvDataFormat setUseMaps(boolean useMaps) {
        this.useMaps = useMaps;
        return this;
    }

    /**
     * Indicates whether the unmarshalling should produce ordered maps instead of lists.
     *
     * @return {@code true} for maps, {@code false} for lists
     */
    public boolean isUseOrderedMaps() {
        return useOrderedMaps;
    }

    /**
     * Sets whether the unmarshalling should produce ordered maps instead of lists.
     *
     * @param  useOrderedMaps {@code true} for maps, {@code false} for lists
     * @return                Current {@code CsvDataFormat}, fluent API
     */
    public CsvDataFormat setUseOrderedMaps(boolean useOrderedMaps) {
        this.useOrderedMaps = useOrderedMaps;
        return this;
    }

    /**
     * Gets the record converter to use. If {@code null} then it will use {@link CsvDataFormat#isUseMaps()} for finding
     * the proper converter.
     *
     * @return Record converter to use
     */
    public CsvRecordConverter<?> getRecordConverter() {
        return recordConverter;
    }

    /**
     * Sets the record converter to use. If {@code null} then it will use {@link CsvDataFormat#isUseMaps()} for finding
     * the proper converter.
     *
     * @param  recordConverter Record converter to use
     * @return                 Current {@code CsvDataFormat}, fluent API
     */
    public CsvDataFormat setRecordConverter(CsvRecordConverter<?> recordConverter) {
        this.recordConverter = recordConverter;
        return this;
    }

    //endregion
    /**
     * Sets whether to trim leading and trailing blanks.
     * <p>
     * If {@code null} then the default value of the format used.
     * </p>
     *
     * @param  trim whether to trim leading and trailing blanks. <code>null</code> value allowed.
     * @return      Current {@code CsvDataFormat}, fluent API.
     */
    public CsvDataFormat setTrim(Boolean trim) {
        this.trim = trim;
        return this;
    }

    /**
     * Indicates whether to trim leading and trailing blanks.
     *
     * @return {@link Boolean#TRUE} if leading and trailing blanks should be trimmed. {@link Boolean#FALSE} otherwise.
     *         Could return <code>null</code> if value has NOT been set.
     */
    public Boolean getTrim() {
        return trim;
    }

    /**
     * Sets whether to ignore case when accessing header names.
     * <p>
     * If {@code null} then the default value of the format used.
     * </p>
     *
     * @param  ignoreHeaderCase whether to ignore case when accessing header names. <code>null</code> value allowed.
     * @return                  Current {@code CsvDataFormat}, fluent API.
     */
    public CsvDataFormat setIgnoreHeaderCase(Boolean ignoreHeaderCase) {
        this.ignoreHeaderCase = ignoreHeaderCase;
        return this;
    }

    /**
     * Indicates whether to ignore case when accessing header names.
     *
     * @return {@link Boolean#TRUE} if case should be ignored when accessing header name. {@link Boolean#FALSE}
     *         otherwise. Could return <code>null</code> if value has NOT been set.
     */
    public Boolean getIgnoreHeaderCase() {
        return ignoreHeaderCase;
    }

    /**
     * Sets whether to add a trailing delimiter.
     * <p>
     * If {@code null} then the default value of the format used.
     * </p>
     *
     * @param  trailingDelimiter whether to add a trailing delimiter.
     * @return                   Current {@code CsvDataFormat}, fluent API.
     */
    public CsvDataFormat setTrailingDelimiter(Boolean trailingDelimiter) {
        this.trailingDelimiter = trailingDelimiter;
        return this;
    }

    /**
     * Indicates whether to add a trailing delimiter.
     *
     * @return {@link Boolean#TRUE} if a trailing delimiter should be added. {@link Boolean#FALSE} otherwise. Could
     *         return <code>null</code> if value has NOT been set.
     */
    public Boolean getTrailingDelimiter() {
        return trailingDelimiter;
    }
}
