basic-lib/src/main/java/com/streamsets/pipeline/stage/origin/http/HttpClientSource.java
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    long start = System.currentTimeMillis();
    int chunksToFetch = Math.min(conf.basic.maxBatchSize, maxBatchSize);
    Optional<String> newSourceOffset = Optional.empty();
    recordCount = 0;

    setPageOffset(lastSourceOffset);

    setResolvedUrl(resolveInitialUrl(lastSourceOffset));
    WebTarget target = client.target(getResolvedUrl());

    // If the request (headers or body) contain a known sensitive EL and we're not using https then fail the request.
    if (requestContainsSensitiveInfo() && !target.getUri().getScheme().toLowerCase().startsWith("https")) {
      LOG.error(Errors.HTTP_07.getMessage());
      throw new StageException(Errors.HTTP_07);
    }

    boolean uninterrupted = true;

    while (!waitTimeExpired(start) && uninterrupted && (recordCount < chunksToFetch)) {
      if (parser != null) {
        // We already have an response that we haven't finished reading.
        newSourceOffset = Optional.of(parseResponse(start, chunksToFetch, batchMaker));
      } else if (shouldMakeRequest()) {

        if (conf.pagination.mode != PaginationMode.NONE) {
          target = client.target(resolveNextPageUrl(newSourceOffset.orElse(null)));
          // Pause between paging requests so we don't get rate limited.
          uninterrupted = ThreadUtil.sleep(conf.pagination.rateLimit);
        }

        makeRequest(target);
        if (lastRequestTimedOut) {
          String actionName = conf.responseTimeoutActionConfig.getAction().name();
          LOG.warn(
              "HTTPClient timed out after waiting {} ms for response from server;" +
              " reconnecting client and proceeding as per configured {} action",
              conf.client.readTimeoutMillis,
              actionName
          );
          reconnectClient();
          return nonTerminating(lastSourceOffset);
        } else {
          newSourceOffset = processResponse(start, chunksToFetch, batchMaker);
        }
      } else if (conf.httpMode == HttpClientMode.BATCH) {
        // We are done.
        return null;
      } else {
        // In polling mode, waiting for the next polling interval.
        uninterrupted = ThreadUtil.sleep(SLEEP_TIME_WAITING_FOR_BATCH_SIZE_MS);
      }
    }

    return newSourceOffset.orElse(lastSourceOffset);
  }
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    long start = System.currentTimeMillis();
    int chunksToFetch = Math.min(conf.basic.maxBatchSize, maxBatchSize);
    Optional<String> newSourceOffset = Optional.empty();
    recordCount = 0;

    setPageOffset(lastSourceOffset);

    setResolvedUrl(resolveInitialUrl(lastSourceOffset));
    WebTarget target = client.target(getResolvedUrl());

    // If the request (headers or body) contain a known sensitive EL and we're not using https then fail the request.
    if (requestContainsSensitiveInfo() && !target.getUri().getScheme().toLowerCase().startsWith("https")) {
      LOG.error(Errors.HTTP_07.getMessage());
      throw new StageException(Errors.HTTP_07);
    }

    boolean uninterrupted = true;

    while (!waitTimeExpired(start) && uninterrupted && (recordCount < chunksToFetch)) {
      if (parser != null) {
        // We already have an response that we haven't finished reading.
        newSourceOffset = Optional.of(parseResponse(start, chunksToFetch, batchMaker));
      } else if (shouldMakeRequest()) {

        if (conf.pagination.mode != PaginationMode.NONE) {
          target = client.target(resolveNextPageUrl(newSourceOffset.orElse(null)));
          // Pause between paging requests so we don't get rate limited.
          uninterrupted = ThreadUtil.sleep(conf.pagination.rateLimit);
        }

        makeRequest(target);
        if (lastRequestTimedOut) {
          String actionName = conf.responseTimeoutActionConfig.getAction().name();
          LOG.warn(
              "HTTPClient timed out after waiting {} ms for response from server;" +
              " reconnecting client and proceeding as per configured {} action",
              conf.client.readTimeoutMillis,
              actionName
          );
          reconnectClient();
          return nonTerminating(lastSourceOffset);
        } else {
          newSourceOffset = processResponse(start, chunksToFetch, batchMaker);
        }
      } else if (conf.httpMode == HttpClientMode.BATCH) {
        // We are done.
        return null;
      } else {
        // In polling mode, waiting for the next polling interval.
        uninterrupted = ThreadUtil.sleep(SLEEP_TIME_WAITING_FOR_BATCH_SIZE_MS);
      }
    }

    return newSourceOffset.orElse(lastSourceOffset);
  }
@ Override public String produce ( String lastSourceOffset , int maxBatchSize , BatchMaker batchMaker ) throws StageException { long start = System . currentTimeMillis ( ) ; int chunksToFetch = Math . min ( conf . basic . maxBatchSize , maxBatchSize ) ; Optional < String > newSourceOffset = Optional . empty ( ) ; recordCount = 0 ; setPageOffset ( lastSourceOffset ) ; setResolvedUrl ( resolveInitialUrl ( lastSourceOffset ) ) ; WebTarget target = client . target ( getResolvedUrl ( ) ) ; // If the request (headers or body) contain a known sensitive EL and we're not using https then fail the request. if ( requestContainsSensitiveInfo ( ) && ! target . getUri ( ) . getScheme ( ) . toLowerCase ( ) . startsWith ( "https" ) ) { LOG . error ( Errors . HTTP_07 . getMessage ( ) ) ; throw new StageException ( Errors . HTTP_07 ) ; } boolean uninterrupted = true ; while ( ! waitTimeExpired ( start ) && uninterrupted && ( recordCount < chunksToFetch ) ) { if ( parser != null ) { // We already have an response that we haven't finished reading. newSourceOffset = Optional . of ( parseResponse ( start , chunksToFetch , batchMaker ) ) ; } else if ( shouldMakeRequest ( ) ) { if ( conf . pagination . mode != PaginationMode . NONE ) { target = client . target ( resolveNextPageUrl ( newSourceOffset . orElse ( null ) ) ) ; // Pause between paging requests so we don't get rate limited. uninterrupted = ThreadUtil . sleep ( conf . pagination . rateLimit ) ; } makeRequest ( target ) ; if ( lastRequestTimedOut ) { String actionName = conf . responseTimeoutActionConfig . getAction ( ) . name ( ) ; LOG . warn ( "HTTPClient timed out after waiting {} ms for response from server;" + " reconnecting client and proceeding as per configured {} action" , conf . client . readTimeoutMillis , actionName ) ; reconnectClient ( ) ; return nonTerminating ( lastSourceOffset ) ; } else { newSourceOffset = processResponse ( start , chunksToFetch , batchMaker ) ; } } else if ( conf . httpMode == HttpClientMode . BATCH ) { // We are done. return null ; } else { // In polling mode, waiting for the next polling interval. uninterrupted = ThreadUtil . sleep ( SLEEP_TIME_WAITING_FOR_BATCH_SIZE_MS ) ; } } return newSourceOffset . orElse ( lastSourceOffset ) ; }
{@inheritDoc}
{