Examples

List<Asyncc.AsyncTask<String, Throwable>> tasks = List.of(
  c -> exec.submit(() -> c.success(fetchA())),
  c -> exec.submit(() -> c.success(fetchB())),
  c -> exec.submit(() -> c.success(fetchC()))
);

Asyncc.Parallel(tasks, (err, results) -> {
  if (err != null) { log.error("at least one failed", err); return; }
  // results.get(0) is fetchA's value, etc.
});

import static org.ores.async.WrapErrFirst.wrap;

Asyncc.Parallel(tasks, wrap(results -> {
  reply.send(combine(results.get(0), results.get(1), results.get(2)));
}));

Asyncc.ParallelLimit(8, downloadTasks, wrap(paths -> {
  // At most 8 downloads run concurrently. The next task starts as soon
  // as one finishes. Errors short-circuit the remaining queue.
  publish(paths);
}));

Asyncc.Series(List.of(
  c -> validate(req, c),
  c -> persist(req, c),
  c -> notify(req, c)
), (err, results) -> {
  if (err != null) return; // first failure stops the chain
  // results is a List of each task's value in order.
});

Asyncc.Waterfall(List.of(
  c -> c.success(parseRequest(raw)),       // -> ParsedRequest
  (req, c) -> c.success(authorize(req)),   // -> AuthorizedRequest
  (authd, c) -> c.success(run(authd))      // -> Result
), wrap(finalValue -> {
  // finalValue is the Result from the last stage.
}));

Asyncc.Race(List.of(
  c -> exec.submit(() -> c.success(fromPrimary())),
  c -> exec.submit(() -> c.success(fromReplica()))
), wrap(winnerValue -> {
  // whichever returned first
}));

Asyncc.Map(userIds, (id, c) -> {
  exec.submit(() -> c.success(fetchProfile(id)));
}, wrap(profiles -> {
  // profiles.get(i) corresponds to userIds.get(i)
}));

Asyncc.FilterMap(candidateIds, (id, c) -> {
  exec.submit(() -> {
    var profile = fetchProfile(id);
    c.success(profile.isActive() ? profile : null);
  });
}, wrap(active -> {
  // active contains only the profiles where isActive() was true
}));

Asyncc.Reduce(transactions, BigDecimal.ZERO, (acc, txn, c) -> {
  exec.submit(() -> c.success(acc.add(txn.amount())));
}, wrap(total -> {
  // total is the final sum
}));

Asyncc.Times(8, (i, c) -> {
  exec.submit(() -> c.success(generateSample(i)));
}, wrap(samples -> {
  // samples.size() == 8
}));

Asyncc.GroupBy(users, (user, c) -> {
  c.success(user.region());
}, wrap(grouped -> {
  // grouped is Map<String, List<User>> — users keyed by region
}));

NeoQueue<Job, Void> queue = new NeoQueue<>(4); // concurrency = 4

queue.setTaskHandler((task, c) -> {
  exec.submit(() -> {
    try { processJob(task.getValue()); c.success(null); }
    catch (Throwable t) { c.fail(t); }
  });
});

queue.saturated(q -> log.warn("queue saturated; in-flight at cap"));
queue.drain(q -> log.info("queue drained"));

incoming.forEach(queue::push);

NeoLock lock = new NeoLock("inventory");

lock.acquire((err, unlock) -> {
  try {
    // critical section — safe to await async work here
    mutate(sharedState);
  } finally {
    unlock.releaseLock();
  }
});

// Or use the leak-safe sync helper (v0.2.5+):
lock.withLock(() -> mutate(sharedState));

NeoRwLock cacheLock = new NeoRwLock("config-cache");

// Reader — concurrent with other readers.
cacheLock.acquireRead((err, unlock) -> {
  try {
    return cache.get(key);
  } finally {
    unlock.releaseLock();
  }
});

// Writer — exclusive.
cacheLock.acquireWrite((err, unlock) -> {
  try {
    cache.put(key, value);
  } finally {
    unlock.releaseLock();
  }
});

// Sync helpers — auto-release even if the body throws.
cacheLock.withRead(() -> renderTemplate(cache));
cacheLock.withWrite(() -> cache.refresh());

// Non-blocking attempt — empty if waiters queued (preserves FIFO).
cacheLock.tryAcquireRead().ifPresent(u -> {
  try { /* read-only work */ } finally { u.releaseLock(); }
});

// Bounded wait — fires onError with TimeoutException on miss.
cacheLock.acquireWrite(500L, (err, unlock) -> {
  if (err instanceof TimeoutException) { backOff(); return; }
  // ... exclusive write under the timeout budget
});

Asyncc.Waterfall(List.of(
  c -> fetchPage(url, c),                            // -> String html
  (html, c) -> c.success(extractLinks(html)),        // -> List<URI>
  (links, c) -> Asyncc.Map(links, (link, inner) -> { // -> List<List<String>>
    Asyncc.Parallel(List.of(
      c2 -> exec.submit(() -> c2.success(headOk(link))),
      c2 -> exec.submit(() -> c2.success(classify(link)))
    ), inner);
  }, c)
), wrap(perLinkData -> {
  // perLinkData is List<List<String>>, ordered the same as the source links list
}));

import static org.ores.async.WrapFuture.toFuture;
import static org.ores.async.WrapFuture.fromStage;

// Return a CompletableFuture to a Spring WebFlux / Akka HTTP / gRPC boundary,
// while using async.java's combinators internally.
public CompletableFuture<String> handle(Request req) {
  return toFuture(c ->
      Asyncc.<String, Throwable>Parallel(List.of(
          cb -> exec.submit(() -> cb.success(fetchA(req))),
          cb -> exec.submit(() -> cb.success(fetchB(req)))
      ), c)
  ).thenApply(parts -> combine(parts.get(0), parts.get(1)));
}

// Consume third-party CompletionStage-returning APIs inside an async.java combinator.
Asyncc.Parallel(List.of(
    fromStage(db.queryAsync("SELECT ...")),
    fromStage(redis.getAsync(key)),
    fromStage(http.sendAsync(req))
), (err, results) -> { /* ... */ });

// Parallel: fan out N tasks, return a future of their ordered results.
CompletableFuture<List<String>> both = AsyncFut.Parallel(List.of(
    () -> CompletableFuture.supplyAsync(this::fetchA, exec),
    () -> CompletableFuture.supplyAsync(this::fetchB, exec)
));

// ParallelLimit: bounded concurrency.
CompletableFuture<List<Path>> downloaded = AsyncFut.ParallelLimit(8, downloads);

// Series: sequential.
CompletableFuture<List<Step>> chain = AsyncFut.Series(List.of(
    () -> validate(req), () -> persist(req), () -> notify(req)
));

// Race: first completer wins.
CompletableFuture<String> winner = AsyncFut.Race(List.of(
    () -> fromPrimary(),
    () -> fromReplica()
));

// Map: async transform preserving input order.
CompletableFuture<List<Profile>> profiles =
    AsyncFut.Map(userIds, id -> fetchProfileAsync(id));

// Reduce: sequential async fold.
CompletableFuture<BigDecimal> total =
    AsyncFut.Reduce(txns, BigDecimal.ZERO, (acc, t) -> computeAsync(acc, t));

// Times: N parallel iterations.
CompletableFuture<List<Sample>> samples =
    AsyncFut.Times(8, i -> generateAsync(i));

// Each: per-element fire-and-forget; future completes when all done.
CompletableFuture<Void> sent = AsyncFut.Each(users, u -> sendEmailAsync(u));

// Compose with regular CompletableFuture operators.
AsyncFut.Parallel(taskSuppliers)
    .thenApply(parts -> combine(parts))
    .thenCompose(combined -> store(combined))
    .exceptionally(err -> { log.error("pipeline failed", err); return null; });