Asynchronous API

spinach is built on netty that is a multithreaded, event-driven IO framework. All communication is handled asynchronously. Every command sent to Disque using the asynchronous API creates a Future that can be cancelled, awaited and listened. You obtain the async API by calling the connect().async() method of the DisqueClient.

There are several ways how to wait or get notified in case a future completes:

1. RedisFuture.get() / RedisFuture.get(long timeout, TimeUnit unit)
2. while(!RedisFuture.isDone())
3. RedisFuture.await(long timeout, TimeUnit unit)
4. RedisFuture.thenAccept(Consumer<? super T> action)

Futures of spinach also carry exceptions, if any occurred. If you call the get() method and an exception occurred, this exception will be rethrown wrapped within an ExecutionException.

Synchronization using RedisFuture.get()

You can fire this way one or more commands and wait until the execution completes. If you use RedisFuture.get() the system will wait indefinitely and block your call. RedisFuture.get(long timeout, TimeUnit unit) will wait at most until the specified timeout. If the result comes back earlier, so you call will also continue earlier. If the timeout exceeds, you will receive a TimeoutException.

Synchronization using isDone()

In this style, you are using the Future whether it's done. You can poll, wait or even do other things until your future comes back. isDone is a non-blocking call.

Waiting using await(long timeout, TimeUnit unit)

You can specify a maximal waiting time, like in the get(long timeout, TimeUnit unit) example. The difference to get() is, that await(long timeout, TimeUnit unit) will come back with a boolean value in every case. So if the execution takes longer than the timeout, await will return false instead of a TimeoutException. After a result of true you can call get() and you'll receive the return value immediately.

Exceptions

Any Disque errors will cause to throw an ExecutionException on calling get().

Examples

Basic operations

DisqueAsyncCommands<String, String> async = client.connect().async();
RedisFuture<String> addjob = async.addjob("queue", "body", 5, TimeUnit.SECONDS);
RedisFuture<KeyScanCursor<String>> scan = async.qscan();

addjob.get() == Job Id
scan.get() == Cursor with Job Id's

Waiting for a future with a timeout

DisqueAsyncCommands<String, String> async = client.connect().async()
RedisFuture<String> addjob = async.addjob("queue", "body", 5, TimeUnit.SECONDS);
RedisFuture<KeyScanCursor<String>> scan = async.qscan();

addjob.await(1, SECONDS) == true
addjob.get() == Job Id
scan.get(1, TimeUnit.MINUTES) == Cursor with Job Id's

Using listeners for a future

DisqueAsyncCommands<String, String> async = client.connect().async()
RedisFuture<String> addjob = async.addjob("queue", "body", 5, TimeUnit.SECONDS);

Runnable listener = new Runnable() {
    @Override
    public void run() {
            ...;
    }
};

addjob.addListener(listener, MoreExecutors.sameThreadExecutor());