A DispatchQueue to sequence RPC calls with gwt-dispatch

One of GWT’s greatest strengths, but one that took some getting used to for this long-time server-side developer, is that it forces an asynchronous programming model. Because the current generation of JavaScript engines are single-threaded, GWT is designed to prohibit any code from blocking the thread. GWT ensures this through extensive use of callbacks, and it’s generally a good thing as it helps keep the browser responsive to user input. This is especially important for RPC calls. If GWT were to allow synchronous RPC calls, a long-running server call would hang the browser.

One of the challenges of asynchronous programming, however, is that you do not know the order in which callbacks will be called. In some cases, this is helpful. For example, in order to minimize trips to the Datastore and do as much processing as possible on the client, I do some data joins on the client. I load the list of parent entities once and cache it (see previous post), then join various child entities on successive calls as needed. In order for this to work, I need to know that the parent entity is present in the client when the child entities are retrieved. That is, when the call to retrieve child entities completes, the parent entities must already be loaded so each child entity can look up properties of the parent entity.

There are various ways to ensure that related things happen together. One way is to use a single Action to retrieve both parent and child entities and populate a compound Result object. However, this is somewhat wasteful as the list of parent entities must be retrieved and sent on each request for related child entities. Likewise, doing the join on the server results in some parent entity data being transmitted multiple times. If you could make synchronous calls (as on the server), you would just make the calls in order: 1) retrieve parent entities (from cache if available), and 2) retrieve child entities. But we can’t make synchronous RPC calls, for good reason as explained previously.

What to do? We cannot determine the order in which callbacks will be invoked for multiple requests made at the same time. However, we can ensure that a request is not even initiated until another call has completed. The trick is to initiate the request within the onSuccess() block of the prerequisite call. We’ll call this callback chaining, and it looks like this:

	void chainCallbacks()
	{
		dispatch.execute(new PrerequisiteAction(), new AsyncCallback<PrerequisiteResult>()
		{
			@Override
			public void onFailure(Throwable caught)
			{
				// TODO Auto-generated method stub
			}

			@Override
			public void onSuccess(PrerequisiteResult result)
			{
				dispatch.execute(new DependentAction(), new AsyncCallback<DependentResult>()
				{
					@Override
					public void onFailure(Throwable caught)
					{
						// TODO Auto-generated method stub
					}

					@Override
					public void onSuccess(DependentResult result)
					{
						// Do something dependent on PrerequisiteResult 
					}
				});
			}
		});
	}

Besides the horror of nested anonymous inner classes, this approach is less than ideal because it introduces latency. The second request cannot begin until the first one has completed. Thus you lose the benefit of being able to execute many requests simultaneously on your server farm. But wait, you say, wasn’t that the whole point? Not exactly. It turns out that our real requirement is not that calls are initiated in a prescribed order, but rather just that our callbacks are invoked in a prescribed order. Is there a way to fire off all requests simultaneously (subject to browser limitations, of course) and still invoke callbacks in the order we want?

We can achieve this by wrapping each callback in another callback that is smart enough to hold the result until all callbacks are completed. I call this the DispatchQueue. Here’s the source for DispatchQueue and CallbackCommand (which is a useful abstraction to hold an Action and its callback together):

DispatchQueue.java

package com.turbomanage.gwt.client.dispatch;

import java.util.ArrayList;
import java.util.HashMap;

import net.customware.gwt.dispatch.client.DispatchAsync;
import net.customware.gwt.dispatch.shared.Action;
import net.customware.gwt.dispatch.shared.Result;

import com.google.gwt.core.client.GWT;
import com.google.gwt.user.client.rpc.AsyncCallback;

/**
 * Sequentially orders commands (service calls) by initiating all requests, then
 * invoking callbacks in order.
 * 
 * @author David Chandler
 */
public class DispatchQueue
{
	private DispatchAsync dispatch;
	private ArrayList<CallbackCommand> commands = new ArrayList<CallbackCommand>();
	private HashMap<Action<Result>, Object> outcomes = new HashMap<Action<Result>, Object>();
	private int numPending;

	/**
	 * Constructor requires a dispatcher 
	 * 
	 * @param dispatch The dispatcher that will be used to handle requests
	 */
	public DispatchQueue(CachingDispatchAsync dispatch)
	{
		this.dispatch = dispatch;
	}

	/**
	 * Convenience method to queue a command with syntax identical to execute()
	 * 
	 * @param action
	 * @param callback
	 * @return
	 */
	public <A extends Action<R>, R extends Result> DispatchQueue add(final A action,
		final AsyncCallback<R> callback)
	{
		CallbackCommand cmd = new CallbackCommand(action, callback);
		return add(cmd);
	}

	/**
	 * Add a CallbackCommand to the queue. Works by wrapping the previous
	 * command's callback with a new callback that calls this command in it
	 * onSuccess() handler
	 * 
	 * @param cmd
	 */
	public DispatchQueue add(final CallbackCommand cmd)
	{
		// Block new entries once flushing has begun
		// if (started)
		if (numPending > 0)
			throw new UnsupportedOperationException("Queue is already running");
		// Wrap with new callback that decrements the counter and saves the
		// outcome
		final AsyncCallback<Result> origCallback = cmd.getCallback();
		AsyncCallback<Result> wrapper = new AsyncCallback<Result>()
		{
			@Override
			public void onFailure(Throwable caught)
			{
				if (numPending > 0)
				{
					// Pass through to original and call next command
					saveOutcome(cmd.getAction(), caught);
					if (--numPending == 0)
						finish();
				}
				else
				{
					origCallback.onFailure(caught);
				}
			}

			@Override
			public void onSuccess(Result result)
			{
				if (numPending > 0)
				{
					// Pass through to original and call next command
					saveOutcome(cmd.getAction(), result);
					if (--numPending == 0)
						finish();
				}
				else
				{
					origCallback.onSuccess(result);
				}
			}
		};
		cmd.setCallback(wrapper);
		commands.add(cmd);
		return this;
	}

	/**
	 * Execute queued commands and clear the queue upon completion.
	 * queueCommand() has already wrapped each callback with a chaining
	 * callback, so this method just has to call the first command.
	 */
	public void flush()
	{
		// Prohibit new entries once started
		this.numPending = commands.size();
		// This starts them all asynchronously, and callback
		// wrappers save the outcomes
		for (CallbackCommand cmd : commands)
		{
			dispatch.execute(cmd.getAction(), cmd.getCallback());
		}
	}

	private void saveOutcome(Action<Result> action, Object outcome)
	{
		outcomes.put(action, outcome);
	}

	private void finish()
	{
		GWT.log("Entering DispatchQueue.finish()", null);
		// Call each callback in sequence
		for (CallbackCommand cmd : commands)
		{
			Action<Result> action = cmd.getAction();
			AsyncCallback<Result> callback = cmd.getCallback();
			Object outcome = outcomes.get(action);
			if (outcome instanceof Result)
			{
				callback.onSuccess((Result) outcome);
			}
			else
			{
				callback.onFailure((Throwable) outcome);
			}
		}
	}
}

CallbackCommand.java

package com.turbomanage.gwt.client.dispatch;

import net.customware.gwt.dispatch.shared.Action;
import net.customware.gwt.dispatch.shared.Result;

import com.google.gwt.user.client.rpc.AsyncCallback;

/**
 * Wraps an action and callback so that they can be added to a DispatchQueue
 * 
 * @author David Chandler
 */
public class CallbackCommand
{
	private Action<Result> action;
	private AsyncCallback<Result> callback;

	@SuppressWarnings("unchecked")
	public CallbackCommand(final Action action, final AsyncCallback callback)
	{
		this.action = action;
		this.callback = callback;
	}

	public Action<Result> getAction()
	{
		return action;
	}

	public AsyncCallback<Result> getCallback()
	{
		return callback;
	}

	public void setCallback(AsyncCallback<Result> callback)
	{
		this.callback = callback;
	}
}

To use DispatchQueue, just create a new queue, add your actions and corresponding callbacks in order, and flush the queue:

	public void doInOrder()
	{
		new DispatchQueue(dispatch)
			.add(new PrerequisiteAction(), new AsyncCallback<PrerequisiteResult>()
			{
				@Override
				public void onFailure(Throwable caught)
				{
					// TODO Auto-generated method stub
				}

				@Override
				public void onSuccess(PrerequisiteResult result)
				{
					// Store results in model
				}
			})
			.add(new DependentAction(), new AsyncCallback<DependentResult>()
			{
				@Override
				public void onFailure(Throwable caught)
				{
					// TODO Auto-generated method stub
				}

				@Override
				public void onSuccess(DependentResult result)
				{
					// Do something that depends on prerequisite results
				}
			})
			// Do it
			.flush();
	}

DispatchQueue is probably one of the most fun pieces of code I’ve ever written. It combines the benefits of asynchronous requests with the determinism of synchronous calls. Used in conjunction with the caching dispatcher from the previous post, it is extremely powerful. Each time you need multiple call(back)s to be executed in order, just queue them up and go. The caching dispatcher returns each result from cache if available, and the dispatch queue ensures that callbacks are called in order . You’ll also find it helpful to store commonly used Action/callback pairs as CallbackCommands. For example, a service that always needs prerequisite data can store the CallbackCommand as a class field and add it to each queue as needed with DispatchQueue.add(CallbackCommand cmd).

Note that in this implementation of DispatchQueue, if any call fails, the onFailure() method for all callbacks will be called, passing the exception from the original failure. I’m not sure this is always the best behavior, but you’re welcome to change it 🙂

Enjoy!