// El problema, en código real de BiblioTech.
ExecutorService ejecutor = Executors.newFixedThreadPool(4);

Future<Catalogo> f1 = ejecutor.submit(() -> cargarCatalogo());
Catalogo c = f1.get();                    // BLOQUEA. El hilo se para aqui.

Future<Double> f2 = ejecutor.submit(() -> calcularMultas(c));
double total = f2.get();                  // BLOQUEA otra vez.

Future<Path> f3 = ejecutor.submit(() -> exportar(total));
Path informe = f3.get();                  // Y otra vez.

// Lo mismo con CompletableFuture: se DECLARA la cadena y se retorna
// inmediatamente. Ningun hilo se bloquea en ningun momento.
CompletableFuture<Path> informe =
        CompletableFuture.supplyAsync(() -> cargarCatalogo(), poolEs)
                         .thenApplyAsync(c -> calcularMultas(c), poolCalculo)
                         .thenApplyAsync(total -> exportar(total), poolEs)
                         .exceptionally(error -> rutaDeError(error));

System.out.println("[main] cadena lanzada; el menu sigue vivo");
mostrarMenu();      // el hilo principal NO ha esperado a nada

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;

// 1. supplyAsync: ejecuta un Supplier y DEVUELVE un valor.
//    Es el punto de entrada mas frecuente.
CompletableFuture<Catalogo> f1 =
        CompletableFuture.supplyAsync(() -> cargarCatalogo());

// Con ejecutor propio (recomendado, apartado 4):
CompletableFuture<Catalogo> f2 =
        CompletableFuture.supplyAsync(() -> cargarCatalogo(), poolEs);

// 2. runAsync: ejecuta un Runnable, NO devuelve valor.
//    El tipo es CompletableFuture<Void>.
CompletableFuture<Void> f3 =
        CompletableFuture.runAsync(() -> registrarAuditoria("arranque"), poolEs);

// 3. completedFuture: ya completado con un valor. No ejecuta nada.
//    Util para valores en cache y para pruebas.
CompletableFuture<Catalogo> f4 =
        CompletableFuture.completedFuture(catalogoEnCache);

// 4. failedFuture (Java 9): ya completado con una excepcion.
CompletableFuture<Catalogo> f5 =
        CompletableFuture.failedFuture(new BiblioTechException("catalogo no disponible"));

// 5. Vacio, para completarlo manualmente despues (apartado 13).
CompletableFuture<Catalogo> f6 = new CompletableFuture<>();

/**
 * Devuelve SIEMPRE un CompletableFuture, tanto si el dato esta en
 * cache (respuesta inmediata) como si hay que calcularlo (asincrono).
 * El llamante no tiene que distinguir los dos casos.
 */
public CompletableFuture<Ficha> obtenerFicha(String isbn) {
    Ficha enCache = cache.get(isbn);
    if (enCache != null) {
        return CompletableFuture.completedFuture(enCache);   // ya lista
    }
    return CompletableFuture.supplyAsync(() -> construirFicha(isbn), poolEs);
}

// PELIGROSO: E/S en el pool comun.
// El pool comun tiene (nucleos - 1) hilos y lo COMPARTE toda la JVM:
// los streams paralelos (10-04), otras librerias y el resto de tu
// codigo. Con 7 hilos y 10 lecturas de fichero lentas, el pool queda
// inservible para todos.
CompletableFuture.supplyAsync(() -> Files.readAllLines(rutaEnorme));

// CORRECTO: ejecutor propio para E/S, dimensionado segun 08-05.
private static final ExecutorService POOL_ES = new ThreadPoolExecutor(
        16, 16, 0L, TimeUnit.MILLISECONDS,
        new ArrayBlockingQueue<>(200),
        r -> new Thread(r, "bibliotech-async-es-" + contador.getAndIncrement()),
        new ThreadPoolExecutor.CallerRunsPolicy());

CompletableFuture.supplyAsync(() -> Files.readAllLines(rutaEnorme), POOL_ES);

import java.util.concurrent.CompletableFuture;

CompletableFuture<Void> cadena =
    CompletableFuture
        // 1. Produce un Catalogo
        .supplyAsync(() -> cargarCatalogo(), POOL_ES)

        // 2. thenApply: Catalogo -> Double. Transforma.
        .thenApply(catalogo -> calcularMultasTotales(catalogo))

        // 3. thenApply: Double -> String. Otra transformacion.
        .thenApply(total -> String.format("Multas pendientes: %.2f EUR", total))

        // 4. thenAccept: consume el String, no produce nada.
        .thenAccept(texto -> System.out.println(texto))

        // 5. thenRun: no recibe ni devuelve. Para efectos finales.
        .thenRun(() -> LOG.info("Informe de multas completado"));

System.out.println("[main] cadena declarada, sigo trabajando");

// Metodo que devuelve un valor normal:
Double calcularMultas(Catalogo c) { ... }

// Metodo que devuelve un CompletableFuture (porque es asincrono):
CompletableFuture<Double> calcularMultasAsync(Catalogo c) { ... }

// EL ERROR: thenApply con una funcion que devuelve CompletableFuture.
CompletableFuture<CompletableFuture<Double>> anidado =
        CompletableFuture.supplyAsync(() -> cargarCatalogo())
                         .thenApply(c -> calcularMultasAsync(c));

// Ahora, para llegar al Double, hay que desenvolver DOS veces:
Double d = anidado.get().get();      // horrible, y bloquea dos veces

// LA SOLUCION: thenCompose APLANA el anidamiento.
CompletableFuture<Double> plano =
        CompletableFuture.supplyAsync(() -> cargarCatalogo())
                         .thenCompose(c -> calcularMultasAsync(c));

Double d2 = plano.join();            // un solo nivel

package com.nexussoftware.bibliotech.servicio;

import java.util.concurrent.CompletableFuture;

public class ConsultaAsincrona {

    /** Sincrono: devuelve el valor directamente. */
    private Material buscarMaterial(String isbn) { ... }

    /** Asincrono: devuelve un CompletableFuture. */
    private CompletableFuture<Ficha> construirFichaAsync(Material m) { ... }

    /** Sincrono: transformacion barata. */
    private String formatear(Ficha f) { ... }

    /**
     * Cadena que alterna operaciones sincronas y asincronas.
     * Fijate en que thenCompose aparece exactamente donde la
     * funcion devuelve un CompletableFuture.
     */
    public CompletableFuture<String> fichaFormateada(String isbn) {
        return CompletableFuture
                .supplyAsync(() -> buscarMaterial(isbn), POOL_ES)  // -> Material
                .thenCompose(this::construirFichaAsync)            // -> Ficha (async)
                .thenApply(this::formatear);                       // -> String (sync)
    }
}

.thenApply(f)                  // 1. sin sufijo
.thenApplyAsync(f)             // 2. con sufijo, ejecutor por defecto
.thenApplyAsync(f, ejecutor)   // 3. con sufijo y ejecutor explicito

public class QueHiloEjecuta {

    public static void main(String[] args) throws Exception {

        ExecutorService pool = Executors.newFixedThreadPool(2,
                r -> new Thread(r, "bibliotech-pool"));

        System.out.println("main ejecuta en: " + Thread.currentThread().getName());

        CompletableFuture<String> f = CompletableFuture
                .supplyAsync(() -> {
                    traza("supplyAsync");
                    return "catalogo";
                }, pool)
                .thenApply(v -> {
                    traza("thenApply");          // el hilo del pool
                    return v + "-procesado";
                })
                .thenApplyAsync(v -> {
                    traza("thenApplyAsync");     // commonPool
                    return v + "-async";
                })
                .thenApplyAsync(v -> {
                    traza("thenApplyAsync(pool)"); // el pool que le damos
                    return v + "-propio";
                }, pool);

        System.out.println("resultado: " + f.join());
        pool.shutdown();
    }

    static void traza(String etapa) {
        System.out.printf("  %-24s -> %s%n", etapa, Thread.currentThread().getName());
    }
}

main ejecuta en: main
  supplyAsync              -> bibliotech-pool
  thenApply                -> bibliotech-pool
  thenApplyAsync           -> ForkJoinPool.commonPool-worker-1
  thenApplyAsync(pool)     -> bibliotech-pool
resultado: catalogo-procesado-async-propio

// MAL: bloquear en una etapa sin sufijo ocupa el hilo anterior,
// que podria ser un hilo del pool comun o el propio main.
.thenApply(catalogo -> escribirEnDisco(catalogo))     // 500 ms bloqueando

// BIEN: el trabajo pesado va a un pool dedicado.
.thenApplyAsync(catalogo -> escribirEnDisco(catalogo), POOL_ES)

package com.nexussoftware.bibliotech.servicio;

import java.util.concurrent.CompletableFuture;

public class ResumenAsincrono {

    /**
     * Las dos consultas son independientes y se lanzan a la vez.
     * El tiempo total es el del MAS LENTO, no la suma.
     */
    public CompletableFuture<String> resumenCompleto(String isbn) {

        CompletableFuture<Material> material =
                CompletableFuture.supplyAsync(() -> catalogo.buscar(isbn), POOL_ES);

        CompletableFuture<Integer> prestamos =
                CompletableFuture.supplyAsync(() -> registro.contarPrestamos(isbn), POOL_ES);

        // thenCombine espera a AMBOS y aplica la BiFunction (04-06).
        return material.thenCombine(prestamos, (m, n) ->
                String.format("%s (%s) - %d prestamos historicos",
                        m.titulo(), m.isbn(), n));
    }

    /** Tres o mas: se encadenan los thenCombine. */
    public CompletableFuture<InformeCompleto> informeCompleto(String isbn) {

        CompletableFuture<Material> material =
                CompletableFuture.supplyAsync(() -> catalogo.buscar(isbn), POOL_ES);
        CompletableFuture<Integer> prestamos =
                CompletableFuture.supplyAsync(() -> registro.contarPrestamos(isbn), POOL_ES);
        CompletableFuture<Double> multas =
                CompletableFuture.supplyAsync(() -> calculadora.multasDe(isbn), POOL_ES);

        return material
                .thenCombine(prestamos, ParcialMaterialPrestamos::new)
                .thenCombine(multas, (parcial, m) ->
                        new InformeCompleto(parcial.material(), parcial.prestamos(), m));
    }

    private record ParcialMaterialPrestamos(Material material, int prestamos) { }
}

package com.nexussoftware.bibliotech.servicio;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;

public class AvisosAsincronos {

    /**
     * Envia todos los avisos en paralelo y espera a que terminen TODOS.
     *
     * El patron de recogida es siempre el mismo:
     *   1. Crear la lista de futuros.
     *   2. allOf(...) sobre el array.
     *   3. thenApply que recorre los futuros y llama a join()
     *      —seguro, porque allOf garantiza que ya estan completos—.
     */
    public CompletableFuture<List<ResultadoAviso>> enviarTodos(List<Prestamo> vencidos) {

        List<CompletableFuture<ResultadoAviso>> futuros = new ArrayList<>();
        for (Prestamo p : vencidos) {
            futuros.add(CompletableFuture.supplyAsync(() -> enviarAviso(p), POOL_ES));
        }

        // allOf recibe un array, no una lista.
        CompletableFuture<Void> todos =
                CompletableFuture.allOf(futuros.toArray(new CompletableFuture[0]));

        return todos.thenApply(v -> {
            List<ResultadoAviso> resultados = new ArrayList<>(futuros.size());
            for (CompletableFuture<ResultadoAviso> f : futuros) {
                // join() aqui NO bloquea: allOf ya garantizo que
                // todos estan completos.
                resultados.add(f.join());
            }
            return resultados;
        });
    }

    /**
     * Variante TOLERANTE A FALLOS: un aviso fallido no tumba el conjunto.
     *
     * OJO: si un futuro falla, el allOf tambien falla y su join()
     * relanzaria la excepcion. La solucion es blindar CADA futuro con
     * su propio exceptionally ANTES de agregarlos.
     */
    public CompletableFuture<List<ResultadoAviso>> enviarTodosTolerante(
            List<Prestamo> vencidos) {

        List<CompletableFuture<ResultadoAviso>> futuros = new ArrayList<>();
        for (Prestamo p : vencidos) {
            futuros.add(CompletableFuture
                    .supplyAsync(() -> enviarAviso(p), POOL_ES)
                    .exceptionally(e -> ResultadoAviso.fallo(p, e.getMessage())));
        }

        return CompletableFuture
                .allOf(futuros.toArray(new CompletableFuture[0]))
                .thenApply(v -> {
                    List<ResultadoAviso> r = new ArrayList<>();
                    for (CompletableFuture<ResultadoAviso> f : futuros) r.add(f.join());
                    return r;
                });
    }
}

// Tres fuentes para el mismo catalogo; nos vale la primera.
CompletableFuture<Catalogo> cache    = leerDeCache();
CompletableFuture<Catalogo> fichero  = leerDeFicheroPrincipal();
CompletableFuture<Catalogo> respaldo = leerDeCopiaSeguridad();

CompletableFuture<Object> primero =
        CompletableFuture.anyOf(cache, fichero, respaldo);

Catalogo c = (Catalogo) primero.join();     // hay que castear

package com.nexussoftware.bibliotech.servicio;

import java.util.concurrent.CompletableFuture;
import java.util.logging.Level;
import java.util.logging.Logger;

public class ManejoDeErroresAsincrono {

    private static final Logger LOG =
            Logger.getLogger(ManejoDeErroresAsincrono.class.getName());

    /** 1. exceptionally: valor de respaldo si algo falla. */
    public CompletableFuture<Catalogo> cargarConRespaldo() {
        return CompletableFuture
                .supplyAsync(() -> cargarDesdeFichero(), POOL_ES)
                .exceptionally(error -> {
                    // 'error' es una CompletionException que ENVUELVE la causa.
                    LOG.log(Level.WARNING, "Fallo al cargar; uso el respaldo",
                            error.getCause());
                    return Catalogo.vacio();      // la cadena CONTINUA
                });
    }

    /** 2. handle: se ejecuta siempre, y unifica los dos caminos. */
    public CompletableFuture<String> informeConHandle() {
        return CompletableFuture
                .supplyAsync(() -> generarInforme(), POOL_ES)
                .handle((resultado, error) -> {
                    // Exactamente uno de los dos es null.
                    if (error != null) {
                        LOG.log(Level.SEVERE, "Informe fallido", error);
                        return "INFORME NO DISPONIBLE: " + causaDe(error).getMessage();
                    }
                    return "INFORME OK: " + resultado;
                });
    }

    /** 3. whenComplete: observa sin alterar. Ideal para trazas y metricas. */
    public CompletableFuture<Catalogo> cargarConTraza() {
        long inicio = System.nanoTime();
        return CompletableFuture
                .supplyAsync(() -> cargarDesdeFichero(), POOL_ES)
                .whenComplete((catalogo, error) -> {
                    long ms = (System.nanoTime() - inicio) / 1_000_000;
                    if (error != null) {
                        LOG.log(Level.WARNING, "Carga fallida en " + ms + " ms", error);
                    } else {
                        LOG.log(Level.INFO, "Catalogo cargado en {0} ms ({1} materiales)",
                                new Object[] { ms, catalogo.tamano() });
                    }
                    // Devolver algo aqui NO cambiaria el resultado:
                    // whenComplete solo OBSERVA. El error sigue propagandose.
                });
    }

    /** Utilidad: desenvolver la causa real. */
    static Throwable causaDe(Throwable t) {
        return (t instanceof java.util.concurrent.CompletionException
                || t instanceof java.util.concurrent.ExecutionException)
                && t.getCause() != null ? t.getCause() : t;
    }
}

CompletableFuture<Path> informe = CompletableFuture
        .supplyAsync(() -> cargarCatalogo(), POOL_ES)
        .thenApplyAsync(this::calcularMultas, POOL_CALCULO)
        .thenApplyAsync(this::exportar, POOL_ES)
        .whenComplete((ruta, error) -> registrarMetrica(ruta, error))  // observar
        .exceptionally(error -> {                                       // manejar
            LOG.log(Level.SEVERE, "Cadena de informe fallida", error);
            return Path.of("informes/error.txt");
        });

public class PropagacionDeErrores {

    public static void main(String[] args) {

        CompletableFuture<String> cadena = CompletableFuture
                .supplyAsync(() -> {
                    System.out.println("  etapa 1: OK");
                    return "catalogo";
                })
                .thenApply(v -> {
                    System.out.println("  etapa 2: lanzando excepcion");
                    throw new IllegalStateException("catalogo corrupto");
                })
                .thenApply(v -> {
                    System.out.println("  etapa 3: NO SE EJECUTA");
                    return v + "-procesado";
                })
                .thenApply(v -> {
                    System.out.println("  etapa 4: TAMPOCO");
                    return v.toUpperCase();
                })
                .exceptionally(error -> {
                    System.out.println("  manejador: capturado " + error.getClass().getSimpleName());
                    System.out.println("             causa real: "
                            + error.getCause().getClass().getSimpleName()
                            + ": " + error.getCause().getMessage());
                    return "VALOR-DE-RESPALDO";
                })
                .thenApply(v -> {
                    System.out.println("  etapa 5: SI se ejecuta, con el respaldo");
                    return v.toLowerCase();
                });

        System.out.println("resultado: " + cadena.join());
    }
}

  etapa 1: OK
  etapa 2: lanzando excepcion
  manejador: capturado CompletionException
             causa real: IllegalStateException: catalogo corrupto
  etapa 5: SI se ejecuta, con el respaldo
resultado: valor-de-respaldo

// TRAMPA: parece que el exceptionally protege toda la cadena.
CompletableFuture<Path> f = CompletableFuture
        .supplyAsync(() -> cargarCatalogo())
        .exceptionally(e -> Catalogo.vacio())      // protege lo de ARRIBA
        .thenApply(c -> calcularMultas(c))          // <-- SIN proteger
        .thenApply(m -> exportar(m));               // <-- SIN proteger

// Si calcularMultas o exportar fallan, la excepcion sale por join()
// y nadie la maneja. El exceptionally quedo demasiado arriba.

// CORRECTO: el manejador al FINAL de la cadena.
CompletableFuture<Path> g = CompletableFuture
        .supplyAsync(() -> cargarCatalogo())
        .thenApply(c -> calcularMultas(c))
        .thenApply(m -> exportar(m))
        .exceptionally(e -> {                       // protege TODO lo anterior
            LOG.log(Level.SEVERE, "Cadena fallida", e);
            return Path.of("informes/error.txt");
        });

// Si esta cadena falla, NO PASA NADA VISIBLE. No hay traza,
// no hay log, no hay excepcion. El fallo simplemente desaparece.
CompletableFuture.supplyAsync(() -> cargarCatalogo())
                 .thenAccept(c -> catalogoGlobal = c);

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

// orTimeout: si no se completa en el plazo, FALLA con TimeoutException.
CompletableFuture<Catalogo> conPlazo = CompletableFuture
        .supplyAsync(() -> cargarCatalogo(), POOL_ES)
        .orTimeout(5, TimeUnit.SECONDS)
        .exceptionally(error -> {
            if (causaDe(error) instanceof java.util.concurrent.TimeoutException) {
                LOG.warning("La carga excedio los 5 segundos");
            }
            return Catalogo.vacio();
        });

// completeOnTimeout: si no se completa en el plazo, se completa
// con el VALOR POR DEFECTO. No falla.
CompletableFuture<Catalogo> conDefecto = CompletableFuture
        .supplyAsync(() -> cargarCatalogo(), POOL_ES)
        .completeOnTimeout(Catalogo.vacio(), 5, TimeUnit.SECONDS);

package com.nexussoftware.bibliotech.persistencia;

import java.util.concurrent.CompletableFuture;

public class AdaptadorCallbacks {

    /**
     * API antigua basada en callbacks. Es incomoda de componer:
     * anidar tres de estas produce la "piramide de la muerte".
     */
    interface LectorConCallback {
        void leerAsync(String isbn, Callback<Material> callback);
    }

    interface Callback<T> {
        void alCompletar(T resultado);
        void alFallar(Throwable error);
    }

    private final LectorConCallback lector;

    public AdaptadorCallbacks(LectorConCallback lector) { this.lector = lector; }

    /**
     * Convierte la API de callbacks en un CompletableFuture componible.
     * Es el patron estandar para modernizar codigo heredado sin tocarlo.
     */
    public CompletableFuture<Material> leer(String isbn) {

        // 1. Futuro vacio, sin tarea asociada.
        CompletableFuture<Material> futuro = new CompletableFuture<>();

        // 2. Se lanza la operacion con un callback que lo completa.
        lector.leerAsync(isbn, new Callback<Material>() {
            @Override
            public void alCompletar(Material m) {
                futuro.complete(m);                    // exito
            }
            @Override
            public void alFallar(Throwable error) {
                futuro.completeExceptionally(error);   // fallo
            }
        });

        // 3. Se devuelve inmediatamente, sin esperar al callback.
        return futuro;
    }

    /**
     * Y ahora la API antigua es componible como cualquier otra:
     * tres lecturas en paralelo y combinacion, sin anidamiento.
     */
    public CompletableFuture<String> compararTres(String i1, String i2, String i3) {
        return leer(i1)
                .thenCombine(leer(i2), (a, b) -> a.titulo() + " / " + b.titulo())
                .thenCombine(leer(i3), (par, c) -> par + " / " + c.titulo());
    }
}

CompletableFuture<Catalogo> f =
        CompletableFuture.supplyAsync(() -> cargarCatalogoLento(), POOL_ES);

TimeUnit.SECONDS.sleep(1);
boolean cancelado = f.cancel(true);      // el 'true' se IGNORA

System.out.println("cancel(): " + cancelado);        // true
System.out.println("isCancelled(): " + f.isCancelled());  // true
// PERO: el hilo de POOL_ES sigue ejecutando cargarCatalogoLento()
// hasta el final. El parametro mayInterruptIfRunning NO HACE NADA.

package com.nexussoftware.bibliotech.persistencia;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Cancelacion REAL de una tarea asincrona: una bandera compartida
 * que la tarea consulta. Es la interrupcion cooperativa de 08-02
 * adaptada a CompletableFuture, que no puede interrumpir por si solo.
 */
public class ImportacionCancelable {

    private final AtomicBoolean cancelada = new AtomicBoolean(false);
    private final CompletableFuture<Informe> futuro;

    public ImportacionCancelable(Path fichero) {
        this.futuro = CompletableFuture.supplyAsync(() -> importar(fichero), POOL_ES);
    }

    private Informe importar(Path fichero) {
        int procesadas = 0;
        for (String linea : leerLineas(fichero)) {

            // PUNTO DE CANCELACION explicito.
            if (cancelada.get() || Thread.currentThread().isInterrupted()) {
                throw new CancellationException(
                        "Importacion cancelada tras " + procesadas + " lineas");
            }
            procesar(linea);
            procesadas++;
        }
        return new Informe(procesadas);
    }

    /** Cancelacion que SI detiene el trabajo. */
    public void cancelar() {
        cancelada.set(true);          // la tarea lo vera y abortara
        futuro.cancel(false);         // y el futuro se completa ya
    }

    public CompletableFuture<Informe> futuro() { return futuro; }
}

// PESIMO: join() dentro de una etapa bloquea un hilo del pool.
// Con suficientes cadenas asi, el pool se agota y todo se para.
.thenApply(catalogo -> {
    Double multas = calcularMultasAsync(catalogo).join();   // BLOQUEA
    return multas;
})

// CORRECTO: thenCompose aplana sin bloquear a nadie.
.thenCompose(catalogo -> calcularMultasAsync(catalogo))

// get(): lanza ExecutionException e InterruptedException, ambas COMPROBADAS.
try {
    Catalogo c = futuro.get();
} catch (ExecutionException | InterruptedException e) { ... }

// join(): lanza CompletionException, NO comprobada. Mas comodo
// dentro de lambdas, donde una excepcion comprobada no compila.
Catalogo c = futuro.join();

// MAL: una sola expresion de veinte lineas.

// BIEN: cada paso con nombre.
public CompletableFuture<Path> generarInformeMensual() {
    return cargarCatalogo()
            .thenCompose(this::enriquecerConPrestamos)
            .thenApplyAsync(this::calcularMultas, POOL_CALCULO)
            .thenApplyAsync(this::formatear, POOL_CALCULO)
            .thenApplyAsync(this::escribirFichero, POOL_ES)
            .orTimeout(60, TimeUnit.SECONDS)
            .whenComplete(this::registrarMetricas)
            .exceptionally(this::informeDeError);
}

package com.nexussoftware.bibliotech.servicio;

import com.nexussoftware.bibliotech.dominio.*;

import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Generacion asincrona del informe mensual de BiblioTech.
 *
 * La cadena completa:
 *   1. Cargar el catalogo (E/S)          -- en paralelo con 2
 *   2. Cargar los prestamos activos (E/S) -- en paralelo con 1
 *   3. Combinar ambos
 *   4. Calcular las multas (CPU)
 *   5. Formatear el informe (CPU)
 *   6. Escribir el fichero (E/S)
 *
 * NINGUN hilo se bloquea esperando: el menu sigue atendiendo al usuario
 * durante toda la operacion.
 */
public class GeneradorInformeAsincrono implements AutoCloseable {

    private static final Logger LOG =
            Logger.getLogger(GeneradorInformeAsincrono.class.getName());

    /** Pool de E/S: muchos hilos porque casi todo es espera (08-01). */
    private final ExecutorService poolEs;
    /** Pool de calculo: un hilo por nucleo. */
    private final ExecutorService poolCalculo;

    private final CatalogoConcurrente catalogo;
    private final RegistroPrestamosSeguro registro;
    private final CalculadoraMultas calculadora;

    public GeneradorInformeAsincrono(CatalogoConcurrente catalogo,
                                     RegistroPrestamosSeguro registro,
                                     CalculadoraMultas calculadora) {
        this.catalogo = catalogo;
        this.registro = registro;
        this.calculadora = calculadora;

        AtomicInteger nEs = new AtomicInteger(1);
        this.poolEs = new ThreadPoolExecutor(
                16, 16, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(200),
                r -> new Thread(r, "bibliotech-async-es-" + nEs.getAndIncrement()),
                new ThreadPoolExecutor.CallerRunsPolicy());

        AtomicInteger nCpu = new AtomicInteger(1);
        this.poolCalculo = Executors.newFixedThreadPool(
                Runtime.getRuntime().availableProcessors(),
                r -> new Thread(r, "bibliotech-async-cpu-" + nCpu.getAndIncrement()));
    }

    // ---------- ETAPAS ----------

    private CompletableFuture<List<Material>> cargarMateriales() {
        return CompletableFuture.supplyAsync(() -> {
            traza("cargando materiales");
            dormir(400);                       // E/S simulada
            return catalogo.porTipo(TipoMaterial.LIBRO);
        }, poolEs);
    }

    private CompletableFuture<List<Prestamo>> cargarPrestamosActivos() {
        return CompletableFuture.supplyAsync(() -> {
            traza("cargando prestamos");
            dormir(500);                       // E/S simulada, EN PARALELO
            return registro.todosLosActivos();
        }, poolEs);
    }

    /** Etapa de CPU: se manda al pool de calculo explicitamente. */
    private DatosInforme calcular(List<Material> materiales, List<Prestamo> prestamos) {
        traza("calculando multas");
        double total = 0;
        int vencidos = 0;
        for (Prestamo p : prestamos) {
            double m = calculadora.calcular(p);
            if (m > 0) { total += m; vencidos++; }
        }
        return new DatosInforme(materiales.size(), prestamos.size(), vencidos, total);
    }

    private String formatear(DatosInforme d) {
        traza("formateando");
        return """
               ===========================================
                 BiblioTech - Informe mensual de multas
                 Nexus Software
               ===========================================
                 Materiales en catalogo : %d
                 Prestamos activos      : %d
                 Prestamos vencidos     : %d
                 Multas acumuladas      : %.2f EUR
               ===========================================
               """.formatted(d.materiales(), d.prestamos(), d.vencidos(), d.multas());
    }

    private Path escribir(String contenido) {
        traza("escribiendo fichero");
        dormir(300);                           // E/S simulada
        Path destino = Path.of("informes", "multas-mensual.txt");
        // En el codigo real: EscrituraAtomica.escribir(destino, contenido) de 07-06
        return destino;
    }

    // ---------- LA CADENA ----------

    /**
     * Declara toda la cadena y RETORNA INMEDIATAMENTE.
     * El hilo que llama no espera a nada.
     */
    public CompletableFuture<Path> generar() {

        long inicio = System.nanoTime();

        // 1 y 2 arrancan A LA VEZ: no se encadenan, se combinan.
        CompletableFuture<List<Material>> materiales = cargarMateriales();
        CompletableFuture<List<Prestamo>> prestamos = cargarPrestamosActivos();

        return materiales
                // 3. Esperar a ambos y combinarlos. El trabajo se manda al
                //    pool de calculo con thenCombineAsync, para no ocupar
                //    un hilo de E/S con trabajo de CPU.
                .thenCombineAsync(prestamos, this::calcular, poolCalculo)

                // 4. Formatear: CPU, mismo pool.
                .thenApplyAsync(this::formatear, poolCalculo)

                // 5. Escribir: E/S, pool de E/S.
                .thenApplyAsync(this::escribir, poolEs)

                // 6. Plazo global.
                .orTimeout(30, TimeUnit.SECONDS)

                // 7. Observar sin alterar: metricas.
                .whenComplete((ruta, error) -> {
                    long ms = (System.nanoTime() - inicio) / 1_000_000;
                    if (error == null) {
                        LOG.log(Level.INFO, "Informe generado en {0} ms: {1}",
                                new Object[] { ms, ruta });
                    } else {
                        LOG.log(Level.SEVERE, "Informe fallido tras " + ms + " ms", error);
                    }
                })

                // 8. Manejar: al FINAL, para cubrir toda la cadena.
                .exceptionally(error -> {
                    Throwable causa = causaDe(error);
                    if (causa instanceof TimeoutException) {
                        LOG.warning("El informe excedio los 30 segundos");
                    }
                    return Path.of("informes", "informe-no-disponible.txt");
                });
    }

    // ---------- AUXILIARES ----------

    record DatosInforme(int materiales, int prestamos, int vencidos, double multas) { }

    static Throwable causaDe(Throwable t) {
        return (t instanceof CompletionException || t instanceof ExecutionException)
                && t.getCause() != null ? t.getCause() : t;
    }

    private static void traza(String etapa) {
        System.out.printf("      [%s] %s%n", Thread.currentThread().getName(), etapa);
    }

    private static void dormir(long ms) {
        try { TimeUnit.MILLISECONDS.sleep(ms); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }

    @Override
    public void close() {
        for (ExecutorService ej : List.of(poolCalculo, poolEs)) {
            ej.shutdown();
            try {
                if (!ej.awaitTermination(15, TimeUnit.SECONDS)) ej.shutdownNow();
            } catch (InterruptedException e) {
                ej.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }
}

package com.nexussoftware.bibliotech.presentacion;

import java.nio.file.Path;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class MenuConInformeAsincrono {

    public static void main(String[] args) throws Exception {

        try (GeneradorInformeAsincrono generador = new GeneradorInformeAsincrono(
                catalogo, registro, calculadora)) {

            System.out.println("=== BiblioTech - Nexus Software ===");
            System.out.println("Lanzando informe mensual en segundo plano...\n");

            // La cadena entera se declara y retorna al instante.
            CompletableFuture<Path> informe = generador.generar();

            // Registramos que hacer al terminar. NO esperamos.
            informe.thenAccept(ruta ->
                    System.out.println("\n>>> Informe listo en: " + ruta));

            // El menu sigue atendiendo al usuario.
            for (int i = 1; i <= 8; i++) {
                System.out.printf("  [menu] opcion %d atendida (el menu NO esta bloqueado)%n", i);
                TimeUnit.MILLISECONDS.sleep(200);
            }

            // Solo al final, si de verdad hace falta el resultado, se espera.
            Path ruta = informe.get(30, TimeUnit.SECONDS);
            System.out.println("\n[main] confirmado: " + ruta);
        }
    }
}

=== BiblioTech - Nexus Software ===
Lanzando informe mensual en segundo plano...

  [menu] opcion 1 atendida (el menu NO esta bloqueado)
      [bibliotech-async-es-1] cargando materiales
      [bibliotech-async-es-2] cargando prestamos
  [menu] opcion 2 atendida (el menu NO esta bloqueado)
  [menu] opcion 3 atendida (el menu NO esta bloqueado)
      [bibliotech-async-cpu-1] calculando multas
      [bibliotech-async-cpu-1] formateando
  [menu] opcion 4 atendida (el menu NO esta bloqueado)
      [bibliotech-async-es-3] escribiendo fichero

>>> Informe listo en: informes/multas-mensual.txt
  [menu] opcion 5 atendida (el menu NO esta bloqueado)
  [menu] opcion 6 atendida (el menu NO esta bloqueado)
  [menu] opcion 7 atendida (el menu NO esta bloqueado)
  [menu] opcion 8 atendida (el menu NO esta bloqueado)
INFO: Informe generado en 812 ms: informes/multas-mensual.txt

[main] confirmado: informes/multas-mensual.txt

// Con CompletableFuture: asincrono, componible, y de lectura exigente.
CompletableFuture<Path> f = CompletableFuture
        .supplyAsync(() -> cargarCatalogo(), poolEs)
        .thenApplyAsync(this::calcularMultas, poolCpu)
        .thenApplyAsync(this::exportar, poolEs)
        .exceptionally(this::respaldo);

// Con hilos virtuales: codigo SECUENCIAL normal, con try/catch normal,
// que bloquea un hilo virtual —cuyo coste es de nanosegundos—.
try (var ejecutor = Executors.newVirtualThreadPerTaskExecutor()) {
    ejecutor.submit(() -> {
        Catalogo c = cargarCatalogo();      // "bloquea", pero es baratisimo
        double m = calcularMultas(c);
        return exportar(m);
    });
}

Future<Catalogo> f1 = ejecutor.submit(() -> cargarCatalogo());
Catalogo c = f1.get();
Future<List<Prestamo>> f2 = ejecutor.submit(() -> cargarPrestamos(c));
List<Prestamo> p = f2.get();
Future<Double> f3 = ejecutor.submit(() -> calcularMultas(p));
double total = f3.get();
Future<Path> f4 = ejecutor.submit(() -> exportar(total));
Path informe = f4.get();

package com.nexussoftware.bibliotech.servicio;

import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class CadenaInforme {

    private final ExecutorService poolEs;
    private final ExecutorService poolCpu;

    public CadenaInforme() {
        AtomicInteger a = new AtomicInteger(1);
        this.poolEs = Executors.newFixedThreadPool(8,
                r -> new Thread(r, "informe-es-" + a.getAndIncrement()));
        AtomicInteger b = new AtomicInteger(1);
        this.poolCpu = Executors.newFixedThreadPool(
                Runtime.getRuntime().availableProcessors(),
                r -> new Thread(r, "informe-cpu-" + b.getAndIncrement()));
    }

    // --- Operaciones simuladas ---

    private Catalogo cargarCatalogo() {
        dormir(300);
        return new Catalogo();
    }

    /**
     * ASINCRONA: devuelve un CompletableFuture, no un valor.
     * Por eso se encadena con thenCompose y no con thenApply.
     */
    private CompletableFuture<List<Prestamo>> cargarPrestamosAsync(Catalogo c) {
        return CompletableFuture.supplyAsync(() -> {
            dormir(300);
            return List.<Prestamo>of();
        }, poolEs);
    }

    private double calcularMultas(List<Prestamo> p) {
        dormir(300);
        return 137.50;
    }

    private Path exportar(double total) {
        dormir(300);
        return Path.of("informes/multas.txt");
    }

    // --- VERSION BLOQUEANTE (la original) ---

    public Path versionConFuture() throws Exception {
        ExecutorService ej = Executors.newFixedThreadPool(4);
        try {
            Future<Catalogo> f1 = ej.submit(this::cargarCatalogo);
            Catalogo c = f1.get();                                 // BLOQUEA

            Future<List<Prestamo>> f2 = ej.submit(() -> {
                dormir(300); return List.<Prestamo>of();
            });
            List<Prestamo> p = f2.get();                           // BLOQUEA

            Future<Double> f3 = ej.submit(() -> calcularMultas(p));
            double total = f3.get();                               // BLOQUEA

            Future<Path> f4 = ej.submit(() -> exportar(total));
            return f4.get();                                       // BLOQUEA
        } finally {
            ej.shutdown();
        }
    }

    // --- VERSION ASINCRONA ---

    public CompletableFuture<Path> versionConCompletableFuture() {

        long inicio = System.nanoTime();

        return CompletableFuture
                // 1. E/S: pool de E/S
                .supplyAsync(this::cargarCatalogo, poolEs)

                // 2. thenCompose porque cargarPrestamosAsync devuelve
                //    un CompletableFuture. Con thenApply obtendriamos
                //    CompletableFuture<CompletableFuture<List<Prestamo>>>.
                .thenCompose(this::cargarPrestamosAsync)

                // 3. CPU: pool de calculo
                .thenApplyAsync(this::calcularMultas, poolCpu)

                // 4. E/S: pool de E/S
                .thenApplyAsync(this::exportar, poolEs)

                // 5. Plazo global
                .orTimeout(20, TimeUnit.SECONDS)

                // 6. Observar sin alterar
                .whenComplete((ruta, error) -> {
                    long ms = (System.nanoTime() - inicio) / 1_000_000;
                    System.out.printf("  [cadena] terminada en %d ms (%s)%n",
                            ms, error == null ? "OK" : "ERROR");
                })

                // 7. Manejar al FINAL: cubre todas las etapas anteriores
                .exceptionally(error -> {
                    System.err.println("  [cadena] fallo: " + causaDe(error).getMessage());
                    return Path.of("informes/no-disponible.txt");
                });
    }

    static Throwable causaDe(Throwable t) {
        return (t instanceof CompletionException || t instanceof ExecutionException)
                && t.getCause() != null ? t.getCause() : t;
    }

    static void dormir(long ms) {
        try { TimeUnit.MILLISECONDS.sleep(ms); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }

    public void cerrar() {
        poolEs.shutdown();
        poolCpu.shutdown();
    }

    public static void main(String[] args) throws Exception {
        CadenaInforme ci = new CadenaInforme();
        try {
            System.out.println("=== VERSION CON FUTURE (bloqueante) ===");
            long t1 = System.nanoTime();
            Path p1 = ci.versionConFuture();
            long ms1 = (System.nanoTime() - t1) / 1_000_000;
            System.out.println("  resultado: " + p1 + "  (" + ms1 + " ms)");
            System.out.println("  el hilo main estuvo bloqueado 4 veces\n");

            System.out.println("=== VERSION CON COMPLETABLEFUTURE ===");
            long t2 = System.nanoTime();
            CompletableFuture<Path> f = ci.versionConCompletableFuture();
            long msDeclaracion = (System.nanoTime() - t2) / 1_000_000;
            System.out.println("  cadena declarada en " + msDeclaracion + " ms");
            System.out.println("  el hilo main sigue libre; hago otras cosas...");

            for (int i = 1; i <= 5; i++) {
                System.out.println("    trabajo del main " + i);
                TimeUnit.MILLISECONDS.sleep(150);
            }

            Path p2 = f.get(20, TimeUnit.SECONDS);
            long ms2 = (System.nanoTime() - t2) / 1_000_000;
            System.out.println("  resultado: " + p2 + "  (" + ms2 + " ms)");
        } finally {
            ci.cerrar();
        }
    }
}

=== VERSION CON FUTURE (bloqueante) ===
  resultado: informes/multas.txt  (1214 ms)
  el hilo main estuvo bloqueado 4 veces

=== VERSION CON COMPLETABLEFUTURE ===
  cadena declarada en 3 ms
  el hilo main sigue libre; hago otras cosas...
    trabajo del main 1
    trabajo del main 2
    trabajo del main 3
    trabajo del main 4
    trabajo del main 5
  [cadena] terminada en 1208 ms (OK)
  resultado: informes/multas.txt  (1211 ms)

package com.nexussoftware.bibliotech.servicio;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ServicioFichas implements AutoCloseable {

    record DatosMaterial(String titulo, String autor, String isbn) {
        static DatosMaterial desconocido(String isbn) {
            return new DatosMaterial("(titulo no disponible)", "(autor desconocido)", isbn);
        }
    }

    record FichaCompleta(DatosMaterial material, int prestamos, double valoracion) {
        static FichaCompleta minima(String isbn) {
            return new FichaCompleta(DatosMaterial.desconocido(isbn), -1, -1.0);
        }
        @Override public String toString() {
            return String.format("%s / %s [%s] - %s prestamos - valoracion %s",
                    material.titulo(), material.autor(), material.isbn(),
                    prestamos < 0 ? "?" : prestamos,
                    valoracion < 0 ? "?" : String.format("%.1f", valoracion));
        }
    }

    private final ExecutorService pool;
    private final boolean fallarValoraciones;

    public ServicioFichas(boolean fallarValoraciones) {
        this.fallarValoraciones = fallarValoraciones;
        AtomicInteger n = new AtomicInteger(1);
        this.pool = Executors.newFixedThreadPool(8,
                r -> new Thread(r, "fichas-" + n.getAndIncrement()));
    }

    // --- Las tres consultas, cada una con su respaldo ---

    private CompletableFuture<DatosMaterial> consultarMaterial(String isbn) {
        return CompletableFuture.supplyAsync(() -> {
            dormir(400);
            return new DatosMaterial("Java Efectivo", "J. Bloch", isbn);
        }, pool)
        // Respaldo INDIVIDUAL: si esta consulta falla, el conjunto
        // sigue adelante con un valor degradado.
        .exceptionally(e -> {
            System.err.println("  [respaldo] material no disponible: " + causaDe(e).getMessage());
            return DatosMaterial.desconocido(isbn);
        });
    }

    private CompletableFuture<Integer> consultarPrestamos(String isbn) {
        return CompletableFuture.supplyAsync(() -> {
            dormir(600);                       // la mas lenta: marca el total
            return 47;
        }, pool)
        .exceptionally(e -> -1);
    }

    private CompletableFuture<Double> consultarValoracion(String isbn) {
        return CompletableFuture.supplyAsync(() -> {
            dormir(300);
            if (fallarValoraciones) {
                throw new IllegalStateException("servicio de valoraciones caido");
            }
            return 4.6;
        }, pool)
        .exceptionally(e -> {
            System.err.println("  [respaldo] valoracion no disponible: "
                    + causaDe(e).getMessage());
            return -1.0;
        });
    }

    // --- La combinacion ---

    public CompletableFuture<FichaCompleta> fichaCompleta(String isbn) {

        // Las TRES se lanzan a la vez: la asignacion ya dispara el trabajo.
        CompletableFuture<DatosMaterial> material   = consultarMaterial(isbn);
        CompletableFuture<Integer>       prestamos  = consultarPrestamos(isbn);
        CompletableFuture<Double>        valoracion = consultarValoracion(isbn);

        return material
                // thenCombine encadenado: primero material+prestamos...
                .thenCombine(prestamos, (m, p) -> new Object[] { m, p })
                // ...y luego el par con la valoracion.
                .thenCombine(valoracion, (par, v) -> new FichaCompleta(
                        (DatosMaterial) par[0], (Integer) par[1], v))
                // Si el conjunto tarda demasiado, ficha minima en lugar de fallo.
                .completeOnTimeout(FichaCompleta.minima(isbn), 2, TimeUnit.SECONDS);
    }

    static Throwable causaDe(Throwable t) {
        return (t instanceof CompletionException || t instanceof ExecutionException)
                && t.getCause() != null ? t.getCause() : t;
    }

    static void dormir(long ms) {
        try { TimeUnit.MILLISECONDS.sleep(ms); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }

    @Override public void close() { pool.shutdown(); }

    public static void main(String[] args) throws Exception {

        System.out.println("=== CASO NORMAL ===");
        try (ServicioFichas s = new ServicioFichas(false)) {
            long t = System.nanoTime();
            FichaCompleta f = s.fichaCompleta("978-0000000001").get();
            System.out.println("  " + f);
            System.out.printf("  tiempo: %d ms (secuencial serian 1300)%n",
                    (System.nanoTime() - t) / 1_000_000);
        }

        System.out.println("\n=== CON EL SERVICIO DE VALORACIONES CAIDO ===");
        try (ServicioFichas s = new ServicioFichas(true)) {
            long t = System.nanoTime();
            FichaCompleta f = s.fichaCompleta("978-0000000001").get();
            System.out.println("  " + f);
            System.out.printf("  tiempo: %d ms%n", (System.nanoTime() - t) / 1_000_000);
            System.out.println("  La ficha se construye igual: degradacion, no fallo (06-07).");
        }
    }
}

=== CASO NORMAL ===
  Java Efectivo / J. Bloch [978-0000000001] - 47 prestamos - valoracion 4.6
  tiempo: 612 ms (secuencial serian 1300)

=== CON EL SERVICIO DE VALORACIONES CAIDO ===
  [respaldo] valoracion no disponible: servicio de valoraciones caido
  Java Efectivo / J. Bloch [978-0000000001] - 47 prestamos - valoracion ?
  tiempo: 608 ms
  La ficha se construye igual: degradacion, no fallo (06-07).

package com.nexussoftware.bibliotech.servicio;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

public class EnvioAvisosAsincrono implements AutoCloseable {

    record Aviso(String id, String empleado) { }

    record ResultadoAviso(String id, boolean correcto, String detalle) {
        static ResultadoAviso exito(Aviso a) {
            return new ResultadoAviso(a.id(), true, "enviado");
        }
        static ResultadoAviso fallo(Aviso a, String motivo) {
            return new ResultadoAviso(a.id(), false, motivo);
        }
    }

    record ResumenEnvio(int total, int enviados, int fallidos, long ms) {
        double porcentaje() { return total == 0 ? 0 : 100.0 * enviados / total; }
    }

    private final ExecutorService poolEs;
    private final LongAdder completados = new LongAdder();

    public EnvioAvisosAsincrono() {
        AtomicInteger n = new AtomicInteger(1);
        this.poolEs = new ThreadPoolExecutor(
                16, 16, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(300),
                r -> new Thread(r, "bibliotech-aviso-" + n.getAndIncrement()),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    /** Envio individual: 250 ms de E/S y fallo simulado en algunos. */
    private ResultadoAviso enviar(Aviso a) {
        try {
            TimeUnit.MILLISECONDS.sleep(250);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new CompletionException(e);
        }
        if (a.id().hashCode() % 29 == 0) {
            throw new IllegalStateException("empleado sin direccion de contacto");
        }
        return ResultadoAviso.exito(a);
    }

    /** La cadena completa. Retorna inmediatamente. */
    public CompletableFuture<ResumenEnvio> enviarTodos(List<Aviso> avisos) {

        long inicio = System.nanoTime();
        final int total = avisos.size();

        // 1-2. Un futuro por aviso, BLINDADO con su propio exceptionally.
        //      Sin ese blindaje, un solo fallo haria fallar el allOf entero.
        List<CompletableFuture<ResultadoAviso>> futuros = new ArrayList<>(total);
        for (Aviso a : avisos) {
            futuros.add(CompletableFuture
                    .supplyAsync(() -> enviar(a), poolEs)
                    .exceptionally(e -> ResultadoAviso.fallo(a, causaDe(e).getMessage()))
                    // Se cuenta al terminar, con exito o con fallo:
                    // asi el main puede mostrar progreso sin bloquearse.
                    .whenComplete((r, e) -> completados.increment()));
        }

        // 3. Agregar.
        CompletableFuture<Void> todos =
                CompletableFuture.allOf(futuros.toArray(new CompletableFuture[0]));

        return todos
                // 4. Recoger y resumir.
                .thenApply(v -> {
                    int ok = 0, ko = 0;
                    for (CompletableFuture<ResultadoAviso> f : futuros) {
                        // join() no bloquea: allOf garantiza que estan completos.
                        if (f.join().correcto()) ok++; else ko++;
                    }
                    return new ResumenEnvio(total, ok, ko,
                            (System.nanoTime() - inicio) / 1_000_000);
                })
                // 5. Plazo global y manejador final.
                .orTimeout(60, TimeUnit.SECONDS)
                .exceptionally(e -> {
                    System.err.println("Envio masivo fallido: " + causaDe(e).getMessage());
                    return new ResumenEnvio(total, 0, total,
                            (System.nanoTime() - inicio) / 1_000_000);
                });
    }

    public long completados() { return completados.sum(); }

    static Throwable causaDe(Throwable t) {
        return (t instanceof CompletionException || t instanceof ExecutionException)
                && t.getCause() != null ? t.getCause() : t;
    }

    @Override public void close() {
        poolEs.shutdown();
        try {
            if (!poolEs.awaitTermination(30, TimeUnit.SECONDS)) poolEs.shutdownNow();
        } catch (InterruptedException e) {
            poolEs.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) throws Exception {

        List<Aviso> avisos = new ArrayList<>();
        for (int i = 1; i <= 200; i++) {
            avisos.add(new Aviso("AV-" + i, "empleado-" + (i % 20)));
        }

        try (EnvioAvisosAsincrono servicio = new EnvioAvisosAsincrono()) {

            System.out.println("=== ENVIO ASINCRONO DE 200 AVISOS ===");
            System.out.println("Secuencial serian " + (200 * 250 / 1000) + " s\n");

            // 6. La cadena se declara y retorna: main NO se bloquea.
            CompletableFuture<ResumenEnvio> cadena = servicio.enviarTodos(avisos);

            // Progreso, leyendo el LongAdder que las tareas incrementan.
            while (!cadena.isDone()) {
                long hechos = servicio.completados();
                System.out.printf("\r  [%s] %d/200 (%.0f%%)",
                        barra(hechos, 200), hechos, 100.0 * hechos / 200);
                TimeUnit.MILLISECONDS.sleep(250);
            }

            ResumenEnvio r = cadena.get();
            System.out.printf("\r  [%s] 200/200 (100%%)%n%n", barra(200, 200));
            System.out.println("=== RESUMEN ===");
            System.out.println("Total     : " + r.total());
            System.out.println("Enviados  : " + r.enviados());
            System.out.println("Fallidos  : " + r.fallidos());
            System.out.println("Tiempo    : " + r.ms() + " ms");
            System.out.printf ("Exito     : %.1f%%%n", r.porcentaje());
            System.out.printf ("Aceleracion: %.1fx%n", 200 * 250.0 / r.ms());
        }
    }

    static String barra(long hechos, long total) {
        int ancho = 30;
        int llenos = (int) (ancho * hechos / total);
        return "#".repeat(llenos) + "-".repeat(ancho - llenos);
    }
}

=== ENVIO ASINCRONO DE 200 AVISOS ===
Secuencial serian 50 s

  [##############################] 200/200 (100%)

=== RESUMEN ===
Total     : 200
Enviados  : 193
Fallidos  : 7
Tiempo    : 3387 ms
Exito     : 96.5%
Aceleracion: 14.8x