Concurrent Programming

/**
 * Asynchronous computation result that may complete with success or failure
 */
interface Future<T> extends Value<T> {
    // Factory methods
    static <T> Future<T> of(Executor executor, CheckedFunction0<? extends T> computation);
    static <T> Future<T> of(CheckedFunction0<? extends T> computation);  // Uses ForkJoinPool
    static <T> Future<T> successful(T value);                           // Immediately completed Future
    static <T> Future<T> failed(Throwable exception);                   // Immediately failed Future
    static <T> Future<T> fromTry(Try<? extends T> result);             // Create from Try
    static <T> Future<T> fromCompletableFuture(CompletableFuture<T> future);
    
    // Completion state
    boolean isCompleted();                     // true if computation finished (success or failure)
    boolean isSuccess();                       // true if completed successfully
    boolean isFailure();                       // true if completed with exception
    boolean isCancelled();                     // true if computation was cancelled
    
    // Value access (blocking operations)
    T get();                                   // Block until completion, return value or throw
    T get(long timeout, TimeUnit unit);       // Block with timeout
    Option<Try<T>> getValue();                 // Non-blocking: get current value if completed
    
    // Transformation operations
    <U> Future<U> map(Function<? super T, ? extends U> mapper);
    <U> Future<U> mapTry(CheckedFunction1<? super T, ? extends U> mapper);
    <U> Future<U> flatMap(Function<? super T, ? extends Future<? extends U>> mapper);
    
    // Error handling
    Future<T> recover(Function<? super Throwable, ? extends T> recovery);
    Future<T> recoverWith(Function<? super Throwable, ? extends Future<? extends T>> recovery);
    
    // Filtering
    Future<T> filter(Predicate<? super T> predicate);
    Future<T> filterTry(CheckedPredicate<? super T> predicate);
    
    // Side effects (non-blocking callbacks)
    Future<T> onComplete(Consumer<? super Try<T>> action);
    Future<T> onSuccess(Consumer<? super T> action);           // Called if successful
    Future<T> onFailure(Consumer<? super Throwable> action);   // Called if failed
    
    // Combining Futures
    <U> Future<Tuple2<T, U>> zip(Future<? extends U> that);
    <U, R> Future<R> zipWith(Future<? extends U> that, BiFunction<? super T, ? super U, ? extends R> mapper);
    Future<T> fallbackTo(Future<? extends T> that);            // Use that if this fails
    
    // Conversion operations
    CompletableFuture<T> toCompletableFuture();
    Try<T> toTry();                            // Blocks until completion
    
    // Executor information
    Executor executor();                       // Get the executor used for this Future
    
    // Cancellation
    void cancel();                            // Cancel the computation if not completed
    void cancel(boolean mayInterruptIfRunning);
}

import io.vavr.concurrent.Future;
import io.vavr.control.Try;
import java.util.concurrent.Executors;

// Creating Futures
Future<String> immediate = Future.successful("Hello");
Future<Integer> computation = Future.of(() -> expensiveComputation());
Future<String> withExecutor = Future.of(
    Executors.newCachedThreadPool(),
    () -> downloadData("http://example.com")
);

// Transforming Futures
Future<String> processed = computation
    .map(i -> "Result: " + i)
    .recover(ex -> "Error: " + ex.getMessage());

// Chaining asynchronous operations
Future<String> chained = Future.of(() -> getUserId())
    .flatMap(id -> Future.of(() -> fetchUserName(id)))
    .flatMap(name -> Future.of(() -> fetchUserEmail(name)));

// Combining multiple Futures
Future<String> name = Future.of(() -> "John");
Future<Integer> age = Future.of(() -> 30);
Future<String> person = name.zipWith(age, (n, a) -> n + " is " + a + " years old");

// Error handling
Future<String> safe = Future.of(() -> riskyOperation())
    .recover(throwable -> "Default value")
    .onFailure(ex -> System.err.println("Operation failed: " + ex));

// Non-blocking callbacks
computation
    .onSuccess(result -> System.out.println("Got: " + result))
    .onFailure(error -> System.err.println("Failed: " + error))
    .onComplete(tryResult -> System.out.println("Completed: " + tryResult));

// Fallback pattern
Future<String> primary = Future.of(() -> primaryService());
Future<String> backup = Future.of(() -> backupService());
Future<String> resilient = primary.fallbackTo(backup);

// Converting to Java types
CompletableFuture<Integer> javaFuture = computation.toCompletableFuture();

// Helper methods
static int expensiveComputation() throws InterruptedException {
    Thread.sleep(1000);
    return 42;
}

static String downloadData(String url) throws IOException {
    // Simulate download
    return "Downloaded from " + url;
}

/**
 * Completable container that provides a Future and allows completing it exactly once
 */
interface Promise<T> {
    // Factory methods
    static <T> Promise<T> make();              // Create new Promise
    static <T> Promise<T> make(Executor executor); // Create with specific executor
    static <T> Promise<T> successful(T value); // Create already completed Promise
    static <T> Promise<T> failed(Throwable exception); // Create already failed Promise
    
    // Completion operations (can only be called once)
    Promise<T> success(T value);               // Complete with success value
    Promise<T> failure(Throwable exception);   // Complete with failure
    Promise<T> complete(Try<? extends T> result); // Complete with Try result
    Promise<T> completeWith(Future<? extends T> other); // Complete when other completes
    
    // Try completion operations (return false if already completed)
    boolean trySuccess(T value);               // Try to complete with success
    boolean tryFailure(Throwable exception);   // Try to complete with failure
    boolean tryComplete(Try<? extends T> result); // Try to complete with Try
    boolean tryCompleteWith(Future<? extends T> other); // Try to complete with other Future
    
    // State checking
    boolean isCompleted();                     // Check if Promise has been completed
    
    // Future access
    Future<T> future();                        // Get the Future backed by this Promise
}

import io.vavr.concurrent.Promise;
import io.vavr.concurrent.Future;
import io.vavr.control.Try;

// Basic Promise usage
Promise<String> promise = Promise.make();
Future<String> future = promise.future();

// Complete the promise in another thread
new Thread(() -> {
    try {
        Thread.sleep(1000);
        promise.success("Hello from another thread!");
    } catch (InterruptedException e) {
        promise.failure(e);
    }
}).start();

// Use the future
future.onSuccess(result -> System.out.println("Got: " + result));

// Bridging callback API to Future API
public Future<String> callbackToFuture(CallbackAPI api) {
    Promise<String> promise = Promise.make();
    
    api.doSomethingAsync(new Callback<String>() {
        @Override
        public void onSuccess(String result) {
            promise.success(result);
        }
        
        @Override
        public void onFailure(Exception error) {
            promise.failure(error);
        }
    });
    
    return promise.future();
}

// Try completion (safe for multiple calls)
Promise<Integer> safePromise = Promise.make();
boolean completed1 = safePromise.trySuccess(42);    // true
boolean completed2 = safePromise.trySuccess(24);    // false - already completed

// Completing with Try
Try<String> result = Try.of(() -> someOperation());
Promise<String> promiseFromTry = Promise.make();
promiseFromTry.complete(result);

// Promise chaining
Promise<String> step1 = Promise.make();
Future<Integer> pipeline = step1.future()
    .map(String::length)
    .filter(len -> len > 5);

step1.success("Hello World");

/**
 * Lazy asynchronous computation that starts only when run
 */
class Task<T> implements Value<T> {
    // Factory methods
    static <T> Task<T> of(CheckedFunction0<? extends T> computation);
    static <T> Task<T> of(Executor executor, CheckedFunction0<? extends T> computation);
    static <T> Task<T> successful(T value);    // Task that immediately succeeds
    static <T> Task<T> failed(Throwable exception); // Task that immediately fails
    static <T> Task<T> fromTry(Try<? extends T> result);
    static <T> Task<T> async(CheckedFunction0<? extends T> computation); // Always async
    
    // Execution
    Future<T> run();                           // Execute the task and return Future
    Future<T> run(Executor executor);          // Execute with specific executor
    
    // Transformation operations (lazy)
    <U> Task<U> map(Function<? super T, ? extends U> mapper);
    <U> Task<U> mapTry(CheckedFunction1<? super T, ? extends U> mapper);
    <U> Task<U> flatMap(Function<? super T, ? extends Task<? extends U>> mapper);
    
    // Error handling (lazy)
    Task<T> recover(Function<? super Throwable, ? extends T> recovery);
    Task<T> recoverWith(Function<? super Throwable, ? extends Task<? extends T>> recovery);
    
    // Filtering (lazy)
    Task<T> filter(Predicate<? super T> predicate);
    Task<T> filterTry(CheckedPredicate<? super T> predicate);
    
    // Combining tasks (lazy)
    <U> Task<Tuple2<T, U>> zip(Task<? extends U> that);
    <U, R> Task<R> zipWith(Task<? extends U> that, BiFunction<? super T, ? super U, ? extends R> mapper);
    Task<T> fallbackTo(Task<? extends T> that);
    
    // Side effects (applied when task runs)
    Task<T> peek(Consumer<? super T> action);
    Task<T> onFailure(Consumer<? super Throwable> action);
    
    // Conversion operations
    Future<T> toFuture();                      // Same as run()
    Try<T> toTry();                           // Run synchronously
    
    // Utility operations
    Task<T> timeout(long timeout, TimeUnit unit);
    Task<T> timeout(long timeout, TimeUnit unit, Supplier<? extends T> fallback);
    Task<T> delay(long delay, TimeUnit unit);  // Delay execution
}

import io.vavr.concurrent.Task;
import io.vavr.concurrent.Future;
import java.util.concurrent.TimeUnit;

// Creating Tasks (lazy - not executed yet)
Task<String> fetchData = Task.of(() -> downloadData("http://api.example.com"));
Task<Integer> compute = Task.of(() -> heavyComputation());
Task<String> immediate = Task.successful("Already computed");

// Transforming Tasks (still lazy)
Task<String> processed = compute
    .map(result -> "Result: " + result)
    .recover(ex -> "Error: " + ex.getMessage());

// Chaining Tasks (lazy composition)
Task<String> pipeline = Task.of(() -> getUserId())
    .flatMap(id -> Task.of(() -> fetchUserData(id)))
    .flatMap(data -> Task.of(() -> processUserData(data)));

// Combining multiple Tasks
Task<String> name = Task.of(() -> fetchUserName());
Task<Integer> age = Task.of(() -> fetchUserAge());
Task<String> profile = name.zipWith(age, (n, a) -> n + " (" + a + " years old)");

// Adding delays and timeouts
Task<String> delayed = fetchData.delay(5, TimeUnit.SECONDS);
Task<String> withTimeout = fetchData.timeout(10, TimeUnit.SECONDS, () -> "Timeout fallback");

// Fallback Tasks
Task<String> primary = Task.of(() -> primaryService());
Task<String> backup = Task.of(() -> backupService());
Task<String> resilient = primary.fallbackTo(backup);

// Execution (this is when computation actually starts)
Future<String> future1 = processed.run();                    // Run with default executor
Future<String> future2 = processed.run(customExecutor);      // Run with custom executor
Try<String> syncResult = processed.toTry();                  // Run synchronously

// Reusable Tasks
Task<List<String>> reusableTask = Task.of(() -> fetchDataList());
Future<List<String>> execution1 = reusableTask.run();        // First execution
Future<List<String>> execution2 = reusableTask.run();        // Second execution (independent)

// Error handling in lazy context
Task<String> safeTask = Task.of(() -> riskyOperation())
    .recover(throwable -> "Fallback value")
    .onFailure(ex -> logger.error("Task failed", ex));

// Helper methods
static String downloadData(String url) throws IOException {
    // Simulate download
    Thread.sleep(2000);
    return "Data from " + url;
}

static Integer heavyComputation() throws InterruptedException {
    Thread.sleep(5000);
    return 42;
}

/**
 * Utility methods for Future operations
 */
class Future {
    // Sequence operations - convert List<Future<T>> to Future<List<T>>
    static <T> Future<Seq<T>> sequence(Iterable<? extends Future<? extends T>> futures);
    static <T> Future<List<T>> sequence(List<? extends Future<? extends T>> futures);
    
    // Traverse operations - map and sequence combined
    static <T, U> Future<Seq<U>> traverse(Iterable<? extends T> values, 
                                          Function<? super T, ? extends Future<? extends U>> mapper);
    
    // Reduce operations
    static <T> Future<T> reduce(Iterable<? extends Future<? extends T>> futures, 
                               BinaryOperator<T> op);
    static <T> Future<Option<T>> reduceOption(Iterable<? extends Future<? extends T>> futures, 
                                             BinaryOperator<T> op);
    
    // Find operations  
    static <T> Future<Option<T>> find(Iterable<? extends Future<? extends T>> futures,
                                     Predicate<? super T> predicate);
    
    // Race operations - return first completed Future
    static <T> Future<T> firstCompletedOf(Iterable<? extends Future<? extends T>> futures);
    
    // Timing operations
    static Future<Void> delay(long delay, TimeUnit unit);
    static <T> Future<T> timeout(Future<T> future, long timeout, TimeUnit unit);
    static <T> Future<T> timeout(Future<T> future, long timeout, TimeUnit unit, 
                                Supplier<? extends T> fallback);
}

/**
 * Utility methods for Promise operations
 */
class Promise {
    // Create Promise from callback-style API
    static <T> Promise<T> fromCallback(Consumer<Consumer<T>> callbackConsumer);
    static <T> Promise<T> fromCallback(Consumer<BiConsumer<T, Throwable>> callbackConsumer);
}

import io.vavr.concurrent.Future;
import io.vavr.collection.List;
import java.util.concurrent.TimeUnit;

// Sequence operations - wait for all Futures to complete
List<Future<String>> futures = List.of(
    Future.of(() -> fetchData("url1")),
    Future.of(() -> fetchData("url2")), 
    Future.of(() -> fetchData("url3"))
);

Future<List<String>> allResults = Future.sequence(futures);
allResults.onSuccess(results -> 
    System.out.println("All completed: " + results));

// Traverse - transform and collect
List<String> urls = List.of("url1", "url2", "url3");
Future<List<String>> traverseResult = Future.traverse(urls, url -> 
    Future.of(() -> fetchData(url)));

// Race - get first completed
Future<String> fastest = Future.firstCompletedOf(futures);
fastest.onSuccess(result -> 
    System.out.println("First completed: " + result));

// Reduce - combine all results
Future<String> combined = Future.reduce(futures, (a, b) -> a + ", " + b);

// Find - get first result matching predicate
Future<Option<String>> found = Future.find(futures, 
    result -> result.contains("important"));

// Timing operations
Future<Void> delayed = Future.delay(5, TimeUnit.SECONDS);
Future<String> withDeadline = Future.timeout(
    Future.of(() -> slowOperation()),
    10, TimeUnit.SECONDS,
    () -> "Timeout occurred"
);

// Promise from callback API
Promise<String> callbackPromise = Promise.fromCallback(callback -> {
    legacyAsyncAPI("param", new LegacyCallback() {
        @Override
        public void onResult(String result) {
            callback.accept(result);
        }
        
        @Override
        public void onError(Exception error) {
            // Handle error case
        }
    });
});