Concurrency example

basic-concurrency/build.gradle

@@ -0,0 +1,23 @@
+plugins {
+    id 'java'
+}
+
+getTasks().withType(JavaCompile.class).configureEach {
+    getOptions().getRelease().set(9)
+}
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+    implementation libs.contracts
+    implementation libs.concurrency
+
+    testImplementation libs.contracts.test
+    testImplementation libs.concurrency.test
+}
+
+test {
+    useJUnitPlatform()
+}

basic-concurrency/src/main/java/io/github/jonloucks/examples/concurrency/basic/Command.java

@@ -0,0 +1,12 @@
+package io.github.jonloucks.examples.concurrency.basic;
+
+@FunctionalInterface
+public interface Command {
+
+    String execute();
+
+    default boolean foreground() {
+        return false;
+    }
+
+}

basic-concurrency/src/main/java/io/github/jonloucks/examples/concurrency/basic/Constants.java

@@ -0,0 +1,58 @@
+package io.github.jonloucks.examples.concurrency.basic;
+
+import io.github.jonloucks.concurrency.api.Waitable;
+import io.github.jonloucks.contracts.api.Contract;
+
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+/**
+ * Contracts can be defined anywhere, placing them here to demonstrate package private access
+ */
+final class Constants {
+
+    /**
+     *   Constant string, but open for uses cases like localization.
+     */
+    static final Contract<String> PROGRAM_NAME = Contract.create("Program name contract");
+
+    /**
+     * The main program implementation, not everything has to be in the class with main()
+     */
+    static final Contract<Program> PROGRAM = Contract.create("Program contract");
+
+    /**
+     * Example of a configuration setting that is a singleton with lazy evaluation of value
+     */
+    static final Contract<Integer> RUNNER_THREAD_COUNT = Contract.create("Number of worker threads contract");
+
+    /**
+     * Example of a shared executor with auto resource management
+     */
+    static final Contract<Dispatcher> RUNNER = Contract.create("Dispatcher contract");
+
+    /**
+     * Set to true when program is quitting
+     */
+    static final Contract<Waitable<Boolean>> IS_QUITTING = Contract.create("Quitting contract");
+
+    /**
+     * The health the program
+     */
+    static final Contract<Waitable<String>> HEALTH = Contract.create("Health contract");
+
+    /**
+     * Command Output
+     */
+    static final Contract<Consumer<String>> OUTPUT = Contract.create("Command output contract");
+
+    /**
+     * Command Input
+     */
+    static final Contract<Supplier<String>> INPUT = Contract.create("Command input contract");
+
+    /**
+     * Command Output
+     */
+    static final Contract<Consumer<String>> ERROR = Contract.create("Command error contract");
+}

basic-concurrency/src/main/java/io/github/jonloucks/examples/concurrency/basic/Dispatcher.java

@@ -0,0 +1,6 @@
+package io.github.jonloucks.examples.concurrency.basic;
+
+public interface Dispatcher {
+
+   void dispatch(Command command);
+}

basic-concurrency/src/main/java/io/github/jonloucks/examples/concurrency/basic/DispatcherImpl.java

@@ -0,0 +1,69 @@
+package io.github.jonloucks.examples.concurrency.basic;
+
+import io.github.jonloucks.concurrency.api.OnCompletion;
+import io.github.jonloucks.concurrency.api.Waitable;
+import io.github.jonloucks.contracts.api.AutoClose;
+import io.github.jonloucks.contracts.api.AutoOpen;
+
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+
+import static io.github.jonloucks.concurrency.api.GlobalConcurrency.*;
+import static io.github.jonloucks.contracts.api.GlobalContracts.claimContract;
+import static io.github.jonloucks.examples.concurrency.basic.Constants.*;
+import static java.util.Optional.ofNullable;
+
+final class DispatcherImpl implements Dispatcher, AutoOpen {
+    @Override
+    public void dispatch(Command command) {
+        incrementRunning();
+        final OnCompletion<String> onCompletion = c -> {
+            c.getValue().ifPresent(output);
+            c.getThrown().ifPresent(e ->error.accept(e.getMessage()));
+            decrementRunning();
+        };
+        if (!command.foreground() && ofNullable(delegate).isPresent()) {
+            completeLater(onCompletion, on -> delegate.execute(() -> completeNow(on, command::execute)));
+        } else {
+            completeNow(onCompletion, command::execute);
+        }
+    }
+
+    @Override
+    public AutoClose open() {
+        delegate = Executors.newFixedThreadPool(claimContract(RUNNER_THREAD_COUNT));
+        return this::privateClose; // only open caller can close
+    }
+
+    private void privateClose() {
+        ofNullable(delegate).ifPresent(executor -> {
+            delegate = null;
+            executor.shutdown();
+            try {
+                if (executor.awaitTermination(5, TimeUnit.MINUTES)) {
+                    executor.shutdownNow();
+                }
+            } catch (InterruptedException ignored) {
+                Thread.currentThread().interrupt();
+            }
+        });
+    }
+
+    private void incrementRunning() {
+        runningCommandCount.incrementAndGet();
+        health.accept("busy");
+    }
+
+    private void decrementRunning() {
+        if (runningCommandCount.decrementAndGet() == 0) {
+            health.accept("idle");
+        }
+    }
+
+    private ExecutorService delegate;
+    private final AtomicInteger runningCommandCount = new AtomicInteger();
+    private final Consumer<String> output = claimContract(OUTPUT);
+    private final Consumer<String> error = claimContract(ERROR);
+    private final Waitable<String> health = claimContract(HEALTH);
+}

basic-concurrency/src/main/java/io/github/jonloucks/examples/concurrency/basic/HelpCommand.java

@@ -0,0 +1,17 @@
+package io.github.jonloucks.examples.concurrency.basic;
+
+final class HelpCommand implements Command {
+    @Override
+    public String execute() {
+        return "This is an example...";
+    }
+
+    @Override
+    public boolean foreground() {
+        return true;
+    }
+
+    HelpCommand() {
+
+    }
+}

basic-concurrency/src/main/java/io/github/jonloucks/examples/concurrency/basic/Main.java

@@ -0,0 +1,95 @@
+package io.github.jonloucks.examples.concurrency.basic;
+
+import io.github.jonloucks.contracts.api.AutoClose;
+import io.github.jonloucks.contracts.api.BindStrategy;
+import io.github.jonloucks.contracts.api.Repository;
+
+import java.util.Scanner;
+
+import static io.github.jonloucks.concurrency.api.GlobalConcurrency.createWaitable;
+import static io.github.jonloucks.contracts.api.BindStrategy.IF_NOT_BOUND;
+import static io.github.jonloucks.contracts.api.GlobalContracts.*;
+import static io.github.jonloucks.examples.concurrency.basic.Constants.*;
+import static java.lang.Boolean.TRUE;
+
+@SuppressWarnings("unused")
+public final class Main {
+
+    /**
+     * Main entry point.
+     * Note. Entry points are where final decisions on dependency inversions are made.
+     * Not all decisions must be made in an entry point, but in this example it is exposed
+     * for visibility
+     * @param args the command line arguments
+     */
+    public static void main(String[] args) {
+        final Repository repository = createMyRepository();
+        try (AutoClose closeRepository = repository.open();
+             AutoClose closeHealthNotify = openHealthNotify()) {
+
+            final Program program = claimContract(PROGRAM);
+
+            waitForQuit();
+
+            waitForIdle();
+        } // all repository resources will be released when this try block exits
+    }
+
+    private static void waitForQuit() {
+        claimContract(IS_QUITTING).getWhen(TRUE::equals);
+    }
+
+    private static void waitForIdle() {
+       claimContract(HEALTH).getWhen("idle"::equals);
+    }
+
+    private static AutoClose openHealthNotify() {
+        return claimContract(HEALTH).notifyIf("idle"::equals, health -> {
+            System.out.println( "All commands have completed.");
+        });
+    }
+
+    /**
+     *  For complex projects with many modules and entry points there a many
+     *  ways to extend and manage dependency inversions without
+     *  have them all defined here.
+     *  For example:
+     *      each module could have its own Repository.
+     *          MyModule.createRepository(contracts)
+     *      each module could have a static method to install contracts to a given repository
+     *          MyModule.install(repository)
+     *      each module could use the ServiceLoader mechanism
+     *      I am sure there are more like SpringBoot
+     */
+    private static Repository createMyRepository() {
+        final Repository repository = claimContract(Repository.FACTORY).get();
+
+        final BindStrategy strategy = IF_NOT_BOUND;
+
+        // Constant string, but could be changed to a localized value without changing uses
+        repository.keep(PROGRAM_NAME, () -> "Concurrency Example");
+
+        // lifeCycle will create a singleton and detect AutoOpen implementations
+        repository.keep(PROGRAM, lifeCycle(ProgramImpl::new));
+
+        repository.keep(OUTPUT, singleton(() -> text -> System.out.println(text)), strategy);
+
+        repository.keep(ERROR, singleton(() -> text -> System.err.println(text)), strategy);
+
+        repository.keep(INPUT, singleton(() -> new Scanner(System.in)::nextLine), strategy);
+
+        // lazy evaluated singleton
+        repository.keep(RUNNER_THREAD_COUNT, singleton(() -> Runtime.getRuntime().availableProcessors() * 8));
+
+        // lifeCycle will create a singleton and detect AutoOpen implementations
+        repository.keep(RUNNER, lifeCycle(DispatcherImpl::new));
+
+        // set to true when program is quiting
+        repository.keep(IS_QUITTING, singleton(() -> createWaitable(false)));
+
+        // program health
+        repository.keep(HEALTH, singleton(() -> createWaitable("ready")));
+
+        return repository;
+    }
+}

basic-concurrency/src/main/java/io/github/jonloucks/examples/concurrency/basic/Program.java

@@ -0,0 +1,4 @@
+package io.github.jonloucks.examples.concurrency.basic;
+
+interface Program {
+}

basic-concurrency/src/main/java/io/github/jonloucks/examples/concurrency/basic/ProgramImpl.java

@@ -0,0 +1,77 @@
+package io.github.jonloucks.examples.concurrency.basic;
+
+import io.github.jonloucks.contracts.api.AutoClose;
+import io.github.jonloucks.contracts.api.AutoOpen;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import static io.github.jonloucks.contracts.api.GlobalContracts.claimContract;
+import static io.github.jonloucks.examples.concurrency.basic.Constants.*;
+import static java.util.Optional.ofNullable;
+
+final class ProgramImpl implements Program, AutoOpen {
+
+    @Override
+    public AutoClose open() {
+        serviceStart = Instant.now();
+        dispatcher = claimContract(Constants.RUNNER);
+        output.accept("Welcome to " + claimContract(PROGRAM_NAME));
+        new Thread(this::commandLoop).start();
+        return this::privateClose;
+    }
+
+    ProgramImpl() {
+    }
+
+    void commandLoop() {
+        while (!isQuitting()) {
+            promptForCommand().ifPresent(dispatcher::dispatch);
+        }
+    }
+
+    private Duration getUptime() {
+        return Duration.between(serviceStart, ofNullable(serviceEnd).orElseGet(Instant::now));
+    }
+
+    private boolean isQuitting() {
+        return claimContract(IS_QUITTING).get();
+    }
+
+    private Optional<Command> promptForCommand() {
+        try {
+            output.accept("Enter command: ");
+            return Optional.of(parseCommand(input.get())); // Read the entire line until a newline character
+        } catch (Exception thrown) {
+            claimContract(IS_QUITTING).accept(true);
+            return Optional.empty();
+        }
+    }
+
+    private Command parseCommand(String commandLine) {
+        switch (ofNullable(commandLine).orElse("").toLowerCase()) {
+            case "":
+            case "help":
+            case "?":
+                return new HelpCommand();
+            case "quit":
+                return new QuitCommand();
+            default:
+                return () -> "Unrecognized command: " + commandLine;
+        }
+    }
+
+    private void privateClose() {
+        serviceEnd = Instant.now();
+        output.accept("Service Uptime: " + getUptime());
+    }
+
+    private final Supplier<String> input = claimContract(INPUT);
+    private final Consumer<String> output = claimContract(OUTPUT);
+    private Instant serviceStart;
+    private Instant serviceEnd;
+    private Dispatcher dispatcher;
+}