basic-lib/src/main/java/com/streamsets/pipeline/stage/processor/http/HttpProcessor.java
@Override
  public void process(Batch batch, SingleLaneBatchMaker batchMaker) throws StageException {
    List<Future<Response>> responses = new ArrayList<>();
    resolvedRecords.clear();

    Iterator<Record> records = batch.getRecords();
    while (records.hasNext()) {

      Record record = records.next();
      String resolvedUrl = httpClientCommon.getResolvedUrl(conf.resourceUrl, record);
      WebTarget target = httpClientCommon.getClient().target(resolvedUrl);

      LOG.debug("Resolved HTTP Client URL: '{}'",resolvedUrl);

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

      // from HttpStreamConsumer
      final MultivaluedMap<String, Object> resolvedHeaders = httpClientCommon.resolveHeaders(conf.headers, record);

      String contentType = HttpStageUtil.getContentTypeWithDefault(resolvedHeaders, conf.defaultRequestContentType);

      final AsyncInvoker asyncInvoker = target.request()
          .property(OAuth1ClientSupport.OAUTH_PROPERTY_ACCESS_TOKEN, httpClientCommon.getAuthToken())
          .headers(resolvedHeaders)
          .async();

      HttpMethod method = httpClientCommon.getHttpMethod(conf.httpMethod, conf.methodExpression, record);

      rateLimiter.acquire();
      if (conf.requestBody != null && !conf.requestBody.isEmpty() && method != HttpMethod.GET) {
        RecordEL.setRecordInContext(bodyVars, record);
        final String requestBody = bodyEval.eval(bodyVars, conf.requestBody, String.class);
        resolvedRecords.put(record, new HeadersAndBody(resolvedHeaders, requestBody, contentType, method, target));
        responses.add(asyncInvoker.method(method.getLabel(), Entity.entity(requestBody, contentType)));
      } else {
        resolvedRecords.put(record, new HeadersAndBody(resolvedHeaders, null, null, method, target));
        responses.add(asyncInvoker.method(method.getLabel()));
      }
    }

    records = batch.getRecords();
    int recordNum = 0;
    while (records.hasNext()) {
      try {
        Record record = processResponse(records.next(), responses.get(recordNum), conf.maxRequestCompletionSecs, false);
        if (record != null) {
          batchMaker.addRecord(record);
        }
      } catch (OnRecordErrorException e) {
        errorRecordHandler.onError(e);
      } finally {
        ++recordNum;
      }
    }
    if (!resolvedRecords.isEmpty()) {
      reprocessIfRequired(batchMaker);
    }
  }
@Override
  public void process(Batch batch, SingleLaneBatchMaker batchMaker) throws StageException {
    List<Future<Response>> responses = new ArrayList<>();
    resolvedRecords.clear();

    Iterator<Record> records = batch.getRecords();
    while (records.hasNext()) {

      Record record = records.next();
      String resolvedUrl = httpClientCommon.getResolvedUrl(conf.resourceUrl, record);
      WebTarget target = httpClientCommon.getClient().target(resolvedUrl);

      LOG.debug("Resolved HTTP Client URL: '{}'",resolvedUrl);

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

      // from HttpStreamConsumer
      final MultivaluedMap<String, Object> resolvedHeaders = httpClientCommon.resolveHeaders(conf.headers, record);

      String contentType = HttpStageUtil.getContentTypeWithDefault(resolvedHeaders, conf.defaultRequestContentType);

      final AsyncInvoker asyncInvoker = target.request()
          .property(OAuth1ClientSupport.OAUTH_PROPERTY_ACCESS_TOKEN, httpClientCommon.getAuthToken())
          .headers(resolvedHeaders)
          .async();

      HttpMethod method = httpClientCommon.getHttpMethod(conf.httpMethod, conf.methodExpression, record);

      rateLimiter.acquire();
      if (conf.requestBody != null && !conf.requestBody.isEmpty() && method != HttpMethod.GET) {
        RecordEL.setRecordInContext(bodyVars, record);
        final String requestBody = bodyEval.eval(bodyVars, conf.requestBody, String.class);
        resolvedRecords.put(record, new HeadersAndBody(resolvedHeaders, requestBody, contentType, method, target));
        responses.add(asyncInvoker.method(method.getLabel(), Entity.entity(requestBody, contentType)));
      } else {
        resolvedRecords.put(record, new HeadersAndBody(resolvedHeaders, null, null, method, target));
        responses.add(asyncInvoker.method(method.getLabel()));
      }
    }

    records = batch.getRecords();
    int recordNum = 0;
    while (records.hasNext()) {
      try {
        Record record = processResponse(records.next(), responses.get(recordNum), conf.maxRequestCompletionSecs, false);
        if (record != null) {
          batchMaker.addRecord(record);
        }
      } catch (OnRecordErrorException e) {
        errorRecordHandler.onError(e);
      } finally {
        ++recordNum;
      }
    }
    if (!resolvedRecords.isEmpty()) {
      reprocessIfRequired(batchMaker);
    }
  }
@ Override public void process ( Batch batch , SingleLaneBatchMaker batchMaker ) throws StageException { List < Future < Response >> responses = new ArrayList <> ( ) ; resolvedRecords . clear ( ) ; Iterator < Record > records = batch . getRecords ( ) ; while ( records . hasNext ( ) ) { Record record = records . next ( ) ; String resolvedUrl = httpClientCommon . getResolvedUrl ( conf . resourceUrl , record ) ; WebTarget target = httpClientCommon . getClient ( ) . target ( resolvedUrl ) ; LOG . debug ( "Resolved HTTP Client URL: '{}'" , resolvedUrl ) ; // If the request (headers or body) contain a known sensitive EL and we're not using https then fail the request. if ( httpClientCommon . requestContainsSensitiveInfo ( conf . headers , conf . requestBody ) && ! target . getUri ( ) . getScheme ( ) . toLowerCase ( ) . startsWith ( "https" ) ) { throw new StageException ( Errors . HTTP_07 ) ; } // from HttpStreamConsumer final MultivaluedMap < String , Object > resolvedHeaders = httpClientCommon . resolveHeaders ( conf . headers , record ) ; String contentType = HttpStageUtil . getContentTypeWithDefault ( resolvedHeaders , conf . defaultRequestContentType ) ; final AsyncInvoker asyncInvoker = target . request ( ) . property ( OAuth1ClientSupport . OAUTH_PROPERTY_ACCESS_TOKEN , httpClientCommon . getAuthToken ( ) ) . headers ( resolvedHeaders ) . async ( ) ; HttpMethod method = httpClientCommon . getHttpMethod ( conf . httpMethod , conf . methodExpression , record ) ; rateLimiter . acquire ( ) ; if ( conf . requestBody != null && ! conf . requestBody . isEmpty ( ) && method != HttpMethod . GET ) { RecordEL . setRecordInContext ( bodyVars , record ) ; final String requestBody = bodyEval . eval ( bodyVars , conf . requestBody , String . class ) ; resolvedRecords . put ( record , new HeadersAndBody ( resolvedHeaders , requestBody , contentType , method , target ) ) ; responses . add ( asyncInvoker . method ( method . getLabel ( ) , Entity . entity ( requestBody , contentType ) ) ) ; } else { resolvedRecords . put ( record , new HeadersAndBody ( resolvedHeaders , null , null , method , target ) ) ; responses . add ( asyncInvoker . method ( method . getLabel ( ) ) ) ; } } records = batch . getRecords ( ) ; int recordNum = 0 ; while ( records . hasNext ( ) ) { try { Record record = processResponse ( records . next ( ) , responses . get ( recordNum ) , conf . maxRequestCompletionSecs , false ) ; if ( record != null ) { batchMaker . addRecord ( record ) ; } } catch ( OnRecordErrorException e ) { errorRecordHandler . onError ( e ) ; } finally { ++ recordNum ; } } if ( ! resolvedRecords . isEmpty ( ) ) { reprocessIfRequired ( batchMaker ) ; } }
{@inheritDoc}
{