class SFTPPut extends BaseStep implements StepInterface {

  public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException {
    meta = (SFTPPutMeta) smi;
    data = (SFTPPutData) sdi;

    boolean sendToErrorRow = false;
    String errorMessage = null;

    Object[] r = getRow(); // get row, set busy!
    if ( r == null ) { // no more input to be expected...

      setOutputDone();
      return false;
    }

    if ( first ) {
      // Go there only for the first row received
      first = false;

      try {
        // String substitution..
        String realServerName = environmentSubstitute( meta.getServerName() );
        String realServerPort = environmentSubstitute( meta.getServerPort() );
        String realUsername = environmentSubstitute( meta.getUserName() );
        String realPassword =
          Encr.decryptPasswordOptionallyEncrypted( environmentSubstitute( meta.getPassword() ) );
        String realKeyFilename = null;
        String realPassPhrase = null;

        if ( meta.isUseKeyFile() ) {
          // We must have here a private keyfilename
          realKeyFilename = environmentSubstitute( meta.getKeyFilename() );
          if ( Utils.isEmpty( realKeyFilename ) ) {
            // Error..Missing keyfile
            logError( BaseMessages.getString( PKG, "SFTPPut.Error.KeyFileMissing" ) );
            return false;
          }
          if ( !KettleVFS.fileExists( realKeyFilename ) ) {
            // Error.. can not reach keyfile
            logError( BaseMessages.getString( PKG, "SFTPPut.Error.KeyFileNotFound", realKeyFilename ) );
            return false;
          }
          realPassPhrase = environmentSubstitute( meta.getKeyPassPhrase() );
        }

        // Let's try to establish SFTP connection....
        // Create sftp client to host ...
        data.sftpclient =
          createSftpClient( realServerName, realServerPort, realUsername, realKeyFilename, realPassPhrase );

        // connection successfully established
        if ( isDetailed() ) {
          logDetailed( BaseMessages.getString(
            PKG, "SFTPPUT.Log.OpenedConnection", realServerName, realServerPort, realUsername ) );
        }

        // Set compression
        data.sftpclient.setCompression( meta.getCompression() );

        // Set proxy?
        String realProxyHost = environmentSubstitute( meta.getProxyHost() );
        if ( !Utils.isEmpty( realProxyHost ) ) {
          // Set proxy
          data.sftpclient.setProxy(
            realProxyHost, environmentSubstitute( meta.getProxyPort() ), environmentSubstitute( meta
              .getProxyUsername() ), environmentSubstitute( meta.getProxyPassword() ), meta.getProxyType() );
        }

        // login to ftp host ...
        data.sftpclient.login( realPassword );

      } catch ( Exception e ) {
        throw new KettleException( BaseMessages.getString( PKG, "SFTPPUT.Error.Connection" ), e );
      }

      // Let's perform some checks

      checkSourceFileField( meta.getSourceFileFieldName(), data );

      checkRemoteFoldernameField( meta.getRemoteDirectoryFieldName(), data );

      checkRemoteFilenameField( meta.getRemoteFilenameFieldName(), data );

      // Move to folder
      if ( meta.getAfterFTPS() == JobEntrySFTPPUT.AFTER_FTPSPUT_MOVE ) {
        checkDestinationFolderField( meta.getDestinationFolderFieldName(), data );
      }
    }

    // Read data top upload
    String sourceData = getInputRowMeta().getString( r, data.indexOfSourceFileFieldName );

    InputStream inputStream = null;
    FileObject destinationFolder = null;
    String destinationFilename;
    FileObject file = null;

    try {
      if ( Utils.isEmpty( sourceData ) ) {
        // Source data is empty
        throw new KettleStepException( BaseMessages.getString( PKG, "SFTPPut.Error.SourceDataEmpty" ) );
      }

      if ( meta.isInputStream() ) {
        // Source data is a stream
        inputStream = new ByteArrayInputStream( getInputRowMeta().getBinary( r, data.indexOfSourceFileFieldName ) );

        if ( data.indexOfRemoteFilename == -1 ) {
          // for the case when put data is transferred via input field
          // it is mandatory to specify remote file name
          throw new KettleStepException( BaseMessages.getString( PKG, "SFTPPut.Error.RemoteFilenameFieldMissing" ) );
        }
        destinationFilename = getInputRowMeta().getString( r, data.indexOfRemoteFilename );
      } else {
        // source data is a file
        // let's check file
        file = KettleVFS.getFileObject( sourceData );

        if ( !file.exists() ) {
          // We can not find file
          throw new KettleStepException( BaseMessages.getString( PKG, "SFTPPut.Error.CanNotFindField", sourceData ) );
        }
        // get stream from file
        inputStream = KettleVFS.getInputStream( file );

        // Destination filename
        destinationFilename = file.getName().getBaseName();
      }

      if ( file != null ) {
        if ( meta.getAfterFTPS() == JobEntrySFTPPUT.AFTER_FTPSPUT_MOVE ) {
          String realDestationFolder = getInputRowMeta().getString( r, data.indexOfMoveToFolderFieldName );

          if ( Utils.isEmpty( realDestationFolder ) ) {
            // Move to destination folder is empty
            throw new KettleStepException( BaseMessages.getString(
              PKG, "SFTPPut.Error.MoveToDestinationFolderIsEmpty" ) );
          }

          destinationFolder = KettleVFS.getFileObject( realDestationFolder );

          if ( !destinationFolder.exists() ) {
            // We can not find folder
            throw new KettleStepException( BaseMessages.getString(
              PKG, "SFTPPut.Error.CanNotFindFolder", realDestationFolder ) );
          }
        }
      }

      // move to spool dir ...
      setSFTPDirectory( getInputRowMeta().getString( r, data.indexOfRemoteDirectory ) );

      // Upload a stream
      data.sftpclient.put( inputStream, destinationFilename );

      if ( file != null ) {
        // We successfully uploaded the file
        // what's next ...
        finishTheJob( file, sourceData, destinationFolder );
      }

      putRow( getInputRowMeta(), r ); // copy row to possible alternate rowset(s).

      if ( checkFeedback( getLinesRead() ) ) {
        if ( isDetailed() ) {
          logDetailed( BaseMessages.getString( PKG, "SFTPPut.Log.LineNumber" ) + getLinesRead() );
        }
      }
    } catch ( Exception e ) {

      if ( getStepMeta().isDoingErrorHandling() ) {
        sendToErrorRow = true;
        errorMessage = e.toString();
      } else {

        logError( BaseMessages.getString( PKG, "SFTPPut.Log.ErrorInStep" ), e );
        setErrors( 1 );
        stopAll();
        setOutputDone(); // signal end to receiver(s)
        return false;
      }

      if ( sendToErrorRow ) {
        // Simply add this row to the error row
        putError( getInputRowMeta(), r, 1, errorMessage, null, "SFTPPUT001" );
      }
    } finally {
      try {
        if ( destinationFolder != null ) {
          destinationFolder.close();
        }
        if ( file != null ) {
          file.close();
        }
        if ( inputStream != null ) {
          inputStream.close();
        }
      } catch ( Exception e ) {
        // ignore this
      }
    }
    return true;
  }

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

}