@Test public void testProcessRow() throws KettleException { PrioritizeStreamsMeta meta = new PrioritizeStreamsMeta(); meta.setStepName( new String[] { "high", "medium", "low" } ); PrioritizeStreamsData data = new PrioritizeStreamsData(); PrioritizeStreamsInner step = new PrioritizeStreamsInner( stepMockHelper ); try { step.processRow( meta, data ); } catch ( NullPointerException e ) { fail( "NullPointerException detecded, seems that RowMetaInterface was not set for RowSet you are attempting" + "to read from." ); } Assert.assertTrue( "First waiting for row set is 'high'", data.currentRowSet.getClass().equals( SingleRowRowSet.class ) ); }
public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (PrioritizeStreamsMeta) smi; data = (PrioritizeStreamsData) sdi; if ( first ) { if ( meta.getStepName() != null || meta.getStepName().length > 0 ) { data.stepnrs = meta.getStepName().length; data.rowSets = new RowSet[data.stepnrs]; for ( int i = 0; i < data.stepnrs; i++ ) { data.rowSets[i] = findInputRowSet( meta.getStepName()[i] ); if ( i > 0 ) { checkInputLayoutValid( data.rowSets[0].getRowMeta(), data.rowSets[i].getRowMeta() ); } } } else { throw new KettleException( BaseMessages.getString( PKG, "PrioritizeStreams.Error.NotInputSteps" ) ); } data.currentRowSet = data.rowSets[0]; } Object[] input = getOneRow(); while ( input == null && data.stepnr < data.stepnrs - 1 && !isStopped() ) { input = getOneRow(); } if ( input == null ) { setOutputDone(); return false; } if ( first ) { data.outputRowMeta = data.currentRowSet.getRowMeta(); first = false; } putRow( data.outputRowMeta, input ); return true; }
PrioritizeStreams extends BaseStep implements StepInterface { public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (PrioritizeStreamsMeta) smi; data = (PrioritizeStreamsData) sdi; if ( first ) { if ( meta.getStepName() != null || meta.getStepName().length > 0 ) { data.stepnrs = meta.getStepName().length; data.rowSets = new RowSet[data.stepnrs]; for ( int i = 0; i < data.stepnrs; i++ ) { data.rowSets[i] = findInputRowSet( meta.getStepName()[i] ); if ( i > 0 ) { checkInputLayoutValid( data.rowSets[0].getRowMeta(), data.rowSets[i].getRowMeta() ); } } } else { throw new KettleException( BaseMessages.getString( PKG, "PrioritizeStreams.Error.NotInputSteps" ) ); } data.currentRowSet = data.rowSets[0]; } Object[] input = getOneRow(); while ( input == null && data.stepnr < data.stepnrs - 1 && !isStopped() ) { input = getOneRow(); } if ( input == null ) { setOutputDone(); return false; } if ( first ) { data.outputRowMeta = data.currentRowSet.getRowMeta(); first = false; } putRow( data.outputRowMeta, input ); return true; } }
PrioritizeStreams extends BaseStep implements StepInterface { public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (PrioritizeStreamsMeta) smi; data = (PrioritizeStreamsData) sdi; if ( first ) { if ( meta.getStepName() != null || meta.getStepName().length > 0 ) { data.stepnrs = meta.getStepName().length; data.rowSets = new RowSet[data.stepnrs]; for ( int i = 0; i < data.stepnrs; i++ ) { data.rowSets[i] = findInputRowSet( meta.getStepName()[i] ); if ( i > 0 ) { checkInputLayoutValid( data.rowSets[0].getRowMeta(), data.rowSets[i].getRowMeta() ); } } } else { throw new KettleException( BaseMessages.getString( PKG, "PrioritizeStreams.Error.NotInputSteps" ) ); } data.currentRowSet = data.rowSets[0]; } Object[] input = getOneRow(); while ( input == null && data.stepnr < data.stepnrs - 1 && !isStopped() ) { input = getOneRow(); } if ( input == null ) { setOutputDone(); return false; } if ( first ) { data.outputRowMeta = data.currentRowSet.getRowMeta(); first = false; } putRow( data.outputRowMeta, input ); return true; } PrioritizeStreams( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr,
    TransMeta transMeta, Trans trans ); }
PrioritizeStreams extends BaseStep implements StepInterface { public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (PrioritizeStreamsMeta) smi; data = (PrioritizeStreamsData) sdi; if ( first ) { if ( meta.getStepName() != null || meta.getStepName().length > 0 ) { data.stepnrs = meta.getStepName().length; data.rowSets = new RowSet[data.stepnrs]; for ( int i = 0; i < data.stepnrs; i++ ) { data.rowSets[i] = findInputRowSet( meta.getStepName()[i] ); if ( i > 0 ) { checkInputLayoutValid( data.rowSets[0].getRowMeta(), data.rowSets[i].getRowMeta() ); } } } else { throw new KettleException( BaseMessages.getString( PKG, "PrioritizeStreams.Error.NotInputSteps" ) ); } data.currentRowSet = data.rowSets[0]; } Object[] input = getOneRow(); while ( input == null && data.stepnr < data.stepnrs - 1 && !isStopped() ) { input = getOneRow(); } if ( input == null ) { setOutputDone(); return false; } if ( first ) { data.outputRowMeta = data.currentRowSet.getRowMeta(); first = false; } putRow( data.outputRowMeta, input ); return true; } PrioritizeStreams( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr,
    TransMeta transMeta, Trans trans ); boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); boolean init( StepMetaInterface smi, StepDataInterface sdi ); void dispose( StepMetaInterface smi, StepDataInterface sdi ); }
PrioritizeStreams extends BaseStep implements StepInterface { public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (PrioritizeStreamsMeta) smi; data = (PrioritizeStreamsData) sdi; if ( first ) { if ( meta.getStepName() != null || meta.getStepName().length > 0 ) { data.stepnrs = meta.getStepName().length; data.rowSets = new RowSet[data.stepnrs]; for ( int i = 0; i < data.stepnrs; i++ ) { data.rowSets[i] = findInputRowSet( meta.getStepName()[i] ); if ( i > 0 ) { checkInputLayoutValid( data.rowSets[0].getRowMeta(), data.rowSets[i].getRowMeta() ); } } } else { throw new KettleException( BaseMessages.getString( PKG, "PrioritizeStreams.Error.NotInputSteps" ) ); } data.currentRowSet = data.rowSets[0]; } Object[] input = getOneRow(); while ( input == null && data.stepnr < data.stepnrs - 1 && !isStopped() ) { input = getOneRow(); } if ( input == null ) { setOutputDone(); return false; } if ( first ) { data.outputRowMeta = data.currentRowSet.getRowMeta(); first = false; } putRow( data.outputRowMeta, input ); return true; } PrioritizeStreams( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr,
    TransMeta transMeta, Trans trans ); boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); boolean init( StepMetaInterface smi, StepDataInterface sdi ); void dispose( StepMetaInterface smi, StepDataInterface sdi );  }