Guidance on using a SynchronousQueue

Question

I need to perform a series of http requests, each of which may depend on a previous http response. I have been able to achieve this using an AsyncTask "tree" of sorts, but as the decision tree grows, the AsyncTask technique grows more unwieldy.

I think that somehow using a SynchronousQueue (or other type of queue) is the best approach, but I can't seem to find any good guidance or tutorials on how to use a Queue for something like http requests.

Can anyone provide any guidance or point to any good tutorials on using SynchronousQueue or suggest the best kind of Queue?

Answer 1

Use a java.util.concurrent.SingleThreadExecutor and make a Runnable out of each HTTP operation and result-handler. You can submit subsequent tasks to it as you determine whether you need to continue progress.

For example, the HTTP "task" would run and submit the Result "task" on success, or the Error "task" on failure. The Result task would in-turn submit another HTTP task when it was done processing. Using SingleThreadExecutor ensures only one task runs at-a-time.

You could use a ThreadPoolExecutor if you can handle multiple operations in-flight at once.

Take all that, and wrap it in an AsyncTask that manages the top-level "kick-off" and waits for everything to complete. It would probably be useful to have a ConditionVariable or something to synchronize the "end" signal (using a Done "task") so you can safely tear down the Executor.

A SynchronousQueue doesn't do anything helpful for you here, because it leaves you to do all the tread management. If you use an Executor that is all handled and all you deal with is Runnables and Futures. That's probably why you are not finding any tutorials. Anyway, the Executors all use one of those queue implementations underneath!

As requested, here is some skeleton Java code. Unsupported untested as-is. This should get you started. You can use a different synchronization object if you don't like ConditionVariable.

This is a generic technique, not specific to Android, feel free to use it in other contexts.

This functions as a State Machine, with HttpTask et al forming the states, and the transitions are hard-coded by submitting the Next State to the ExecutorService. There's even a "Big Bang at the end, so everyone knows when to clap" in the form of the ConditionVariable.

Some may consider DoneTask and FailedTask overkill, but it keeps the Next State mechanism consistent, and lets Future<? extends ResultTask> function as a somewhat type-safe container for the results, and certainly keeps you from mis-assigning to it.

abstract class BasicTask {
    final ExecutorService es;
    final ConditionVariable cv;
    public BasicTask(ExecutorService es, ConditionVariable cv) {
        this.es = es;
        this.cv = cv;
    }
}
abstract class HttpTask extends BasicTask {
// source omitted.
// you should make a class to prepare e.g. Apache HTTP resources for specific tasks (see below).
}
abstract class ResultTask implements Runnable {
    final ConditionVariable cv;
    public ResultTask(ConditionVariable cv) {
        this.cv = cv;
    }
    public void run() {
        cv.open();
    }
}
final class FailedTask extends ResultTask {
    final Exception ex;
    public FailedTask(ConditionVariable cv, Exception ex) {
        super(cv);
        this.ex = ex;
    }
    public Exception getError() { return ex; }
}
final class DoneTask<T> extends ResultTask {
    final T results;
    public DoneTask(ConditionVariable cv, T results) {
        super(cv);
        this.results = results;
    }
    public T getResults() { return results; }
}

class HttpSequence extends AsyncTask<Void,Void,Object> {
    // this will capture the ending task
    Future<? extends ResultTask> result;
    // this is an inner class, in order to set Result. Refactor so these are small.
    // if you don't like inner classes, you still need to arrange for capturing the "answer"
    final class SomeHttpTask extends HttpTask implements Runnable {
        public void run() {
            try {
                final SomeType thisStep = doTheStuff(lastStep);
                if(thisStep.isDone()) {
                    // we are done here
                    result = es.submit(new DoneTask<SomeType>(cv, thisStep));
                }
                else if(thisStep.isFailed()) {
                    // not done: we can't proceed because of something in the response
                    throw thisStep.getError();
                }
                else {
                    // not done, everything is ok for next step
                    es.submit(new NextHttpTask(es, cv, thisStep));
                }
            }
            catch(Exception ex) {
                result = es.submit(new FailedTask(cv, ex));
            }
        }
    }
    final class TheFirstTask extends HttpTask implements Runnable {
    // source omitted.  just emphasizing you need one of these for each "step".
    // if you don't need to set Result, this could be a static inner class.
    }

@Override
public Object doInBackground(Void...) {
    final ExecutorService es = Executors.newSingleThreadExecutor();
    final ConditionVariable cv = new ConditionVariable(false);
    try {
        es.submit(new TheFirstTask(es, cv));
        // you can choose not to timeout at this level and simply block until something happens...
        final boolean done = cv.block(timeout);
        if(!done) {
            // you will need to account for unfinished threads, see finally section!
            return new IllegalStateException("timed out waiting on completion!");
        }
        if(result != null) {
            final ResultTask done = result.get();
            if(done instanceof DoneTask) {
                // pass SomeType to onPostExecute()
                return ((DoneTask<SomeTYpe>)done).getResults();
            }
            else if(done instanceof FailedTask) {
                // pass Exception to onPostExecute()
                return ((FailedTask)done).getError();
            }
            else {
                // something bad happened, pass it to onPostExecute()
                return new IllegalStateException("something unexpected signalled CV!");
            }
        }
        else {
            // something bad happened, pass it to onPostExecute()
            return new IllegalStateException("something signalled CV without setting result!");
        }
    }
    catch(Exception ex) {
        // something outside workflow failed, pass it to onPostExecute()
        return ex;
    }
    finally {
        // naive shutdown (doesn't interrupt running tasks): read JavaDoc on ExecutorService for details
        es.shutdown();
    }
}
@Override
public void onPostExecute(Object result) {
    if(result instanceof SomeType) {
        // success UI
    }
    else if(result instanceof Exception) {
        // error UI
    }
}
}

Answer 2

I can't say for sure without knowing the details of your use case, but you probably want to avoid the SynchronousQueue, as it will block the thread putting things into the queue until the listener thread takes it back out of the queue. If you were putting things in using the UI thread you'd be locking up the UI.

I think a BlockingQueue may suit your needs. The JavaDoc has a good producer-consumer example.