package com.datastax.bdp.cassandra.index.solr;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.apache.lucene.document.Document;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.core.PluginInfo;
import org.apache.solr.core.SolrCore;
import org.apache.solr.schema.SchemaField;
import org.apache.solr.util.plugin.PluginInfoInitialized;

public class CopyFieldInputTransformer extends FieldInputTransformer implements PluginInfoInitialized
{
    private volatile SolrCore core;
    private volatile Set<String> sourceFields = new HashSet<String>();
    private volatile String destinationField;

    public CopyFieldInputTransformer(SolrCore core)
    {
        this.core = core;
    }

    public void init(PluginInfo info) {
    	NamedList args = info.initArgs;

    	String source = (String)args.get("source");

    	if (StringUtils.isEmpty(source)) {
    		throw new RuntimeException("source must be defined");
    	}

    	String[] sources = StringUtils.split(source, ',');
    	for (String s : sources) {
    		this.sourceFields.add(s);
    	}

    	destinationField = (String)args.get("destination");

    	if (StringUtils.isEmpty(destinationField)) {
    		throw new RuntimeException("destination must be defined");
    	}
    }

    @Override
    public boolean evaluate(String field)
    {
        return (sourceFields.contains(field));
    }

    @Override
    public void addFieldToDocument(String fieldName, float docBoost,
            String key, Document doc, SchemaField fieldInfo, String value,
            boolean makeLazy, SolrSecondaryIndex ssi) throws IOException
    {
        try
        {
            SchemaField destinationSchemaField = core.getSchema().getFieldOrNull(destinationField);
            if (sourceFields.contains(fieldName))
            {
                SolrSecondaryIndex.addFieldToDocument(core, docBoost, key, doc,
                		destinationSchemaField, value, false);
            }
        }
        catch (Exception ex)
        {
            throw new RuntimeException(ex);
        }
    }
}