redis-lib/src/main/java/com/streamsets/pipeline/stage/origin/redis/RedisSubscriptionSource.java
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    // Offsets can vary depending on the data source. Here we use an integer as an example only.
    long nextSourceOffset = 0;
    lastSourceOffset = lastSourceOffset == null ? "" : lastSourceOffset;
    if (!lastSourceOffset.equals("")) {
      nextSourceOffset = Long.parseLong(lastSourceOffset);
    }

    int recordCounter = 0;
    long startTime = System.currentTimeMillis();
    int maxRecords = Math.min(maxBatchSize, conf.maxBatchSize);

    while (recordCounter < maxRecords && (startTime + conf.maxWaitTime) > System.currentTimeMillis()) {
      String message = buffer.poll();
      if (null == message) {
        try {
          Thread.sleep(100);
        } catch (Exception e) {
          LOG.debug(e.getMessage(), e);
          break;
        }
      } else {
        List<Record> records = processRedisMessage("id::" + nextSourceOffset, message);
        for (Record record : records) {
          batchMaker.addRecord(record);
        }
        recordCounter += records.size();
        ++nextSourceOffset;
      }
    }
    return lastSourceOffset;
  }
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    // Offsets can vary depending on the data source. Here we use an integer as an example only.
    long nextSourceOffset = 0;
    lastSourceOffset = lastSourceOffset == null ? "" : lastSourceOffset;
    if (!lastSourceOffset.equals("")) {
      nextSourceOffset = Long.parseLong(lastSourceOffset);
    }

    int recordCounter = 0;
    long startTime = System.currentTimeMillis();
    int maxRecords = Math.min(maxBatchSize, conf.maxBatchSize);

    while (recordCounter < maxRecords && (startTime + conf.maxWaitTime) > System.currentTimeMillis()) {
      String message = buffer.poll();
      if (null == message) {
        try {
          Thread.sleep(100);
        } catch (Exception e) {
          LOG.debug(e.getMessage(), e);
          break;
        }
      } else {
        List<Record> records = processRedisMessage("id::" + nextSourceOffset, message);
        for (Record record : records) {
          batchMaker.addRecord(record);
        }
        recordCounter += records.size();
        ++nextSourceOffset;
      }
    }
    return lastSourceOffset;
  }
@ Override public String produce ( String lastSourceOffset , int maxBatchSize , BatchMaker batchMaker ) throws StageException { // Offsets can vary depending on the data source. Here we use an integer as an example only. long nextSourceOffset = 0 ; lastSourceOffset = lastSourceOffset == null ? "" : lastSourceOffset ; if ( ! lastSourceOffset . equals ( "" ) ) { nextSourceOffset = Long . parseLong ( lastSourceOffset ) ; } int recordCounter = 0 ; long startTime = System . currentTimeMillis ( ) ; int maxRecords = Math . min ( maxBatchSize , conf . maxBatchSize ) ; while ( recordCounter < maxRecords && ( startTime + conf . maxWaitTime ) > System . currentTimeMillis ( ) ) { String message = buffer . poll ( ) ; if ( null == message ) { try { Thread . sleep ( 100 ) ; } catch ( Exception e ) { LOG . debug ( e . getMessage ( ) , e ) ; break ; } } else { List < Record > records = processRedisMessage ( "id::" + nextSourceOffset , message ) ; for ( Record record : records ) { batchMaker . addRecord ( record ) ; } recordCounter += records . size ( ) ; ++ nextSourceOffset ; } } return lastSourceOffset ; }
{@inheritDoc}
{