How to flatten a list inside a stream of completable futures?

user_196032 0
Viewed 703

I have this:

Stream<CompletableFuture<List<Item>>>

how can I convert it to

Stream<CompletableFuture<Item>>

Where: the second stream is comprised of each and all the Items inside each of the lists in the first stream.

I looked into thenCompose but that solves a completely different problem which is also referred to as "flattening".

How can this be done efficiently, in a streaming fashion, without blocking or prematurely consuming more stream items than necessary?

Here is my best attempt so far:

    ExecutorService pool = Executors.newFixedThreadPool(PARALLELISM);
    Stream<CompletableFuture<List<IncomingItem>>> reload = ... ;

    @SuppressWarnings("unchecked")
    CompletableFuture<List<IncomingItem>> allFutures[] = reload.toArray(CompletableFuture[]::new);
    CompletionService<List<IncomingItem>> queue = new ExecutorCompletionService<>(pool);
    for(CompletableFuture<List<IncomingItem>> item: allFutures) {
        queue.submit(item::get);
    }
    List<IncomingItem> THE_END = new ArrayList<IncomingItem>();
    CompletableFuture<List<IncomingItem>> ender = CompletableFuture.allOf(allFutures).thenApply(whatever -> {
        queue.submit(() -> THE_END);
        return THE_END;
    });
    queue.submit(() -> ender.get());
    Iterable<List<IncomingItem>> iter = () -> new Iterator<List<IncomingItem>>() {
        boolean checkNext = true;
        List<IncomingItem> next = null;
        @Override
        public boolean hasNext() {
            if(checkNext) {
                try {
                    next = queue.take().get();
                } catch (InterruptedException | ExecutionException e) {
                    throw new RuntimeException(e);
                }
                checkNext = false;
            }
            if(next == THE_END || next == null) {
                return false;
            }
            else {
                return true;
            }
        }
        @Override
        public List<IncomingItem> next() {
            if(checkNext) {
                hasNext();
            }
            if(!hasNext()) {
                throw new IllegalStateException();
            }
            checkNext = true;
            return next;
        }
    };
    Stream<IncomingItem> flat = StreamSupport.stream(iter.spliterator(), false).flatMap(List::stream);

This works at first, unfortunately, it has a fatal bug: the resulting stream seems to terminate prematurely, before retrieving all the items.

java-stream java-8 java
2 Answers

As I wrote in my comment, this is impossible.

Consider a some arbitrary service, which will return a CompletableFuture<Integer>:

CompletableFuture<Integer> getDiceRoll();

I can now convert this CompletableFuture<Integer> to a Stream<CompletableFuture<List<Object>>> without any problem:

Stream<CompletableFuture<List<Object>>> futureList = Stream.of(getDiceRoll().thenApply(n -> List.of(new Object[n])));

Let's suppose there would be a general way to turn a Stream<CompletableFuture<List<T>>> into a Stream<CompletableFuture<T>>:

<T> Stream<CompletableFuture<T> magic(Stream<CompletableFuture<List<T>>> arg);

Then I can do the following:

int diceRoll = magic(Stream.of(getDiceRoll().thenApply(n -> List.of(new Object[n])))).count();

Wait, what?
I am now able to get an arbitrary integer out of a CompletableFuture. Which means, with some engineering effort I can get all the information out of a CompletableFuture - after all, memory is just some numbers.

So we have to conclude that a method like magic can not exist, without violating the time fabric.
And this is the answer: There is no such method, because it can not exist.

user_8454140

Agreed with Johannes Kuhn. You can't know Futures's state while it's still executing and thus can not convert from Stream<CompletableFuture<List>> to Stream<CompletableFuture> .

Although the output of stream can be merged using following piece of code - java Stream<CompletableFuture<List<Item>>> to java List<Item> or

java List<CompletableFuture<List<AuditRecord>>> to java List<Item> 

List<Item> output = input.map(CompletableFuture::join).collect(toList()).stream()
    .flatMap(Collection::stream).collect(toList());
user_159738110
Related