CompletableFuture：组合任务、传播失败与管理超时

import java.util.*;
import java.util.concurrent.*;

public class Main {
    record Outcome<T>(T value, Throwable error) { boolean success() { return error==null; } }
    static Throwable unwrap(Throwable error) {
        while ((error instanceof CompletionException || error instanceof ExecutionException)
                && error.getCause()!=null) error=error.getCause();
        return error;
    }
    static <T> CompletableFuture<List<Outcome<T>>> collect(List<CompletableFuture<T>> input) {
        List<CompletableFuture<T>> snapshot=List.copyOf(input); // 固定输入次序并拒绝null阶段。
        List<CompletableFuture<Outcome<T>>> outcomes=new ArrayList<>();
        for (CompletableFuture<T> future:snapshot)
            outcomes.add(future.handle((value,error) -> new Outcome<>(value,unwrap(error))));
            // 正常null值合法；失败保留为显式结果，避免丢失异常。
        CompletableFuture<?>[] all=outcomes.toArray(new CompletableFuture<?>[0]);
        return CompletableFuture.allOf(all).thenApply(ignored -> {
            List<Outcome<T>> result=new ArrayList<>(outcomes.size());
            for (CompletableFuture<Outcome<T>> item:outcomes) result.add(item.join());
            // allOf已确认每个转换阶段完成，按输入顺序读取就绪结果。
            return List.copyOf(result);
        });
    }
    static void check(boolean ok) { if (!ok) throw new AssertionError(); }
    public static void main(String[] args) throws Exception {
        CompletableFuture<Integer> a=new CompletableFuture<>(), b=new CompletableFuture<>();
        CompletableFuture<List<Outcome<Integer>>> batch=collect(List.of(a,b));
        b.complete(20); check(!batch.isDone()); a.complete(10);
        List<Outcome<Integer>> ordered=batch.get(5,TimeUnit.SECONDS);
        check(ordered.get(0).value()==10 && ordered.get(1).value()==20);
        IllegalStateException error=new IllegalStateException("failed");
        Outcome<Integer> failed=collect(List.of(CompletableFuture.<Integer>failedFuture(error))).join().get(0);
        check(!failed.success() && failed.error()==error);
        check(collect(List.of(CompletableFuture.<Integer>completedFuture(null))).join().get(0).success());
        check(collect(List.<CompletableFuture<Integer>>of()).join().isEmpty());
        CompletableFuture<Integer> cancelled=new CompletableFuture<>(); cancelled.cancel(true);
        check(collect(List.of(cancelled)).join().get(0).error() instanceof CancellationException);
        check(CompletableFuture.completedFuture(3).thenCompose(x -> CompletableFuture.completedFuture(x*2)).join()==6);
        System.out.println("PASS 7");
    }
}