package be.ibridge.kettle.trans.step.blockingstep;

import be.ibridge.kettle.core.Const;
import be.ibridge.kettle.core.Row;
import be.ibridge.kettle.core.exception.KettleException;
import be.ibridge.kettle.trans.Trans;
import be.ibridge.kettle.trans.TransMeta;
import be.ibridge.kettle.trans.step.BaseStep;
import be.ibridge.kettle.trans.step.StepDataInterface;
import be.ibridge.kettle.trans.step.StepInterface;
import be.ibridge.kettle.trans.step.StepMeta;
import be.ibridge.kettle.trans.step.StepMetaInterface;

public class BlockingStep extends BaseStep implements StepInterface {

    private StepMeta meta;
    private StepDataInterface data;
    private Row lastRow;
    
    public BlockingStep(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta, Trans trans)
    {
        super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
    }
    
    
    public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
        Row r=getRow();       // Get row from input rowset & set row busy!
        if (r==null)  // no more input to be expected...
        {
            if(lastRow != null) {
                putRow(lastRow);
            }
            setOutputDone();
            return false;
        }
        
        lastRow = r;
        return true;
    }

    //
    // Run is were the action happens!
    public void run()
    {
        try
        {
            while (processRow((StepMetaInterface)meta, data) && !isStopped());
        }
        catch(Exception e)
        {
            logError("Unexpected error in '"+" : "+e.toString());
            logError(Const.getStackTracker(e));
            setErrors(1);
            stopAll();
        }
        finally
        {
            logSummary();
            markStop();
        }
    }
}

@@ -603,15 +603,24 @@
 Meta
+Interface
  meta;%0A
-
     
@@ -890,24 +890,261 @@
 %0A    %7D%0A    %0A
+%09public boolean init(StepMetaInterface smi, StepDataInterface sdi)%0A%09%7B%0A%09%09meta=(BlockingStepMeta)smi;%0A%09%09data=(BlockingStepData)sdi;%0A%09%09%0A%09%09if (super.init(smi, sdi))%0A%09%09%7B%0A%09%09    // Add init code here.%0A%09%09    return true;%0A%09%09%7D%0A%09%09return false;%0A%09%7D %0A
     %0A    pub
@@ -1671,32 +1671,73 @@
   try%0A        %7B%0A
+        %09logBasic(%22Starting to run...%22);%0A
             whil
@@ -1754,27 +1754,8 @@
 Row(
-(StepMetaInterface)
 meta
@@ -2018,32 +2018,62 @@
 nally%0A        %7B%0A
+        %09dispose(meta, data);%0A
             logS
@@ -2122,10 +2122,9 @@
 %7D%0A    %7D%0A
+
 %7D
-%0A

Fix for #4602 : put the blocking step in finished mode when finished.