/* Copyright (c) 2007 Pentaho Corporation.  All rights reserved. 
 * This software was developed by Pentaho Corporation and is provided under the terms 
 * of the GNU Lesser General Public License, Version 2.1. You may not use 
 * this file except in compliance with the license. If you need a copy of the license, 
 * please go to http://www.gnu.org/licenses/lgpl-2.1.txt. The Original Code is Pentaho 
 * Data Integration.  The Initial Developer is Pentaho Corporation.
 *
 * Software distributed under the GNU Lesser Public License is distributed on an "AS IS" 
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or  implied. Please refer to 
 * the license for the specific language governing your rights and limitations.*/

package org.pentaho.di.trans.steps.filesfromresult;

import org.pentaho.di.core.Const;
import org.pentaho.di.core.Result;
import org.pentaho.di.core.ResultFile;
import org.pentaho.di.core.RowMetaAndData;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;

/**
 * Reads results from a previous transformation in a Job
 * 
 * @author Matt
 * @since 2-jun-2003
 */
public class FilesFromResult extends BaseStep implements StepInterface
{
	private FilesFromResultMeta meta;

	private FilesFromResultData data;

	public FilesFromResult(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr,
			TransMeta transMeta, Trans trans)
	{
		super(stepMeta, stepDataInterface, copyNr, transMeta, trans);

		meta = (FilesFromResultMeta) getStepMeta().getStepMetaInterface();
		data = (FilesFromResultData) stepDataInterface;
	}

	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
	{
		if (data.resultFilesList == null || linesRead >= data.resultFilesList.size())
		{
			setOutputDone();
			return false;
		}

		ResultFile resultFile = (ResultFile) data.resultFilesList.get((int) linesRead);
		RowMetaAndData r = resultFile.getRow();
		data.outputRowMeta = getInputRowMeta().clone();
		smi.getFields(data.outputRowMeta, getStepname(), null, null, this);
		linesRead++;

		putRow(data.outputRowMeta, r.getData()); // copy row to possible alternate
										// rowset(s).

		if (checkFeedback(linesRead))
			logBasic(Messages.getString("FilesFromResult.Log.LineNumber") + linesRead); //$NON-NLS-1$

		return true;
	}

	public boolean init(StepMetaInterface smi, StepDataInterface sdi)
	{
		meta = (FilesFromResultMeta) smi;
		data = (FilesFromResultData) sdi;

		if (super.init(smi, sdi))
		{
			Result result = getTransMeta().getPreviousResult();

			if (result != null)
			{
				data.resultFilesList = result.getResultFilesList();
			} else
			{
				data.resultFilesList = null;
			}

			// Add init code here.
			return true;
		}
		return false;
	}

	//
	// Run is were the action happens!
	public void run()
	{
		try
		{
			logBasic(Messages.getString("System.Log.StartingToRun")); //$NON-NLS-1$
			
			while (processRow(meta, data) && !isStopped());
		} 
		catch(Throwable t)
		{
			logError(Messages.getString("System.Log.UnexpectedError") + " : "); //$NON-NLS-1$ //$NON-NLS-2$
			logError(Const.getStackTracker(t));
			setErrors(1);
			stopAll();
		} 
		finally
		{
			dispose(meta, data);
			logSummary();
			markStop();
		}
	}
}
@@ -2225,16 +2225,20 @@
 etRow();
+%0D%0A%09%09
 %0A%09%09data.
@@ -2257,15 +2257,12 @@
 a = 
+r.
 ge
-tInpu
 tRow
@@ -2271,17 +2271,13 @@
 ta()
-.clone();
+;%0D%0A%09%09
 %0A%09%09s

Fix for PDI-533 : Get Files From result: throws NullPointerException