ScatterGathererUnirestFuture.java

package parallel;

import com.mashape.unirest.http.HttpResponse;
import com.mashape.unirest.http.Unirest;
import com.mashape.unirest.http.async.Callback;
import com.mashape.unirest.http.exceptions.UnirestException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;

public class ScatterGathererUnirestFuture {

  public static void main(String[] args) {
    ScatterGathererUnirestFuture scatterGathererUnirestFuture = new ScatterGathererUnirestFuture();
    scatterGathererUnirestFuture.doStuff();
  }

  public void doStuff() {
    List<String> websites = new ArrayList<>();

    websites.add("https://www.google.com");
    websites.add("https://www.facebook.com");
    websites.add("https://login.cwatch.comodo.com");
    websites.add("http://xyztts.logo");

    List<Future<HttpResponse<String>>> reqFutureList = new ArrayList<>();
    final CountDownLatch responseWaiter = new CountDownLatch(websites.size());
    final LinkedBlockingQueue<HttpResponse<String>> linkedBlockingQueue = new LinkedBlockingQueue<>();

    for (String webSite : websites) {
      Future<HttpResponse<String>> response = Unirest.get(webSite)
          .asStringAsync(new AsyncHttpCallback<>(webSite, responseWaiter, linkedBlockingQueue));

      reqFutureList.add(response);
    }

    try {
      responseWaiter.await();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.err.println("Finished " + System.currentTimeMillis());

    ArrayList<HttpResponse<String>> responses = new ArrayList<HttpResponse<String>>(linkedBlockingQueue);

    for(HttpResponse<String> response : responses){
      System.err.println(response.getHeaders());
    }

  }

  private class AsyncHttpCallback<T> implements Callback<T> {

    private final CountDownLatch responseWaiter;
    private final String website;
    private final LinkedBlockingQueue<HttpResponse<T>> linkedBlockingQueue;

    AsyncHttpCallback(String website, CountDownLatch responseWaiter,
        LinkedBlockingQueue<HttpResponse<T>> linkedBlockingQueue) {
      this.responseWaiter = responseWaiter;
      this.website = website;
      this.linkedBlockingQueue = linkedBlockingQueue;
    }

    @Override
    public void completed(HttpResponse<T> httpResponse) {
      System.err.println(
          website + " " + Thread.currentThread() + " completed " + System.currentTimeMillis());
      try {
        linkedBlockingQueue.put(httpResponse);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }finally {
        responseWaiter.countDown();
      }
    }

    @Override
    public void failed(UnirestException e) {
      System.err.println(website + " " +"failed");
      responseWaiter.countDown();
    }

    @Override
    public void cancelled() {
      System.err.println(website + " " +"CANCELLED!");
      responseWaiter.countDown();
    }
  }

}