class OraBulkLoader extends BaseStep implements StepInterface {

  public String createCommandLine( OraBulkLoaderMeta meta, boolean password ) throws KettleException {
    StringBuilder sb = new StringBuilder( 300 );

    if ( meta.getSqlldr() != null ) {
      try {
        FileObject fileObject =
          KettleVFS.getFileObject( environmentSubstitute( meta.getSqlldr() ), getTransMeta() );
        String sqlldr = KettleVFS.getFilename( fileObject );
        sb.append( sqlldr );
      } catch ( KettleFileException ex ) {
        throw new KettleException( "Error retrieving sqlldr string", ex );
      }
    } else {
      throw new KettleException( "No sqlldr application specified" );
    }

    if ( meta.getControlFile() != null ) {
      try {
        FileObject fileObject =
          KettleVFS.getFileObject( environmentSubstitute( meta.getControlFile() ), getTransMeta() );

        sb.append( " control=\'" );
        sb.append( KettleVFS.getFilename( fileObject ) );
        sb.append( "\'" );
      } catch ( KettleFileException ex ) {
        throw new KettleException( "Error retrieving controlfile string", ex );
      }
    } else {
      throw new KettleException( "No control file specified" );
    }

    if ( OraBulkLoaderMeta.METHOD_AUTO_CONCURRENT.equals( meta.getLoadMethod() ) ) {
      sb.append( " data=\'-\'" );
    }

    if ( meta.getLogFile() != null ) {
      try {
        FileObject fileObject =
          KettleVFS.getFileObject( environmentSubstitute( meta.getLogFile() ), getTransMeta() );

        sb.append( " log=\'" );
        sb.append( KettleVFS.getFilename( fileObject ) );
        sb.append( "\'" );
      } catch ( KettleFileException ex ) {
        throw new KettleException( "Error retrieving logfile string", ex );
      }
    }

    if ( meta.getBadFile() != null ) {
      try {
        FileObject fileObject =
          KettleVFS.getFileObject( environmentSubstitute( meta.getBadFile() ), getTransMeta() );

        sb.append( " bad=\'" );
        sb.append( KettleVFS.getFilename( fileObject ) );
        sb.append( "\'" );
      } catch ( KettleFileException ex ) {
        throw new KettleException( "Error retrieving badfile string", ex );
      }
    }

    if ( meta.getDiscardFile() != null ) {
      try {
        FileObject fileObject =
          KettleVFS.getFileObject( environmentSubstitute( meta.getDiscardFile() ), getTransMeta() );

        sb.append( " discard=\'" );
        sb.append( KettleVFS.getFilename( fileObject ) );
        sb.append( "\'" );
      } catch ( KettleFileException ex ) {
        throw new KettleException( "Error retrieving discardfile string", ex );
      }
    }

    DatabaseMeta dm = meta.getDatabaseMeta();
    if ( dm != null ) {
      String user = Const.NVL( dm.getUsername(), "" );
      String pass =
        Const.NVL( Encr.decryptPasswordOptionallyEncrypted( environmentSubstitute( dm.getPassword() ) ), "" );
      if ( !password ) {
        pass = "******";
      }
      String dns = Const.NVL( dm.getDatabaseName(), "" );
      sb.append( " userid=" ).append( environmentSubstitute( user ) ).append( "/" ).append(
        environmentSubstitute( pass ) ).append( "@" );

      String overrideName = meta.getDbNameOverride();
      if ( Utils.isEmpty( Const.rtrim( overrideName ) ) ) {
        sb.append( environmentSubstitute( dns ) );
      } else {
        // if the database name override is filled in, do that one.
        sb.append( environmentSubstitute( overrideName ) );
      }
    } else {
      throw new KettleException( "No connection specified" );
    }

    if ( meta.isDirectPath() ) {
      sb.append( " DIRECT=TRUE" );

      if ( getStepMeta().getCopies() > 1 || meta.isParallel() ) {
        sb.append( " PARALLEL=TRUE" );
      }
    }

    return sb.toString();
  }

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

}