Package org.ores.async

Class NeoQueue<T,V>

java.lang.Object
org.ores.async.NeoQueue<T,V>
Type Parameters:
T - task input type
V - task result type
public class NeoQueue<T,V> extends Object
A bounded async work queue.

Hand the queue a task handler; push tasks; the queue dispatches them via the shared executor, respecting a concurrency cap. Callers can subscribe to saturated, unsaturated, and drain lifecycle events for back-pressure modelling.

Why a queue (and not just Asyncc.ParallelLimit(int, java.util.Map<java.lang.Object, org.ores.async.Asyncc.AsyncTask<T, E>>, org.ores.async.Asyncc.IAsyncCallback<java.util.Map<java.lang.Object, T>, E>))?

ParallelLimit takes a fixed list of tasks. A NeoQueue accepts streaming arrivals — you push tasks as they show up and the queue keeps the concurrency cap honoured. Useful for incoming WS frames, paged downstream calls, file watcher events.

Usage

The handler's continuation parameter is conventionally named c (for continuation). Fire it via c.success(value), c.fail(error), or the canonical c.done(err, value). 

   NeoQueue<JobSpec, JobResult> queue = new NeoQueue<>(4); // concurrency = 4

   queue.setTaskHandler((task, c) -> {
       try {
           c.success(processJob(task.getValue()));
       } 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(spec -> queue.push(spec));

With virtual threads

The queue ships with a daemon-threaded default executor. For production deployments on JDK 21+, swap it for a VT executor once at startup: 

   NeoQueue.setExecutor(Executors.newVirtualThreadPerTaskExecutor());