/* Stratosphere output format for Ptolemy tokens.
 * 
 * Copyright (c) 2014 The Regents of the University of California.
 * All rights reserved.
 *
 * '$Author: crawl $'
 * '$Date: 2015-08-24 22:42:20 +0000 (Mon, 24 Aug 2015) $' 
 * '$Revision: 33628 $'
 * 
 * Permission is hereby granted, without written agreement and without
 * license or royalty fees, to use, copy, modify, and distribute this
 * software and its documentation for any purpose, provided that the above
 * copyright notice and the following two paragraphs appear in all copies
 * of this software.
 *
 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY
 * FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES
 * ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF
 * THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE
 * PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND THE UNIVERSITY OF
 * CALIFORNIA HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES,
 * ENHANCEMENTS, OR MODIFICATIONS.
 *
 */
package org.kepler.stratosphere.io.output;

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

import org.kepler.ddp.Utilities;
import org.kepler.ddp.actor.pattern.DDPDataSink;
import org.kepler.stratosphere.stub.StubUtilities;
import org.kepler.stratosphere.type.TypeUtilities;

import eu.stratosphere.api.common.io.OutputFormat;
import eu.stratosphere.api.common.operators.GenericDataSink;
import eu.stratosphere.configuration.Configuration;
import eu.stratosphere.types.Record;
import eu.stratosphere.types.Value;
import ptolemy.data.RecordToken;
import ptolemy.data.Token;
import ptolemy.kernel.util.IllegalActionException;

/** An output format for Ptolemy tokens.
 *  
 * @author Daniel Crawl
 * @version $Id: TokenOutputFormat.java 33628 2015-08-24 22:42:20Z crawl $
 * 
 */
public class TokenOutputFormat implements OutputFormat<Record> {

        public TokenOutputFormat() {
        }

        /**
         * Configures this output format. Since output formats are instantiated generically and hence parameterless, 
         * this method is the place where the output formats set their basic fields based on configuration values.
         * <p>
         * This method is always called first on a newly instantiated output format. 
         *  
         * @param parameters The configuration with all parameters.
         */
        @Override
    public void configure(Configuration parameters) {
                
            // TODO: possible to use this when kepler is in a separate JVM?
            if(!parameters.getBoolean(Utilities.CONFIGURATION_KEPLER_SAME_JVM, false)) {
                throw new RuntimeException("TokenOutputFormat only works when running Kepler in the same JVM.");
        }
            
            this.sinkActorName = parameters.getString(Utilities.CONFIGURATION_KEPLER_SINK_ACTOR_NAME, null);
            if(sinkActorName == null) {
                throw new RuntimeException("Name of DDPDataSink actor not in configuration.");
            }


        this.numFields = parameters.getInteger(NUM_FIELDS_PARAMETER, -1);
        if (this.numFields < 1) {
            throw new IllegalArgumentException("Invalid configuration for TokenOutputFormat: " +
                    "Need to specify number of fields > 0.");
        }
        if(this.numFields > 2) {
            throw new IllegalArgumentException("Invalid configuration for TokenOutputFormat: " +
                    "Need to specify number of fields < 3.");
        }
        
        @SuppressWarnings("unchecked")
        Class<Value>[] arr = new Class[this.numFields];
        this.classes = arr;
        
        for (int i = 0; i < this.numFields; i++)
        {
            @SuppressWarnings("unchecked")
            Class<? extends Value> clazz = (Class<? extends Value>) parameters.getClass(FIELD_TYPE_PARAMETER_PREFIX + i, null);
            if (clazz == null) {
                throw new IllegalArgumentException("Invalid configuration for CsvOutputFormat: " +
                    "No type class for parameter " + i);
            }
            
            this.classes[i] = clazz;
        }
        
        this.recordPositions = new int[this.numFields];
        boolean anyRecordPosDefined = false;
        boolean allRecordPosDefined = true;
        
        for(int i = 0; i < this.numFields; i++) {
            
            int pos = parameters.getInteger(RECORD_POSITION_PARAMETER_PREFIX + i, Integer.MIN_VALUE);
            
            if(pos != Integer.MIN_VALUE) {
                anyRecordPosDefined = true;
                
                if(pos < 0) {
                    throw new IllegalArgumentException("Invalid configuration for CsvOutputFormat: " +
                            "Invalid record position for parameter " + i);
                }

                this.recordPositions[i] = pos;
                
            } else {
                allRecordPosDefined = false;
                
                this.recordPositions[i] = i;
            }
        }
        
        if(anyRecordPosDefined && !allRecordPosDefined) {
            throw new IllegalArgumentException("Invalid configuration for CsvOutputFormat: " +
                    "Either none or all record positions must be defined.");
        }
        
        //this.recordDelimiter = parameters.getString(RECORD_DELIMITER_PARAMETER, AbstractConfigBuilder.NEWLINE_DELIMITER);
        //if (this.recordDelimiter == null) {
            //throw new IllegalArgumentException("The delimiter in the DelimitedOutputFormat must not be null.");
        //}
        //this.fieldDelimiter = parameters.getString(FIELD_DELIMITER_PARAMETER, "|");
        //this.lenient = parameters.getBoolean(LENIENT_PARSING, false);         
        }
        
        /**
         * Opens a parallel instance of the output format to store the result of its parallel instance.
         * <p>
         * When this method is called, the output format it guaranteed to be configured.
         * 
         * @param taskNumber The number of the parallel instance.
         * @throws IOException Thrown, if the output could not be opened due to an I/O problem.
         */
        @Override
    public void open(int taskNumber) throws IOException {
            this.tokenList = new ArrayList<Token>();
        }
        
        
        /**
         * Adds a record to the output.
         * <p>
         * When this method is called, the output format it guaranteed to be opened.
         * 
         * @param record The records to add to the output.
         * @throws IOException Thrown, if the records could not be added to to an I/O problem.
         */
        @Override
    public void writeRecord(Record record) throws IOException {
            
        int numRecFields = record.getNumFields();
        
        Token recordToken;
        
        try {
            // see if there's only one field
            if(numRecFields == 1) {
                // since there's only one field, assume it is the value
                Token valueToken = StubUtilities.convertPactDataToToken(record, 0, classes[0]);
                recordToken = new RecordToken(new String[] {"key", "value"}, new Token[] {Token.NIL, valueToken});
            } else if(numRecFields == 2) {
                Token keyToken = StubUtilities.convertPactDataToToken(record,
                        TypeUtilities.KEY_FIELD, classes[TypeUtilities.KEY_FIELD]);
                Token valueToken = StubUtilities.convertPactDataToToken(record,
                        TypeUtilities.VALUE_FIELD, classes[TypeUtilities.VALUE_FIELD]);
                recordToken = new RecordToken(new String[] {"key", "value"}, new Token[] {keyToken, valueToken});
            } else {
                throw new RuntimeException("Incorrect number of fields in record: " + numRecFields);            
            }
        } catch(IllegalActionException e) {
            throw new RuntimeException("Error creating RecordToken.", e);
        }
        
        this.tokenList.add(recordToken);
                
        }
        
        /**
         * Method that marks the end of the life-cycle of parallel output instance. Should be used to close
         * channels and streams and release resources.
         * After this method returns without an error, the output is assumed to be correct.
         * <p>
         * When this method is called, the output format it guaranteed to be opened.
         *  
         * @throws IOException Thrown, if the input could not be closed properly.
         */
        @Override
    public void close() throws IOException {
            try {
            DDPDataSink.addTokens(this.sinkActorName, this.tokenList);
        } catch (IllegalActionException e) {
            throw new IOException("Error writing token.", e);
        }
            this.tokenList = null;
        }

           /**
     * Creates a configuration builder that can be used to set the input format's parameters to the config in a fluent
     * fashion.
     * 
     * @return A config builder for setting parameters.
     */
    public static ConfigBuilder configureRecordFormat(GenericDataSink target) {
        return new ConfigBuilder(target.getParameters());
    }
    
    //public static final String RECORD_DELIMITER_PARAMETER = "pact.output.record.delimiter";
    
    //public static final String FIELD_DELIMITER_PARAMETER = "pact.output.record.field-delimiter";
    
    public static final String NUM_FIELDS_PARAMETER = "pact.output.record.num-fields";
    
    public static final String FIELD_TYPE_PARAMETER_PREFIX = "pact.output.record.type_";
    
    public static final String RECORD_POSITION_PARAMETER_PREFIX = "pact.output.record.position_";
    
    //public static final String LENIENT_PARSING = "pact.output.record.lenient";

    /**
     * Abstract builder used to set parameters to the input format's configuration in a fluent way.
     */
    protected static abstract class AbstractConfigBuilder<T>
    {
        //private static final String NEWLINE_DELIMITER = "\n";
        
        // --------------------------------------------------------------------
        
        /**
         * The configuration into which the parameters will be written.
         */
        protected final Configuration config;
        
        // --------------------------------------------------------------------
        
        /**
         * Creates a new builder for the given configuration.
         * 
         * @param targetConfig The configuration into which the parameters will be written.
         */
        protected AbstractConfigBuilder(Configuration targetConfig) {
            this.config = targetConfig;
        }
        
        // --------------------------------------------------------------------
        
        /**
         * Sets the delimiter to be a single character, namely the given one. The character must be within
         * the value range <code>0</code> to <code>127</code>.
         * 
         * @param delimiter The delimiter character.
         * @return The builder itself.
         */
        /*
        public T recordDelimiter(char delimiter) {
            if (delimiter == '\n') {
                this.config.setString(RECORD_DELIMITER_PARAMETER, NEWLINE_DELIMITER);
            } else {
                this.config.setString(RECORD_DELIMITER_PARAMETER, String.valueOf(delimiter));
            }
            @SuppressWarnings("unchecked")
            T ret = (T) this;
            return ret;
        }
        */
        
        /**
         * Sets the delimiter to be the given string. The string will be converted to bytes for more efficient
         * comparison during input parsing. The conversion will be done using the platforms default charset.
         * 
         * @param delimiter The delimiter string.
         * @return The builder itself.
         */
        /*
        public T recordDelimiter(String delimiter) {
            this.config.setString(RECORD_DELIMITER_PARAMETER, delimiter);
            @SuppressWarnings("unchecked")
            T ret = (T) this;
            return ret;
        }
        */
                
        /**
         * Sets the delimiter that delimits the individual fields in the records textual output representation.
         * 
         * @param delimiter The character to be used as a field delimiter.
         * @return The builder itself.
         */
        /*
        public T fieldDelimiter(char delimiter) {
            this.config.setString(FIELD_DELIMITER_PARAMETER, String.valueOf(delimiter));
            @SuppressWarnings("unchecked")
            T ret = (T) this;
            return ret;
        }
        */
        
        /**
         * Adds a field of the record to be serialized to the output. The field at the given position will
         * be interpreted as the type represented by the given class. The types {@link Object#toString()} method
         * will be invoked to create a textual representation. 
         * 
         * @param type The type of the field.
         * @param recordPosition The position in the record.
         * @return The builder itself.
         */
        public T field(Class<? extends Value> type, int recordPosition) {
            final int numYet = this.config.getInteger(NUM_FIELDS_PARAMETER, 0);
            this.config.setClass(FIELD_TYPE_PARAMETER_PREFIX + numYet, type);
            this.config.setInteger(RECORD_POSITION_PARAMETER_PREFIX + numYet, recordPosition);
            this.config.setInteger(NUM_FIELDS_PARAMETER, numYet + 1);
            @SuppressWarnings("unchecked")
            T ret = (T) this;
            return ret;
        }
        
        /**
         * Sets the leniency for the serializer. A lenient serializer simply skips missing fields and null
         * fields in the record, while a non lenient one throws an exception.
         * 
         * @param lenient True, if the serializer should be lenient, false otherwise.
         * @return The builder itself.
         */
        /*
        public T lenient(boolean lenient) {
            this.config.setBoolean(LENIENT_PARSING, lenient);
            @SuppressWarnings("unchecked")
            T ret = (T) this;
            return ret;
        }
        */
    }
    
    /**
     * A builder used to set parameters to the input format's configuration in a fluent way.
     */
    public static final class ConfigBuilder extends AbstractConfigBuilder<ConfigBuilder>
    {
        /**
         * Creates a new builder for the given configuration.
         * 
         * @param targetConfig The configuration into which the parameters will be written.
         */
        protected ConfigBuilder(Configuration targetConfig) {
            super(targetConfig);
        }
        
    }
    
    // --------------------------------------------------------------------------------------------
    
    private int numFields;
    
    private Class<? extends Value>[] classes;
    
    private int[] recordPositions;
    
    //private String fieldDelimiter;
    
    //private String recordDelimiter;
        
    //private boolean lenient;
    
    /** A list of tokens accumulated from writeRecord(). */
    private List<Token> tokenList;
    
    /** The full name of the DDPDataSink actor to write the list of tokens. */
    private String sinkActorName;
    
    /** Required field for IO formats. */
    private static final long serialVersionUID = 1L;

}