How to implement asynchronous queue?

concurrency, java, java.util.concurrent

Solution

Your `AsyncQueue` is very similar to a `BlockingQueue` such as `ArrayBlockingQueue`. The `Future` returned would simply delegate to the `ArrayBlockingQueue` methods. `Future.get` would call `blockingQueue.poll` for instance.

As for your update, I'm assuming the thread that calls `add` should invoke the callback if there's one waiting? If so it's a simple task of creating one queue for elements, and one queue for callbacks.

- Upon add, check if there's a callback waiting, then call it, otherwise put the element on the element queue

- Upon poll, check if there's an element waiting, then call the callback with that element, otherwise put the callback on the callback queue

Code outline:

class AsyncQueue<E> {

    Queue<Consumer<E>> callbackQueue = new LinkedList<>();
    Queue<E> elementQueue = new LinkedList<>();

    public synchronized void add(E e) {
        if (callbackQueue.size() > 0)
            callbackQueue.remove().accept(e);
        else
            elementQueue.offer(e);
    }

    public synchronized void poll(Consumer<E> c) {
        if (elementQueue.size() > 0)
            c.accept(elementQueue.remove());
        else
            callbackQueue.offer(c);
    }
}

Problem

Given following variation of queue: ``` interface AsyncQueue<T> { //add new element to the queue void add(T elem); //request single element from the queue via callback //callback will be called once for single polled element when it is available //so, to request multiple elements, poll() must be called multiple times with (possibly) different callbacks void poll(Consumer<T> callback); } ``` I found out i do not know how to implement it using java.util.concurrent primitives! So questions are: - What is the right way to implement it using java.util.concurrent package? - Is it possible to do this w/o using additional thread pool?

Original source