package com.instaclick.pentaho.plugin.riak.processor;

import com.basho.riak.client.bucket.Bucket;
import com.instaclick.pentaho.plugin.riak.RiakPlugin;
import com.instaclick.pentaho.plugin.riak.RiakPluginData;

abstract class AbstractProcessor implements Processor
{
    protected final RiakPluginData data;
    protected final RiakPlugin plugin;
    protected final Bucket bucket;

    public AbstractProcessor(final Bucket bucket, final RiakPlugin plugin, final RiakPluginData data)
    {
        this.plugin = plugin;
        this.bucket = bucket;
        this.data   = data;
    }

    protected String getRiakKey(final Object[] r) throws Exception
    {
        if (r.length > data.keyFieldIndex) {
            return (r[data.keyFieldIndex] == null) ? null : String.valueOf(r[data.keyFieldIndex]);
        }

        final String message = plugin.getLinesRead() + " - Invalid key row";

        if (plugin.isDebug()) {
            plugin.logDebug(message);
        }

        plugin.putError(plugin.getInputRowMeta(), r, 1, message, null, "ICRiakPlugin001");

        return null;
    }

    protected String getRiakValue(final Object[] r) throws Exception
    {
        if (r.length > data.valueFieldIndex) {
            return (r[data.valueFieldIndex] == null) ? null : String.valueOf(r[data.valueFieldIndex]);
        }

        final String message = plugin.getLinesRead() + " - Invalid value row";

        if (plugin.isDebug()) {
            plugin.logDebug(message);
        }

        plugin.putError(plugin.getInputRowMeta(), r, 1, message, null, "ICRiakPlugin002");

        return null;
    }
}

@@ -754,39 +754,24 @@
 l) ? null : 
-String.valueOf(
 r%5Bdata.keyFi
@@ -771,32 +771,42 @@
 a.keyFieldIndex%5D
+.toString(
 );%0A        %7D%0A%0A  
@@ -1270,23 +1270,8 @@
 l : 
-String.valueOf(
 r%5Bda
@@ -1289,16 +1289,26 @@
 ldIndex%5D
+.toString(
 );%0A     

use toString instead of String.valueOf