RuleEngine with Modern Java Concurrency

Engine.java

import javax.net.ssl.SSLContext;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.StructuredTaskScope;
import java.util.concurrent.StructuredTaskScope.Joiner;

/// This engine uses [JEP 506](https://openjdk.org/jeps/506) {@link ScopedValue} technique to forward execution context thread safe.
/// ScopedValues are now FINAL in Java 25!
class Engine {

    public static abstract class Rule {
        protected static final LazyConstant<SSLContext> sslContext = LazyConstant.of(() -> {
            try {
                return SSLContext.getDefault();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

        /// Models the Address of an attendee.
        /// The `record` implementation ensures that the data is immutable.
        record Address(String city, String street, int house) {
        }

        /// Read the address from the **context** via the scoped value API and generates the `Address` PoJo
        protected Address readAddress() throws InterruptedException {
            String city = CONTEXT.get().get("city");
            String street = CONTEXT.get().get("street");
            int house = Integer.parseInt(CONTEXT.get().get("house"));
            Thread.sleep(Duration.ofSeconds(1)); // simulate a time consuming operation
            return new Address(city, street, house);
        }

        protected String readName() {
            return CONTEXT.get().getOrDefault("name", "UNKNOWN");
        }

        /// Method to implement by the business rules.
        /// The implementations can Thread-safe leverage the `Rule.read*` methods.
        abstract void fire();
    }

    private final static ScopedValue<Map<String, String>> CONTEXT = ScopedValue.newInstance();

    void execute(List<Rule> rules, Map<String, String> data) {
        ScopedValue.Carrier executionScope = ScopedValue.where(CONTEXT, data);

        // Execute all rules in parallel using structured concurrency (JEP 505)
        executionScope.run(() -> {
            try (var scope = StructuredTaskScope.open(Joiner.<Object>awaitAllSuccessfulOrThrow())) {
                for (var rule : rules) {
                    scope.fork(rule::fire);
                }
                scope.join();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("Rule execution interrupted", e);
            }
        });
    }
}

Example.java

void main() {
    var merlin = Map.of(
            "name", "Merlin",
            "city", "Herne",
            "street", "Mont-Cenis-Str.",
            "house", "294");

    var falk = Map.of(
            "name", "Falk",
            "city", "Darmstadt",
            "street", "Donno",
            "house", "12");

    var engine = new Engine();
    var rules = List.<Engine.Rule>of(new PrintAttendeeInfos());

    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        for (int i = 0; i < 10; i++) {
            executor.submit(() -> engine.execute(rules, merlin));
            executor.submit(() -> engine.execute(rules, falk));
        }
    }

    IO.println("Execution finished");
}

private String getLastName(SSLContext sslContext) {
    return null;
}

MillionThreadChallange.java

import java.time.LocalTime;

void main() {
    var counter = new AtomicInteger();
    var dtf = DateTimeFormatter.ofPattern("HH:mm:ss.SSS");
    
    IO.println(String.format("Started at %s", LocalTime.now().format(dtf)));

//    try (ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor()) {
    try (ExecutorService executorService = Executors.newSingleThreadExecutor()) {
        for (int i = 0; i < 1_000_000; i++) {
            executorService.submit(() -> {
                try {
                    Thread.sleep(Duration.ofSeconds(1));
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(e);
                }
                counter.incrementAndGet();
            });
        }
    }

    IO.println("Processed: %s tasks".formatted(counter.get()));
    IO.println("Finished at %s".formatted(LocalTime.now().format(dtf)));
}

PrintAttendeeInfos.java

import java.util.concurrent.StructuredTaskScope;
import java.util.concurrent.StructuredTaskScope.Joiner;

/// This Rule leverages the [JEP 505](https://openjdk.org/jeps/505) StructuredConcurrency API to obtain data for the logic
class PrintAttendeeInfos extends Engine.Rule {
    @Override
    void fire() {
        // Use Joiner.awaitAllSuccessfulOrThrow() - fails fast on first exception,
        // cancels remaining tasks, and throws FailedException wrapping the cause
        try (var scope = StructuredTaskScope.open(Joiner.<Object>awaitAllSuccessfulOrThrow())) {

            StructuredTaskScope.Subtask<String> name = scope.fork(() -> {
                IO.println("\tRead Name");
                var tmp = readName();
                IO.println("\tGot Name");
                return tmp;
            });

            StructuredTaskScope.Subtask<Address> address = scope.fork(() -> {
                IO.println("\tRead Address");
                var tmp = readAddress();
                IO.println("\tGot Address");
                return tmp;
            });

            scope.join(); // Throws FailedException if any subtask failed

            var addr = address.get();
            IO.println("%s's home is %s %d, %s".formatted(
                name.get(), addr.street(), addr.house(), addr.city()));

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Task interrupted", e);
        } catch (StructuredTaskScope.FailedException e) {
            throw new RuntimeException("Subtask failed", e.getCause());
        }
    }
}