CompletableFuture

Detailní dokumentace k asynchronním operacím pomocí CompletableFuture v Hytale.

---

Přehled

CompletableFuture
    │
    ├── runAsync()      - Void operace na ForkJoinPool
    ├── supplyAsync()   - Vrací hodnotu
    │
    ├── thenApply()     - Transformace výsledku
    ├── thenAccept()    - Konzumace výsledku
    ├── thenRun()       - Akce po dokončení
    │
    ├── exceptionally() - Error handling
    └── whenComplete()  - Finální callback

---

CompletableFutureUtil

Hytale utility třída pro práci s futures:

public class CompletableFutureUtil {
    // Globální exception handler
    public static final Function fn = throwable -> {
        if (!(throwable instanceof TailedRuntimeException)) {
            HytaleLogger.getLogger().at(Level.SEVERE)
                .withCause(throwable)
                .log("Unhandled exception! " + Thread.currentThread());
        }
        throw new TailedRuntimeException(throwable);
    };
    // Propojení dvou futures
    @Nonnull
    public static  CompletableFuture whenComplete(
        @Nonnull CompletableFuture future,
        @Nonnull CompletableFuture callee
    ) {
        return future.whenComplete((result, throwable) -> {
            if (throwable != null) {
                callee.completeExceptionally(throwable);
            } else {
                callee.complete(result);
            }
        });
    }
    // Kontrola zda je future zrušeno
    public static boolean isCanceled(Throwable throwable) {
        return throwable instanceof CancellationException
            || (throwable instanceof CompletionException
                && throwable.getCause() != null
                && throwable.getCause() != throwable
                && isCanceled(throwable.getCause()));
    }
    // Automatický exception logging
    @Nonnull
    public static  CompletableFuture _catch(@Nonnull CompletableFuture future) {
        return future.exceptionally((Function)fn);
    }
    // Vytvoření zrušeného future
    @Nonnull
    public static  CompletableFuture completionCanceled() {
        CompletableFuture out = new CompletableFuture<>();
        out.cancel(false);
        return out;
    }
}

---

joinWithProgress

Čekání na více futures s progress callback:

public static void joinWithProgress(
    @Nonnull List> list,
    @Nonnull ProgressConsumer callback,
    int millisSleep,
    int millisProgress
) throws InterruptedException {
    CompletableFuture all = CompletableFuture.allOf(
        list.toArray(CompletableFuture[]::new)
    );
    long last = System.nanoTime();
    long nanosProgress = TimeUnit.MILLISECONDS.toNanos(millisProgress);
    int listSize = list.size();
    while (!all.isDone()) {
        Thread.sleep(millisSleep);
        long now;
        if (last + nanosProgress < (now = System.nanoTime())) {
            last = now;
            int done = 0;
            for (CompletableFuture c : list) {
                if (c.isDone()) {
                    done++;
                }
            }
            if (done < listSize) {
                callback.accept((double)done / (double)listSize, done, listSize);
            }
        }
    }
    callback.accept(1.0, listSize, listSize);
    all.join();
}
@FunctionalInterface
public interface ProgressConsumer {
    void accept(double progress, int done, int total);
}

Příklad Použití

List> tasks = new ArrayList<>();
for (Player player : players) {
    tasks.add(CompletableFuture.runAsync(() -> savePlayer(player)));
}
CompletableFutureUtil.joinWithProgress(
    tasks,
    (progress, done, total) -> {
        getLogger().info(String.format("Saving: %.0f%% (%d/%d)", progress * 100, done, total));
    },
    100,   // Check every 100ms
    1000   // Log every 1000ms
);

---

Základní Vzory

Jednorázový Async Task

CompletableFuture.runAsync(() -> {
    // Těžká práce (I/O, databáze)
    saveToDatabase(data);
});

S Návratovou Hodnotou

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    return loadFromDatabase(uuid);
});
// Později...
PlayerData data = future.join();  // Blokující
// nebo
future.thenAccept(data -> useData(data));  // Non-blokující

Řetězení Operací

CompletableFuture
    .supplyAsync(() -> {
        // Krok 1: Načti data
        return loadRawData();
    })
    .thenApply(raw -> {
        // Krok 2: Parsuj
        return parseData(raw);
    })
    .thenApply(parsed -> {
        // Krok 3: Validuj
        return validateData(parsed);
    })
    .thenAccept(valid -> {
        // Krok 4: Použij
        world.execute(() -> applyData(valid));
    })
    .exceptionally(error -> {
        getLogger().at(Level.SEVERE).withCause(error).log("Pipeline failed");
        return null;
    });

---

World Thread Integration

Async → World Thread

CompletableFuture.supplyAsync(() -> {
    // Na async thread - I/O operace
    return loadPlayerData(uuid);
}).thenAccept(data -> {
    // POZOR: Stále na async thread!
    // Přepni na world thread pro komponenty
    world.execute(() -> {
        Ref ref = playerRef.getReference();
        if (ref != null && ref.isValid()) {
            Store store = ref.getStore();
            // Bezpečný přístup ke komponentám
            applyDataToPlayer(ref, store, data);
        }
    });
});

Čekání na World Operaci

public CompletableFuture performWorldOperation(World world, Runnable operation) {
    CompletableFuture future = new CompletableFuture<>();
    world.execute(() -> {
        try {
            operation.run();
            future.complete(null);
        } catch (Exception e) {
            future.completeExceptionally(e);
        }
    });
    return future;
}

---

Kombinování Futures

allOf - Všechny musí dokončit

List> saveFutures = new ArrayList<>();
for (Player player : Universe.get().getPlayers()) {
    CompletableFuture save = CompletableFuture.runAsync(() -> {
        savePlayer(player.getUuid());
    });
    saveFutures.add(save);
}
// Počkej na všechny
CompletableFuture.allOf(saveFutures.toArray(new CompletableFuture[0]))
    .thenRun(() -> {
        getLogger().info("All players saved!");
    });

anyOf - První dokončený

CompletableFuture