Future Composition in Vert.x

For those who don't know, Vert.x is an event driven and non blocking application toolkit.
It's polyglot, so you can use it with different languages (as Java, Kotlin, JavaScript, Groovy, Ruby or Scala).

What does "non blocking" mean?

In synchronous programming, when a function is called, the caller has to wait until the result is returned.
This behaviour could lead to performance issues.

Often the "obvious solution" seems to be concurrent programming, but dealing with shared resources and threads is not easy and deadlocks are around the corner.

Explaining how Vert.x guarantees asynchrony through the event loop concept is not in the scope of this article, but anything you may want to know gets unraveled in the great Gentle guide to async programming with Vert.x.

In a "non blocking" world, when the result of a function can be provided immediately, it will be. Otherwise a handler is provided to handle it when it will be ready.

asyncFunctionCall(result -> {
    doSometingWith(result);
})

Usualy a big "but" gets raised the first time an async piece of code is seen: But... I need that result now!

This kind of programming requires a mindset change: it's necessary to know and understand the main patterns.

Futures vs Callbacks

There are two ways to deal with async calls in Vert.x.

• Pass a callback that will be executed on call completion.
• Handle a future returned from the function/method call.

Callback

asyncComputation(asyncResult -> {
  if (asyncResult.succeeded()) {
    asyncResult.result()
    // do stuff with result
  } else {
    // handle failure
  }
})

A callback is a function passed to an async method used to handle the result of its computation.
It's simple to implement but it brings some drawbacks:

• Unit testing becomes not-so-immediate
• Leads to the dreaded Callback Hell
• There's more code to be written.

Future

Future future = asyncComputation()

future.onSuccess(result -> {
  // do stuff
})

future.onFailure(cause -> {
  // handle failure
})

To avoid the problems listed for the "callback way", Vert.x implements a concept called Future.

A future is an object that represents the result of an action that may, or may not, have occurred yet (cf. apidoc).

How to switch from a callback-like call to a future one

Consider the callback example shown above.
We want a Future object to take advantage of the patterns described below, but asyncComputation is a function from another library, so we cannot modify it.

We can use a Promise.
According to the apidoc, it represents the writable side of an action that may, or may not, have occurred yet.
It perfectly fits our needs:

Promise promise = Promise.promise();
asyncComputation(asyncResult -> {
  if (asyncResult.succeeded()) {
    promise.complete(asyncResult.result());
  } else {
    promise.fail(asyncResult.cause());
  }
})
return promise.future()

That's it. We transformed a callback into a future.
The Promise API's give us a way to make this code more readable:

Promise promise = Promise.promise();
asyncComputation(promise::handle)
return promise.future()

The handle method takes care of the completion or failure of the promise, given an async result.

Future patterns

The Future object implements some interesting patterns that smartly help resolving async issues:

• Map
• Compose
• Recover

Future Mapping

For those of you who know about the map function, part of the java Stream API, this feature should be immediate to understand.

The Future's map function accepts a function that transforms the result from one type to another.

asyncComputation() // returns a Future<String>
  .map(Integer::valueOf) // returns a Future<Integer>
  .onSuccess(...)
  .onFailure(...)

Future Composition

The compose method is similar to map, but is used when the mapping is an async operation itself:

asyncComputation()
  .map(Integer::valueOf)
  .compose(id -> retrieveDataById(id)) // retrieveDataById returns a Future
  .onSuccess(...)
  .onFailure(...)

Future Recovery

Futures can succeed, but can also fail.
To handle a failed future and consider a different behaviour, recover function can be used:

asyncComputation()
  .map(Integer::valueOf)
  .recover(cause -> Future.succeededFuture(0)) // when Integer::valueOf fails, the future could be recovered with a default value
  .compose(id -> retrieveDataById(id))
  .onSuccess(...)
  .onFailure(...)

Concurrent composition

To handle multiple future results at the same time, CompositeFuture is the class needed, it provides two static factory methods:

all returns a future that succeeds if all the futures passed as parameters succeed, and fails if at least one fails.

CompositeFuture.all(futureOne, futureTwo)
  .onSuccess(compositeResult ->
    // all futures succeeded
  )
  .onFailure(cause -> 
    // at least one failed
  );

any returns a future that succeeds if any one of the futures passed as parameters succeed, and fails if all fail.

CompositeFuture.any(futureOne, futureTwo)
  .onSuccess(compositeResult ->
    // at least one succeed
  )
  .onFailure(cause -> 
    // all failed
  );

Conclusions

Future composition API in Vert.x represents a solid way to write simple and affordable async code.
Always remember:

• never throw exceptions into async code, use failed future instead to handle failure behaviours.
• at the end of a future composition, don't forget to handle future's successes (onSuccess) and failures (onFailure)

Comments

[Nezam uddin]

Very good explanation for a vertx newbie like me. Please write more on vertx.