How to execute an Array of CompletableFuture and compose their results

Viewed 1599

I am investigating Java 8 CompletableFutures and read (and seen) that I should employ thenCompose instead of thenApply.

I have converted my code to use thenCompose but I have a feeling in an incorrect manner.

Here is my controlling code...

final CompletableFuture<List<String>> extractor = get(htmlPageSource);
@SuppressWarnings("unchecked")
final CompletableFuture<List<Documentable>>[] completableFutures =
   new CompletableFuture[ENDPOINT.EXTRACTABLES.size()];
int index = 0;
for( ENDPOINT endpoint : ENDPOINT.EXTRACTABLES ) {
   final CompletableFuture<List<Documentable>> metaData =
      extractor.thenComposeAsync(
         s -> endpoint.contactEndpoit(s), executorService );
   completableFutures[index++] = metaData.exceptionally(x -> failedList(x));
}
CompletableFuture
   .allOf( completableFutures )
   .thenComposeAsync( dummy -> combineDocuments( completableFutures ))
   .thenAccept   ( x -> finish( x ))
   .exceptionally( x -> failed( x ));

private List<Documentable> failedList(final Throwable x) {
   LOGGER.error("failedList", x);
   final List<Documentable> metaData = new ArrayList<>();
   return metaData;
}

private Void failed(final Throwable x) {
   LOGGER.error("failed", x);
   return null;
}

Which I believe is acceptable

However the code that makes me uneasy is this:-

WWW_SITE_ONE("https://example.site.one/") {
   @Override
   public <T extends Documentable> CompletionStage<List<T>> contactEndpoit( final List<String> elements) {
      LOGGER.info("WWW_SITE_ONE " + Thread.currentThread().getName());
      final List<T> SITE_ONEs = new ArrayList<>();
      for (final String element : elements) {
         try {
            final String json = Jsoup.connect(ENDPOINT.WWW_SITE_ONE.getBaseUrl() + element).ignoreContentType(true).ignoreHttpErrors(true).maxBodySize(0).timeout(60000).execute().body();
            if (json.contains("errors")) {
               continue;
            }
            final T SITE_ONE = OBJECT_READER_SITE_ONE.readValue(json);
            SITE_ONEs.add(SITE_ONE);
         }
         catch( final Throwable e ) {
            LOGGER.error("WWW_SITE_ONE failed", e);
            throw new RuntimeException(e);
         }
      }
      return CompletableFuture.supplyAsync(() -> SITE_ONEs);  
   }
},
WWW_SITE_TWO("https://example.site.two/") {
   @Override
   public <T extends Documentable> CompletionStage<List<T>> contactEndpoit(final List<String> elements) {
      LOGGER.info("WWW_SITE_TWO " + Thread.currentThread().getName());      
      final List<T> SITE_TWOs = new ArrayList<>();
      for (final String element : elements) {
         try {
            final String json = Jsoup.connect(ENDPOINT.WWW_SITE_TWO.getBaseUrl() + element).ignoreContentType(true).ignoreHttpErrors(true).maxBodySize(0).timeout(60000).execute().body();
            if (json.equals("Resource not found.")) {
               continue;
            }
            final T SITE_TWO = OBJECT_READER_SITE_TWO.readValue(json);
            SITE_TWOs.add(SITE_TWO);
         }
         catch (final Throwable e) {
            LOGGER.error("WWW_SITE_TWO failed", e);
            throw new RuntimeException(e);
         }
      }
      return CompletableFuture.supplyAsync(() -> SITE_TWOs);  
   }
},
WWW_SITE_THREE("https://example.site.three/") {
   @Override
   public <T extends Documentable> CompletionStage<List<T>> contactEndpoit(final List<String> elements) {
      LOGGER.info("WWW_SITE_THREE " + Thread.currentThread().getName());        
      final List<T> SITE_THREEs = new ArrayList<>();
      for (final String element : elements) {
         try {
            final String SITE_THREEJsonString = Jsoup
               .connect( ENDPOINT.WWW_SITE_THREE.getBaseUrl() + element)
               .ignoreContentType(true)
               .ignoreHttpErrors(true)
               .maxBodySize(0)
               .timeout(60000)
               .execute()
               .body();
            final SITE_THREE SITE_THREE_Json = OBJECT_READER_SITE_THREE.readValue(SITE_THREEJsonString);
            final T SITE_THREE = (T) SITE_THREE_Json;
            if (SITE_THREE_Json.getHitCount() > 0) {
               SITE_THREEs.add(SITE_THREE);
            }
         }
         catch (final Throwable e) {
            LOGGER.error("WWW_SITE_THREE failed", e);
            throw new RuntimeException(e);
         }
      }
      return CompletableFuture.supplyAsync(() -> SITE_THREEs);  
   }
};

Its where I am returning CompletableFuture.supplyAsync(() -> SITE_THREEs);

Is this the correct approach?

Or does this start another asynchronous thread to simply return my List<>?

1 Answers
Related