import dev.modzuozhi.core.compat.ManagedFutureAwait; import java.util.concurrent.*; /** Standalone regression: dimension work needs a callback on the waiting server thread. */ public final class ManagedFutureAwaitRegression { public static void main(String[] args) throws Exception { Thread serverThread = Thread.currentThread(); BlockingQueue serverTasks = new LinkedBlockingQueue<>(); CompletableFuture chunk = new CompletableFuture<>(); Thread worker = new Thread(() -> { try { CompletableFuture physics = new CompletableFuture<>(); serverTasks.add(() -> { if (Thread.currentThread() != serverThread) { physics.completeExceptionally(new AssertionError("wrong callback thread")); } else { physics.complete(null); } }); physics.get(3, TimeUnit.SECONDS); chunk.complete(42); } catch (Throwable failure) { chunk.completeExceptionally(failure); } }, "test-dimension-worker"); worker.start(); int value = ManagedFutureAwait.join(chunk, done -> { long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); while (!done.getAsBoolean()) { if (System.nanoTime() > deadline) throw new AssertionError("circular wait"); try { Runnable task = serverTasks.poll(10, TimeUnit.MILLISECONDS); if (task != null) task.run(); } catch (InterruptedException failure) { throw new AssertionError(failure); } } }); worker.join(1000); if (value != 42 || worker.isAlive()) throw new AssertionError("request did not finish"); if (ManagedFutureAwait.join(CompletableFuture.completedFuture(7), done -> { throw new AssertionError("completed future pumped callbacks"); }) != 7) throw new AssertionError("wrong result"); RuntimeException failure = new RuntimeException("expected"); try { ManagedFutureAwait.join(CompletableFuture.failedFuture(failure), done -> {}); throw new AssertionError("exception swallowed"); } catch (CompletionException expected) { if (expected.getCause() != failure) throw new AssertionError("exception changed"); } CompletableFuture cancelled = new CompletableFuture<>(); cancelled.cancel(false); try { ManagedFutureAwait.join(cancelled, done -> {}); throw new AssertionError("cancellation swallowed"); } catch (CancellationException expected) {} // An apparently server-owned cache can switch owner after the entry check. // The request must remain queued regardless; an inline-capable execute() is unsafe. BlockingQueue chunkTasks = new LinkedBlockingQueue<>(); java.util.concurrent.atomic.AtomicReference owner = new java.util.concurrent.atomic.AtomicReference<>(serverThread); java.util.concurrent.atomic.AtomicBoolean executed = new java.util.concurrent.atomic.AtomicBoolean(); int queued = ManagedFutureAwait.callQueued(() -> { if (owner.get() != Thread.currentThread()) throw new AssertionError("wrong chunk owner"); executed.set(true); return 19; }, task -> { chunkTasks.add(task); if (executed.get()) throw new AssertionError("request executed inline"); }, done -> { Thread dimension = new Thread(() -> { owner.set(Thread.currentThread()); chunkTasks.remove().run(); }); dimension.start(); try { dimension.join(2000); } catch (InterruptedException e) { throw new AssertionError(e); } if (!done.getAsBoolean()) throw new AssertionError("queued request stuck"); }); if (queued != 19 || !executed.get()) throw new AssertionError("lost queued result"); System.out.println("PASS: callback handoff, completion, failure and cancellation"); System.out.println("PASS: unconditional enqueue survives owner switch after entry"); } }