public boolean init( StepMetaInterface smi, StepDataInterface sdi ) {
    meta = (MailInputMeta) smi;
    data = (MailInputData) sdi;

    if ( !super.init( smi, sdi ) ) {
      return false;
    }

    if ( !meta.isDynamicFolder() ) {
      try {
        // Create the output row meta-data
        data.outputRowMeta = new RowMeta();
        meta.getFields( data.outputRowMeta, getStepname(), null, null, this, repository, metaStore ); // get the
                                                                                                      // metadata
                                                                                                      // populated

      } catch ( Exception e ) {
        logError( BaseMessages.getString( PKG, "MailInput.ErrorInit", e.toString() ) );
        logError( Const.getStackTracker( e ) );
        return false;
      }
    }
    data.usePOP = meta.getProtocol().equals( MailConnectionMeta.PROTOCOL_STRING_POP3 );

    String realserver = environmentSubstitute( meta.getServerName() );
    if ( meta.getProtocol().equals( MailConnectionMeta.PROTOCOL_STRING_MBOX )
      && StringUtils.startsWith( realserver, "file://" ) ) {
      realserver = StringUtils.remove( realserver, "file://" );
    }

    String realusername = environmentSubstitute( meta.getUserName() );
    String realpassword = Utils.resolvePassword( variables, meta.getPassword() );
    int realport = Const.toInt( environmentSubstitute( meta.getPort() ), -1 );
    String realProxyUsername = environmentSubstitute( meta.getProxyUsername() );
    if ( !meta.isDynamicFolder() ) {
      //Limit field has absolute priority
      String reallimitrow = environmentSubstitute( meta.getRowLimit() );
      int limit = Const.toInt( reallimitrow, 0 );
      //Limit field has absolute priority
      if ( limit == 0 ) {
        limit = getReadFirst( meta.getProtocol() );
      }
      data.rowlimit = limit;
    }
    Date beginDate = null;
    Date endDate = null;
    SimpleDateFormat df = new SimpleDateFormat( MailInputMeta.DATE_PATTERN );

    // check search terms
    // Received Date
    try {
      switch ( meta.getConditionOnReceivedDate() ) {
        case MailConnectionMeta.CONDITION_DATE_EQUAL:
        case MailConnectionMeta.CONDITION_DATE_GREATER:
        case MailConnectionMeta.CONDITION_DATE_SMALLER:
          String realBeginDate = environmentSubstitute( meta.getReceivedDate1() );
          if ( Utils.isEmpty( realBeginDate ) ) {
            throw new KettleException( BaseMessages.getString(
              PKG, "MailInput.Error.ReceivedDateSearchTermEmpty" ) );
          }
          beginDate = df.parse( realBeginDate );
          break;
        case MailConnectionMeta.CONDITION_DATE_BETWEEN:
          realBeginDate = environmentSubstitute( meta.getReceivedDate1() );
          if ( Utils.isEmpty( realBeginDate ) ) {
            throw new KettleException( BaseMessages.getString(
              PKG, "MailInput.Error.ReceivedDatesSearchTermEmpty" ) );
          }
          beginDate = df.parse( realBeginDate );
          String realEndDate = environmentSubstitute( meta.getReceivedDate2() );
          if ( Utils.isEmpty( realEndDate ) ) {
            throw new KettleException( BaseMessages.getString(
              PKG, "MailInput.Error.ReceivedDatesSearchTermEmpty" ) );
          }
          endDate = df.parse( realEndDate );
          break;
        default:
          break;
      }
    } catch ( Exception e ) {
      logError( BaseMessages.getString( PKG, "MailInput.Error.SettingSearchTerms", e.getMessage() ) );
      setErrors( 1 );
      stopAll();
    }
    try {
      // create a mail connection object
      data.mailConn =
        new MailConnection(
          log, MailConnectionMeta.getProtocolFromString(
            meta.getProtocol(), MailConnectionMeta.PROTOCOL_IMAP ), realserver, realport, realusername,
          realpassword, meta.isUseSSL(), meta.isUseProxy(), realProxyUsername );
      // connect
      data.mailConn.connect();
      // Need to apply search filters?
      applySearch( beginDate, endDate );

      if ( !meta.isDynamicFolder() ) {
        // pass static folder name
        String realIMAPFolder = environmentSubstitute( meta.getIMAPFolder() );
        // return folders list
        // including sub folders if necessary
        data.folders = getFolders( realIMAPFolder );
      }
    } catch ( Exception e ) {
      logError( BaseMessages.getString( PKG, "MailInput.Error.OpeningConnection", e.getMessage() ) );
      setErrors( 1 );
      stopAll();
    }
    data.nrFields = meta.getInputFields() != null ? meta.getInputFields().length : 0;

    return true;
  }
@Test
  public void testInitSetGetFirstForPOP3() {
    MailInput step =
        new MailInput( mockHelper.stepMeta, mockHelper.stepDataInterface, 0, mockHelper.transMeta, mockHelper.trans );
    MailInputData data = new MailInputData();
    MailInputMeta meta = mock( MailInputMeta.class );
    when( meta.isDynamicFolder() ).thenReturn( false );
    when( meta.getProtocol() ).thenReturn( MailConnectionMeta.PROTOCOL_STRING_POP3 );
    when( meta.getFirstIMAPMails() ).thenReturn( "2" );
    when( meta.getFirstMails() ).thenReturn( "3" );

    step.init( meta, data );

    Assert.assertEquals( "Row Limit is set up to 3 rows.", 3, data.rowlimit );
  }