class BaseStreamStep extends BaseStep {

  @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException {
    Preconditions.checkArgument( first,
      BaseMessages.getString( PKG, "BaseStreamStep.ProcessRowsError" ) );
    Preconditions.checkNotNull( source );
    Preconditions.checkNotNull( window );

    source.open();

    bufferStream().forEach( result -> {
      if ( result.isSafeStop() ) {
        getTrans().safeStop();
      }

      putRows( result.getRows() );
    } );
    super.setOutputDone();

    // Needed for when an Abort Step is used.
    source.close();
    return false;
  }

  public  BaseStreamStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr,
                         TransMeta transMeta, Trans trans );

}