omniture-lib/src/main/java/com/streamsets/pipeline/stage/origin/omniture/OmniturePollingConsumer.java
public void getReport(int reportId)
      throws InterruptedException, ExecutionException, TimeoutException, IOException, StageException {
    int waitTime = 1000;
    Response response = null;
    while (!stop) {
      final AsyncInvoker asyncInvoker = getResource.request()
          .header(WSSE_HEADER, OmnitureAuthUtil.getHeader(username.get(), sharedSecret.get()))
          .async();

      LOG.debug("Getting report using URL {} with report ID {}", getResource.getUri().toURL().toString(), reportId);
      final Future<Response> responseFuture = asyncInvoker.post(Entity.json("{ \"reportID\": " + reportId + " }"));
      response = responseFuture.get(responseTimeoutMillis, TimeUnit.MILLISECONDS);
      String input = response.readEntity(String.class);

      ObjectMapper mapper = new ObjectMapper();
      JsonNode root = mapper.readTree(input);

      // If the report has an error field, it means the report has not finished generating
      if (!root.has("error")) {
        boolean accepted = entityQueue.offer(input, responseTimeoutMillis, TimeUnit.MILLISECONDS);
        if (!accepted) {
          LOG.warn("Response buffer full, dropped record.");
        }
        break;
      } else {
        // Exponential backoff while making subsequent Report.Get requests
        if (root.get("error").textValue().equals("report_not_ready")) {
          waitTime *= 2;
          LOG.info("Report not available. Sleeping for {} seconds", waitTime / 1000);
          Thread.sleep(waitTime);
        } else {
          throw new StageException(Errors.OMNITURE_02,
              root.get("error").get("error_description").asText());
        }
      }
    }
    response.close();
  }
public void getReport(int reportId)
      throws InterruptedException, ExecutionException, TimeoutException, IOException, StageException {
    int waitTime = 1000;
    Response response = null;
    while (!stop) {
      final AsyncInvoker asyncInvoker = getResource.request()
          .header(WSSE_HEADER, OmnitureAuthUtil.getHeader(username.get(), sharedSecret.get()))
          .async();

      LOG.debug("Getting report using URL {} with report ID {}", getResource.getUri().toURL().toString(), reportId);
      final Future<Response> responseFuture = asyncInvoker.post(Entity.json("{ \"reportID\": " + reportId + " }"));
      response = responseFuture.get(responseTimeoutMillis, TimeUnit.MILLISECONDS);
      String input = response.readEntity(String.class);

      ObjectMapper mapper = new ObjectMapper();
      JsonNode root = mapper.readTree(input);

      // If the report has an error field, it means the report has not finished generating
      if (!root.has("error")) {
        boolean accepted = entityQueue.offer(input, responseTimeoutMillis, TimeUnit.MILLISECONDS);
        if (!accepted) {
          LOG.warn("Response buffer full, dropped record.");
        }
        break;
      } else {
        // Exponential backoff while making subsequent Report.Get requests
        if (root.get("error").textValue().equals("report_not_ready")) {
          waitTime *= 2;
          LOG.info("Report not available. Sleeping for {} seconds", waitTime / 1000);
          Thread.sleep(waitTime);
        } else {
          throw new StageException(Errors.OMNITURE_02,
              root.get("error").get("error_description").asText());
        }
      }
    }
    response.close();
  }
public void getReport ( int reportId ) throws InterruptedException , ExecutionException , TimeoutException , IOException , StageException { int waitTime = 1000 ; Response response = null ; while ( ! stop ) { final AsyncInvoker asyncInvoker = getResource . request ( ) . header ( WSSE_HEADER , OmnitureAuthUtil . getHeader ( username . get ( ) , sharedSecret . get ( ) ) ) . async ( ) ; LOG . debug ( "Getting report using URL {} with report ID {}" , getResource . getUri ( ) . toURL ( ) . toString ( ) , reportId ) ; final Future < Response > responseFuture = asyncInvoker . post ( Entity . json ( "{ \"reportID\": " + reportId + " }" ) ) ; response = responseFuture . get ( responseTimeoutMillis , TimeUnit . MILLISECONDS ) ; String input = response . readEntity ( String . class ) ; ObjectMapper mapper = new ObjectMapper ( ) ; JsonNode root = mapper . readTree ( input ) ; // If the report has an error field, it means the report has not finished generating if ( ! root . has ( "error" ) ) { boolean accepted = entityQueue . offer ( input , responseTimeoutMillis , TimeUnit . MILLISECONDS ) ; if ( ! accepted ) { LOG . warn ( "Response buffer full, dropped record." ) ; } break ; } else { // Exponential backoff while making subsequent Report.Get requests if ( root . get ( "error" ) . textValue ( ) . equals ( "report_not_ready" ) ) { waitTime *= 2 ; LOG . info ( "Report not available. Sleeping for {} seconds" , waitTime / 1000 ) ; Thread . sleep ( waitTime ) ; } else { throw new StageException ( Errors . OMNITURE_02 , root . get ( "error" ) . get ( "error_description" ) . asText ( ) ) ; } } } response . close ( ) ; }
Posts a request to the Omniture API to get a report back. Reports may take a while to generate,
so this will loop on the Report.Get request, and ignore any errors indicating that the report
is not yet ready.
@param reportId ID of report to get
@throws InterruptedException
@throws ExecutionException
@throws TimeoutException
@throws IOException
Posts a request to the Omniture API to get a report back . Reports may take a while to generate so this will loop on the Report . Get request and ignore any errors indicating that the report is not yet ready .