Consuming Rest Service the functional way

Question

So I was curious as to how would the following code be rewritten more effiently with the java8 streams API.

public static List<FlightInfo> getResults(String origin,List<String> destinations)  {

    final String uri = "https://api.searchflight.com/;
    List<FlightInfo> results = new LinkedList<FlightInfo>();

    for(String destination:destinations) {


            RestTemplate restTemplate = new RestTemplate();

            String params = getParams(origin,destination);
            FlightInfo result = restTemplate.postForObject(uri+params,FlightInfo.class);

            results.add(result);
    }

    return results;

}

After this method is done doing what its is doing and I receive the list of FLightInfo objects, I am converting it into a stream and will be doing various transformations on it (group by, etc). Now it is pretty apparant that this is a long running operation. Furthermore it actually combines multiple rest calls to the web service, so I already have most of the data obtained by the time I make the last call, but I would not start processing it before the whole method returns.

Is there a way to do all that a bit more reactively? Could I return a stream immediatelly and have the operations on that stream process data as it comes down the pipe or this is a bit too much to ask? How would that be done in Java 8. That

Answer 1

Well it all depends on when you need the outcome. If you wish it to be sequential, this below is still a decent way as its lazy. But it would boil at a terminal operation (say during collect).

public static Stream<FlightInfo> getResults(String origin,List<String> destinations)  {
    final String uri = "https://api.searchflight.com/";
    return destinations.stream().map(destination -> {
        RestTemplate restTemplate = new RestTemplate();
        String params = getParams(origin,destination);
        FlightInfo result = restTemplate.postForObject(uri+params,FlightInfo.class);
        return result;
    })    
}

Or I would do it with destinations.stream().parallel() if I can. This is a reasonable outcome in most circumstances. But still it will not start processing it in parallel till the time you call a terminal operation for it. Which absolutely makes sense.

But it looks to me that you wish for a producer-consumer type of a thing. For which:

public static CompletableFuture<List<FlightInfo>> getResults(String origin,List<String> destinations)  {
    final String uri = "https://api.searchflight.com/";
    List<CompletableFuture<FlightInfo>> collect = destinations
           .stream()
           .map(destination -> CompletableFuture.supplyAsync(() -> {
                 RestTemplate restTemplate = new RestTemplate();
                 String params = getParams(origin,destination);
                 FlightInfo result = restTemplate.postForObject(uri+params,FlightInfo.class);
                 return result;            
           })).collect(Collectors.toList());
    return sequence(collect);       //line-1 
}

public static <T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> com) {
    return CompletableFuture.allOf(com.toArray(new CompletableFuture[com.size()]))
            .thenApply(v -> com.stream()
                            .map(CompletableFuture::join)
                            .collect(Collectors.toList())
            );
}

For simplicity, at line-1 you can simply return collect instead of sequence(collect). You can then iterate over the list to fetch each value.

But with sequence, you have a single CompletableFuture object to worry about, which you can then check for values at once if completed.