Virtual-Threads-Lab

package com.example.run8b;
public final class Run8BDemo {
    public static void main(String[] args) {
        MetricsRegistry metrics = new MetricsRegistry();
        OrderView view = OrderViewService.demoService(metrics).build(new OrderId("ORD-8B-1"));
        PerformanceDecision decision = new PerformanceAdvisor().decide(true, true, false);
        System.out.println("RUN8B_DEMO_OK " + view.order().id().value() + " " + decision.classifier());
        System.out.println(metrics.snapshot());
    }
}

package com.example.run8b;
import java.time.Duration;
import java.util.concurrent.*;
// Pattern: Application Service - koordiniert Use Case, Ports und technische Schutzgrenzen.
public final class OrderViewService {
    private final FakeOrderRepository orders;
    private final FakePaymentClient payments;
    private final FakeInventoryClient inventory;
    private final Bulkhead paymentBulkhead;
    private final Bulkhead inventoryBulkhead;
    private final TimeoutPolicy timeout;
    public OrderViewService(FakeOrderRepository orders, FakePaymentClient payments, FakeInventoryClient inventory,
                            Bulkhead paymentBulkhead, Bulkhead inventoryBulkhead, TimeoutPolicy timeout) {
        this.orders = orders; this.payments = payments; this.inventory = inventory; this.paymentBulkhead = paymentBulkhead; this.inventoryBulkhead = inventoryBulkhead; this.timeout = timeout;
    }
    public OrderView build(OrderId id) {
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            Future<OrderData> order = executor.submit(() -> orders.load(id));
            Future<PaymentData> payment = executor.submit(() -> paymentBulkhead.call(() -> payments.load(id)));
            Future<InventoryData> stock = executor.submit(() -> inventoryBulkhead.call(() -> inventory.load(id)));
            return new OrderView(timeout.get(order, "order"), timeout.get(payment, "payment"), timeout.get(stock, "inventory"), false);
        }
    }
    public static OrderViewService demoService(MetricsRegistry metrics) {
        SimulatedConnectionPool pool = new SimulatedConnectionPool(2, metrics);
        return new OrderViewService(new FakeOrderRepository(pool), new FakePaymentClient(new WindowRateLimiter(10, Duration.ofSeconds(1).toMillis())), new FakeInventoryClient(), new Bulkhead("payment", 4, metrics), new Bulkhead("inventory", 4, metrics), new TimeoutPolicy(Duration.ofMillis(500)));
    }
}

package com.example.run8b;
import java.util.concurrent.Semaphore;
import java.util.function.Supplier;
// Pattern: Bulkhead - schützt knappe Downstream-Ressourcen vor Überlast.
public final class Bulkhead {
    private final String name;
    private final Semaphore permits;
    private final MetricsRegistry metrics;
    public Bulkhead(String name, int maxConcurrentCalls, MetricsRegistry metrics) {
        this.name = name;
        this.permits = new Semaphore(maxConcurrentCalls);
        this.metrics = metrics;
    }
    public <T> T call(Supplier<T> work) {
        if (!permits.tryAcquire()) {
            metrics.increment(name + ".rejected");
            throw new BulkheadRejectedException("bulkhead rejected: " + name);
        }
        metrics.increment(name + ".accepted");
        try { return work.get(); }
        finally { permits.release(); }
    }
}

package com.example.run8b;
import java.time.Duration;
import java.util.concurrent.*;
// Pattern: Policy Object - kapselt Timeout-Entscheidung je Use Case.
public final class TimeoutPolicy {
    private final Duration timeout;
    public TimeoutPolicy(Duration timeout) { this.timeout = timeout; }
    public <T> T get(Future<T> future, String label) {
        try { return future.get(timeout.toMillis(), TimeUnit.MILLISECONDS); }
        catch (TimeoutException e) { future.cancel(true); throw new TimeoutExceededException("timeout in " + label); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new TimeoutExceededException("interrupted in " + label); }
        catch (ExecutionException e) { throw new RuntimeException(e.getCause()); }
    }
}

package com.example.run8b;
// Pattern: Rate Limiter - begrenzt Aufrufe pro Zeitfenster.
public final class WindowRateLimiter {
    private final int maxPerWindow;
    private final long windowMillis;
    private long windowStart = System.currentTimeMillis();
    private int used;
    public WindowRateLimiter(int maxPerWindow, long windowMillis) { this.maxPerWindow = maxPerWindow; this.windowMillis = windowMillis; }
    public synchronized boolean tryAcquire() {
        long now = System.currentTimeMillis();
        if (now - windowStart >= windowMillis) { windowStart = now; used = 0; }
        if (used >= maxPerWindow) return false;
        used++;
        return true;
    }
}

package com.example.run8b;
import java.util.concurrent.Semaphore;
import java.util.function.Supplier;
// Pattern: Resource Pool - modelliert begrenzte Datenbankverbindungen.
public final class SimulatedConnectionPool {
    private final Semaphore connections;
    private final MetricsRegistry metrics;
    public SimulatedConnectionPool(int size, MetricsRegistry metrics) { this.connections = new Semaphore(size); this.metrics = metrics; }
    public <T> T withConnection(Supplier<T> query) {
        if (!connections.tryAcquire()) {
            metrics.increment("db.pool.exhausted");
            throw new PoolExhaustedException("no database connection available");
        }
        metrics.increment("db.pool.used");
        try { return query.get(); }
        finally { connections.release(); }
    }
}

package com.example.run8b;
public enum WorkloadClassifier { IO_BOUND, CPU_BOUND, RESOURCE_BOUND }

package com.example.run8b;
// Pattern: Strategy/Decision Service - ordnet Workload-Profile einer Architekturentscheidung zu.
public final class PerformanceAdvisor {
    public PerformanceDecision decide(boolean waitsForNetwork, boolean usesDbPool, boolean heavyCpu) {
        if (heavyCpu) return new PerformanceDecision(WorkloadClassifier.CPU_BOUND, "limit to CPU cores and profile allocation");
        if (usesDbPool) return new PerformanceDecision(WorkloadClassifier.RESOURCE_BOUND, "use virtual threads but protect DB pool with bulkhead and timeout");
        if (waitsForNetwork) return new PerformanceDecision(WorkloadClassifier.IO_BOUND, "virtual threads are a good fit with downstream limits");
        return new PerformanceDecision(WorkloadClassifier.CPU_BOUND, "measure first; do not add concurrency blindly");
    }
}

package com.example.run8b;
import java.time.Duration;
public final class Run8BTestRunner {
    public static void main(String[] args) {
        buildsOrderViewWithVirtualThreads();
        rejectsWhenBulkheadIsFull();
        detectsRateLimit();
        classifiesResourceBoundWorkload();
        cpuChecksumIsStable();
        System.out.println("RUN8B_TESTS_OK");
    }
    static void buildsOrderViewWithVirtualThreads() {
        MetricsRegistry metrics = new MetricsRegistry();
        OrderView view = OrderViewService.demoService(metrics).build(new OrderId("ORD-1"));
        require(view.payment().authorized(), "payment should be authorized");
        require(metrics.get("payment.accepted") == 1, "payment metric missing");
    }
    static void rejectsWhenBulkheadIsFull() {
        MetricsRegistry metrics = new MetricsRegistry();
        Bulkhead bulkhead = new Bulkhead("tiny", 1, metrics);
        bulkhead.call(() -> "ok");
        require(metrics.get("tiny.accepted") == 1, "accepted metric missing");
    }
    static void detectsRateLimit() {
        WindowRateLimiter limiter = new WindowRateLimiter(1, Duration.ofSeconds(60).toMillis());
        require(limiter.tryAcquire(), "first call allowed");
        require(!limiter.tryAcquire(), "second call rejected");
    }
    static void classifiesResourceBoundWorkload() {
        PerformanceDecision d = new PerformanceAdvisor().decide(true, true, false);
        require(d.classifier() == WorkloadClassifier.RESOURCE_BOUND, "db pool workload must be resource bound");
    }
    static void cpuChecksumIsStable() {
        long a = new CpuWorkload().checksum(1000);
        long b = new CpuWorkload().checksum(1000);
        require(a == b, "checksum must be deterministic");
    }
    static void require(boolean condition, String message) { if (!condition) throw new AssertionError(message); }
}