basic-lib/src/main/java/com/streamsets/pipeline/stage/origin/http/HttpClientSource.java
@VisibleForTesting
  String parseResponse(long start, int maxRecords, BatchMaker batchMaker) throws StageException {
    HttpSourceOffset sourceOffset = new HttpSourceOffset(
        getResolvedUrl(),
        currentParameterHash,
        System.currentTimeMillis(),
        getCurrentPage()
    );
    InputStream in = null;
    if (parser == null) {
      // Only get a new parser if we are done with the old one.
      in = getResponse().readEntity(InputStream.class);
      try {
        parser = parserFactory.getParser(sourceOffset.toString(), in, "0");
      } catch (DataParserException e) {
        if (e.getErrorCode() == JSON_PARSER_00) {
          LOG.warn("No data returned in HTTP response body.", e);
          return sourceOffset.toString();
        }
        LOG.warn("Error parsing response", e);
        throw e;
      }
    }

    Record record = null;
    int subRecordCount = 0;
    try {

      do {
        record = parser.parse();

        if (record == null) {
          break;
        }

        // LINK_FIELD pagination
        if (conf.pagination.mode == PaginationMode.LINK_FIELD) {
          // evaluate stopping condition
          RecordEL.setRecordInContext(stopVars, record);
          haveMorePages = !stopEval.eval(stopVars, conf.pagination.stopCondition, Boolean.class);
          if (haveMorePages) {
            next = Link.fromUri(record.get(conf.pagination.nextPageFieldPath).getValueAsString()).build();
          } else {
            next = null;
          }
        }

        if (conf.pagination.mode != PaginationMode.NONE && record.has(conf.pagination.resultFieldPath)) {
          subRecordCount = parsePaginatedResult(batchMaker, sourceOffset.toString(), record);
          recordCount += subRecordCount;
        } else {
          addResponseHeaders(record.getHeader());
          batchMaker.addRecord(record);
          ++recordCount;
        }

      } while (recordCount < maxRecords && !waitTimeExpired(start));

    } catch (IOException e) {
      LOG.error(Errors.HTTP_00.getMessage(), e.toString(), e);
      errorRecordHandler.onError(Errors.HTTP_00, e.toString(), e);

    } finally {
      try {
        if (record == null) {
          cleanupResponse(in);
        }
        if (subRecordCount != 0) {
          incrementSourceOffset(sourceOffset, subRecordCount);
        }
      } catch(IOException e) {
        LOG.warn(Errors.HTTP_28.getMessage(), e.toString(), e);
        errorRecordHandler.onError(Errors.HTTP_28, e.toString(), e);
      }
    }

    return sourceOffset.toString();
  }
@VisibleForTesting
  String parseResponse(long start, int maxRecords, BatchMaker batchMaker) throws StageException {
    HttpSourceOffset sourceOffset = new HttpSourceOffset(
        getResolvedUrl(),
        currentParameterHash,
        System.currentTimeMillis(),
        getCurrentPage()
    );
    InputStream in = null;
    if (parser == null) {
      // Only get a new parser if we are done with the old one.
      in = getResponse().readEntity(InputStream.class);
      try {
        parser = parserFactory.getParser(sourceOffset.toString(), in, "0");
      } catch (DataParserException e) {
        if (e.getErrorCode() == JSON_PARSER_00) {
          LOG.warn("No data returned in HTTP response body.", e);
          return sourceOffset.toString();
        }
        LOG.warn("Error parsing response", e);
        throw e;
      }
    }

    Record record = null;
    int subRecordCount = 0;
    try {

      do {
        record = parser.parse();

        if (record == null) {
          break;
        }

        // LINK_FIELD pagination
        if (conf.pagination.mode == PaginationMode.LINK_FIELD) {
          // evaluate stopping condition
          RecordEL.setRecordInContext(stopVars, record);
          haveMorePages = !stopEval.eval(stopVars, conf.pagination.stopCondition, Boolean.class);
          if (haveMorePages) {
            next = Link.fromUri(record.get(conf.pagination.nextPageFieldPath).getValueAsString()).build();
          } else {
            next = null;
          }
        }

        if (conf.pagination.mode != PaginationMode.NONE && record.has(conf.pagination.resultFieldPath)) {
          subRecordCount = parsePaginatedResult(batchMaker, sourceOffset.toString(), record);
          recordCount += subRecordCount;
        } else {
          addResponseHeaders(record.getHeader());
          batchMaker.addRecord(record);
          ++recordCount;
        }

      } while (recordCount < maxRecords && !waitTimeExpired(start));

    } catch (IOException e) {
      LOG.error(Errors.HTTP_00.getMessage(), e.toString(), e);
      errorRecordHandler.onError(Errors.HTTP_00, e.toString(), e);

    } finally {
      try {
        if (record == null) {
          cleanupResponse(in);
        }
        if (subRecordCount != 0) {
          incrementSourceOffset(sourceOffset, subRecordCount);
        }
      } catch(IOException e) {
        LOG.warn(Errors.HTTP_28.getMessage(), e.toString(), e);
        errorRecordHandler.onError(Errors.HTTP_28, e.toString(), e);
      }
    }

    return sourceOffset.toString();
  }
@ VisibleForTesting String parseResponse ( long start , int maxRecords , BatchMaker batchMaker ) throws StageException { HttpSourceOffset sourceOffset = new HttpSourceOffset ( getResolvedUrl ( ) , currentParameterHash , System . currentTimeMillis ( ) , getCurrentPage ( ) ) ; InputStream in = null ; if ( parser == null ) { // Only get a new parser if we are done with the old one. in = getResponse ( ) . readEntity ( InputStream . class ) ; try { parser = parserFactory . getParser ( sourceOffset . toString ( ) , in , "0" ) ; } catch ( DataParserException e ) { if ( e . getErrorCode ( ) == JSON_PARSER_00 ) { LOG . warn ( "No data returned in HTTP response body." , e ) ; return sourceOffset . toString ( ) ; } LOG . warn ( "Error parsing response" , e ) ; throw e ; } } Record record = null ; int subRecordCount = 0 ; try { do { record = parser . parse ( ) ; if ( record == null ) { break ; } // LINK_FIELD pagination if ( conf . pagination . mode == PaginationMode . LINK_FIELD ) { // evaluate stopping condition RecordEL . setRecordInContext ( stopVars , record ) ; haveMorePages = ! stopEval . eval ( stopVars , conf . pagination . stopCondition , Boolean . class ) ; if ( haveMorePages ) { next = Link . fromUri ( record . get ( conf . pagination . nextPageFieldPath ) . getValueAsString ( ) ) . build ( ) ; } else { next = null ; } } if ( conf . pagination . mode != PaginationMode . NONE && record . has ( conf . pagination . resultFieldPath ) ) { subRecordCount = parsePaginatedResult ( batchMaker , sourceOffset . toString ( ) , record ) ; recordCount += subRecordCount ; } else { addResponseHeaders ( record . getHeader ( ) ) ; batchMaker . addRecord ( record ) ; ++ recordCount ; } } while ( recordCount < maxRecords && ! waitTimeExpired ( start ) ) ; } catch ( IOException e ) { LOG . error ( Errors . HTTP_00 . getMessage ( ) , e . toString ( ) , e ) ; errorRecordHandler . onError ( Errors . HTTP_00 , e . toString ( ) , e ) ; } finally { try { if ( record == null ) { cleanupResponse ( in ) ; } if ( subRecordCount != 0 ) { incrementSourceOffset ( sourceOffset , subRecordCount ) ; } } catch ( IOException e ) { LOG . warn ( Errors . HTTP_28 . getMessage ( ) , e . toString ( ) , e ) ; errorRecordHandler . onError ( Errors . HTTP_28 , e . toString ( ) , e ) ; } } return sourceOffset . toString ( ) ; }
Parses the response of a completed request into records and adds them to the batch.
If more records are available in the response than we can add to the batch, the
response is not closed and parsing will continue on the next batch.

@param maxRecords maximum number of records to add to the batch.
@param batchMaker batch to add records to.
@return the next source offset to commit
@throws StageException if an unhandled error is encountered
Parses the response of a completed request into records and adds them to the batch . If more records are available in the response than we can add to the batch the response is not closed and parsing will continue on the next batch .