/* A Stratosphere Reduce stub that executes a ScriptEngine.
 * 
 * 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.stub;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import javax.script.Invocable;
import javax.script.ScriptEngine;
import javax.script.ScriptEngineFactory;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.kepler.ddp.Utilities;
import org.kepler.stratosphere.type.TypeUtilities;

import eu.stratosphere.api.java.record.functions.ReduceFunction;
import eu.stratosphere.configuration.Configuration;
import eu.stratosphere.types.Key;
import eu.stratosphere.types.Record;
import eu.stratosphere.types.Value;
import eu.stratosphere.util.Collector;
import ptolemy.actor.TypeAttribute;
import ptolemy.data.type.Type;
import ptolemy.kernel.util.Workspace;

/** A Stratosphere Reduce stub that executes a ScriptEngine. The engine
 *  is loaded in configure(), and the reduce() function invoked each time
 *  this class's reduce() is called.
 * 
 *  @author Daniel Crawl
 *  @version $Id: ScriptEngineReduceStub.java 33628 2015-08-24 22:42:20Z crawl $
 */
public class ScriptEngineReduceStub extends ReduceFunction {

    /*
    public CodeReduceStub() {
    }
    */
    
    @Override
    public void close() throws Exception {
        super.close();
        _engine = null;
        _inputKeyType = null;
        _inputValueType = null;
        _resultKeys = null;
        _resultValues = null;
    }
    
    @Override
    public void open(Configuration parameters) throws Exception {
        
        // get the script engine factory name
        String scriptEngineFactoryName = 
                parameters.getString(Utilities.CONFIGURATION_KEPLER_SCRIPT_ENGINE_FACTORY_NAME, null);
        if(scriptEngineFactoryName == null) {
            throw new RuntimeException("No script engine factory specified.");
        }
        
        // get the code
        String codeStr = parameters.getString(Utilities.CONFIGURATION_KEPLER_STUB_CODE, null);
        if(codeStr == null) {
            throw new RuntimeException("No code specified.");
        }

        // create the engine
        
        try {
            Class<?> clazz = Class.forName(scriptEngineFactoryName);
            ScriptEngineFactory factory = (ScriptEngineFactory)clazz.newInstance();
            _engine = factory.getScriptEngine();
        } catch(Throwable t) {
            throw new RuntimeException("Error creating script engine for " + scriptEngineFactoryName, t);
        }
        
        if(_engine == null) {
            throw new RuntimeException("Could not find script engine for " + scriptEngineFactoryName);
        }
        
        // parse the code
        _engine.eval(codeStr);
        
        if(!(_engine instanceof Invocable)) {
            throw new RuntimeException("Script engine for " + scriptEngineFactoryName + " is not invocable.");
        }

        _invocable = (Invocable) _engine;

        
        StubUtilities.initializePtolemy();

        // get the key and value types

        TypeAttribute typeAttribute = new TypeAttribute(new Workspace());
        
        String keyTypeStr = parameters.getString(Utilities.CONFIGURATION_KEPLER_INPUT_KEY_TYPE, null);
        if(keyTypeStr == null) {
            throw new RuntimeException("No key type specified.");
        }

        typeAttribute.setExpression(keyTypeStr);
        Type keyType = typeAttribute.getType();
        _inputKeyType = TypeUtilities.getPactKeyTypeForPtolemyType(keyType);

        String valueTypeStr = parameters.getString(Utilities.CONFIGURATION_KEPLER_INPUT_VALUE_TYPE, null);
        if(valueTypeStr == null) {
            throw new RuntimeException("No value type specified.");
        }
        typeAttribute.setExpression(valueTypeStr);
        Type valueType = typeAttribute.getType();
        _inputValueType = TypeUtilities.getPactValueTypeForPtolemyType(valueType);
        
        _resultKeys = new LinkedList<Object>();
        _resultValues = new LinkedList<Object>();

        //LOG.info("reduce types: ik = " + _inputKeyType + " iv = " + _inputValueType);
        
    }

    @Override
    public void reduce(Iterator<Record> records, Collector<Record> out)
            throws Exception {

        // get key and value from record and put in engine
        Object key = null;
        List<Object> values = new LinkedList<Object>();
        
        while(records.hasNext()) {
            final Record record = records.next();
            
            if(key == null) {
                key = StubUtilities.convertPactDataToObject(record, TypeUtilities.KEY_FIELD, _inputKeyType);
            }
                
            values.add(StubUtilities.convertPactDataToObject(record, TypeUtilities.VALUE_FIELD, _inputValueType));
        }

        //System.out.println("input key: " + key + " values = " + values);
        
        _invocable.invokeFunction("reduce", key, values, _resultKeys, _resultValues);
                
        //System.out.println("result keys: " + _resultKeys);
        //System.out.println("result values: " + _resultValues);
        
        // convert results to collector
        StubUtilities.convertResultsToCollector(_resultKeys, _resultValues, out);        

        // clear the lists since the results are no longer needed.
        _resultKeys.clear();
        _resultValues.clear();
        
    }

    /** Input key type. */
    private Class<? extends Key> _inputKeyType;

    /** Input value type. */
    private Class<? extends Value> _inputValueType;

    /** Logging. */
    private static final Log LOG = LogFactory.getLog(ScriptEngineReduceStub.class);

    /** The engine to parse scripts. */
    private ScriptEngine _engine;
    
    /** Interface to execute functions in scripts. */
    private Invocable _invocable;
    
    /** A list of result keys. */
    private List<?> _resultKeys;
    
    /** A list of result values. */
    private List<?> _resultValues;

}