DEV Community: CherryChain

Vert.x Circuit Breaker

Stanislav Petrosyan — Fri, 28 May 2021 14:56:37 +0000

Scenario

We can imagine a scenario where we have a restaurant. We have the kitchen service and the orders service that communicate with each other. When a new order arrives, this will be sent to the kitchen, which gets to work and prepares the dishes. When the dishes are ready, they are delivered to the customer. But if we suppose that a really large number of customers come, so consequently a large number of orders, the kitchen gets overloaded by all these orders and therefore it cannot prepare the dishes at the same rate as the orders arrive. At this point the customers are going to wait much time before receiving their dishes because the kitchen is using all the resources for the previous orders.
So we have a deadlock. To make the situation worse, we can immagine that the chef, due to the excessive work, has cut themself and for some time they cannot work.
The scenario just described is defined as cascading failure.

A definition that we can give of a cascading failure is: a process in a system of interconnected parts in which the failure of one or few parts can trigger the failure of other parts and so on. Such a failure may happen in many types of systems, including power transmission, computer networking, finance, transportation systems, organisms, the human body, and ecosystems.

Explore the pattern

The Circuit Breaker Pattern is one solution for the cascading failure. This pattern can prevent an application from repeatedly trying to execute an operation that is likely to fail. Allowing it to continue without waiting for the fault to be fixed. The Circuit Breaker pattern also enables an application to detect whether the fault has been resolved. If the problem appears to have been fixed, the application can try to invoke the operation.

In the previous scenario the circuit breaker is applied to the orders service: every time a new customer comes, the orders service asks the kitchen if it can prepare the dish in a short time. If the kitchen's response is negative, the orders service consider it as a failure and so it understands that the kitchen service is broken and advises the new customer to have a drink at the bar service in order to not overload the kitchen's one.

So a circuit breaker acts as a proxy for operations that might fail. The proxy should monitor the number of recent failures that have occurred and use this information to decide whether to allow the operation to proceed, or simply return an exception immediately.

In this article we see how to use an implentation of this pattern: Vertx Circuit Breaker.

Setup

To use the Vert.x Circuit Breaker, add the following dependency to the dependencies section of your build descriptor:

Maven (in your pom.xml)

<dependency>
 <groupId>io.vertx</groupId>
 <artifactId>vertx-circuit-breaker</artifactId>
 <version>4.0.3</version>
</dependency>

Gradle (in your build.gradle file):

compile 'io.vertx:vertx-circuit-breaker:4.0.3'

So now you can use a circuit breaker in your java/kotlin file, here is an example in kotlin:

val breaker = CircuitBreaker.create("test", vertx, CircuitBreakerOptions()
  .setMaxFailures(2)
  .setTimeout(1000)
  .setMaxRetries(10)
  .setResetTimeout(10000)
)

Don’t recreate a circuit breaker on every call. A circuit breaker is a stateful entity. It is recommended to store the circuit breaker instance in a class field.

Inside CircuitBreakerOptions we can set some settings:

setMaxFailures: number of failures before opening the circuit
setTimeout: consider a failure if the operation does not succeed in time
setMaxRetries: number of retries if the operation fails
setResetTimeout: time to wait for switch from OPEN to HALF OPEN state

For executing code with the circuit breaker we just use execute function.
The executed block receives a Future object as parameter, to denote the success or failure of the operation as well as the result.

// we set promise type to String
breaker.execute<String> { promise ->
  stuff()
  // some code executing with the breaker
  // the code reports failures or success on the given promise.
}.onComplete {
  // Get the operation result.
  println("stuff done") 
}

Let's see how it works this circuit breaker.

State Machine

The circuit breaker uses a state machine for registering all events and behave accordingly.
Below we can see an image of this state machine.

Circuit Breaker goes into the OPEN state when the service is failed
Circuit Breaker goes into the CLOSE state when the service it works correctly
Circuit Breaker goes into the HALF OPEN state when the reset timeout is expired

Open state

When a circuit is in the open state, it means that no code inside an execute block will be ran for the same instance.
The circuit opens when the maxFailures property is satisfied.

Let's see an example:

circuitBreaker.execute<String> { promise ->
  promise.fail("error")
}.onComplete { println(it) }

So we can declare a failure when an execution block fails independently of how many calls fail inside that block. In this case the count of failure will be set to 1.

Close state

When a circuit is in the closed state, it means that the number of failures is less than the maximum. In this state all code will be executed inside the execute block.

In this state it is possible to define the retry number (retries are available just when the circuit is in the closed state). This number specifies how many times the circuit may execute the code before failing. So for example, if we have setMaxRetries to 2, the operation may be called 3 times: the initial attempt plus 2 retries. So at the end of the third call that executed block fails and will increment the number of failures by 1.

Retries can be useful when the answer is not deterministic. In fact, we can take advantage of our scenario by imagining that the order service asks if the dish is ready in the kitchen. This request has an nondeterministic answer as, depending on the moment, different answers can be received. In fact, if the kitchen is busy and the dish is not ready, you may not receive a response, thus sending the request to timeout and creating a failure. Otherwise, if the kitchen has finished, the answer will be successful so the dish will be delivered to the customer. We don't want the first situation to happen because the circuit breaker would immediately count the failure, instead we want to make several calls because we assume that the resource we want takes a longer time to be ready. So we set up retries for this very reason. If in the end the resource is still not available then we will be sure that there is a problem and we will activate the circuit breaker.

By default the timeout between retries is set to 0 which means that retries will be executed one after another without any delay. This, however, will result in increased load on the called service and may delay its recovery. In order to mitigate this problem it is recommended to execute retries with delays.
The retryPolicy method can be used to specify retry policy. Retry policy is a function which receives retry count as a single argument and returns the timeout in milliseconds before retry is executed and allows to implement a more complex policy. Below is an example of retry timeout which grows linearly with each retry count:

val circuitBreaker = CircuitBreaker.create("my-circuit-breaker", vertx, CircuitBreakerOptions()
  .setMaxFailures(5)
  .setMaxRetries(5)
  .setTimeout(2000)
).retryPolicy(retryCount -> retryCount * 100L)

Half open state

When a circuit is in the open state, every call that you want to execute will fail, without executing really the code. At this point, after some time (called ResetTimeout) which by default is set to 30000 ms, the circuit will be set to half open state. In this state the next execution of protected code will define the next state of circuit. If execution succeeds then the state will be set to closed otherwise it will be set to open.

Event Bus

Every time that the state of circuit changes, an event is published on the event bus. The address of the bus can be configured using setNotificationAddress. If null is passed to this method, the notifications will be disabled. By default, the used address is vertx.circuit-breaker.

Each message contains at least:

state
name
failures

For doing this you just need few row of code:

vertx.eventBus().localConsumer<JsonObject>("vertx.circuit-breaker") {
  println(it.body())
}

Handlers

It is possible to configure some callbacks when circuit changes the state. Here's an example:

val breaker = CircuitBreaker.create("my-circuit-breaker", vertx, CircuitBreakerOptions()
  .setMaxFailures(5)
  .setTimeout(2000)
).openHandler {
  println("Circuit opened")
}.closeHandler {
  println("Circuit closed")
}.halfOpenHandler {
  println("Circuit half-opened")
}

Conclusion

In conclusion this pattern is great when you have a system with a large number of service calls which are used to manage a large number of resources.

So as a last tip we recommend to use this pattern to prevent an application from trying to invoke a remote service or access a shared resource if this operation is highly likely to fail.

Don't use this pattern as a replacement for exception handling in your application business logic.

Now you can implement circuit breaker in your project. :)

JavaFX, JLink and JPackage

Luca Guadagnini — Thu, 25 Mar 2021 16:57:25 +0000

A few days ago, for the Java User Group Trentino-Alto Adige-Südtirol I shown some experiments on JavaFx, JLink and JPackage. The purpose of the talk was to show the chance to use JavaFX working with new JDK tools such as JLink (since OpenJDK 9) and JPackage (provided by the new OpenJDK 16) for rich client development.
Although some may state «JavaFX is dead» since it's not part of the OpenJDK anymore, JavaFX is actually alive and kicking. OpenJFX (aka Open JavaFX) is a side project still maintained and evolved by Oracle, Gluon, BellSoft and others. Suddenly, because of the advent of OpenJDK, Adoptium and other free OpenJDK builds, it may not be that straightforward to understand how to get started with JavaFX, especially if you still working on an Oracle Jave SE 1.8.0 distribution. Let's try to get there.

Before starting

To grasp the example, you need:

Git
Apache Maven
Sdkman
a Linux shell or a Wsl2 shell for Windows (sorry I know nothing about MacOS)
an editor for the pom file (even Vim is fine, but I don't remember how to exit from it, so don't ask me).

DISCLAIMER
Keep in mind that I will write regarding module-info and Java Platform Module System, but I won't explain too much about them.
The goal of this post is not about how you have to organize a project, but how to make the tools work for you. I will instead write something more about JPMS in a future post (it is quite intriguing and intricated).

What is JavaFX?

JavaFX is a framework of graphics and media packages that enables developers to create and deploy rich client applications.

Example

Let's take a basic JavaFX example, for instance, this one:

https://gitlab.com/lucaguada/treefx.

The source code is pretty old and is a JavaFX 2.1 example taken from here: Oracle Docs.

I configured the pom.xml as simple as possible, or at least this was my intention. I just set two plugins: maven-compiler-plugin to avoid compilation issues when we advance the Java version during our tests and exec-maven-plugin for being able to run our application from the terminal.

Oracle Java SE to OpenJDK

As I mentioned before, JavaFX is not part of the OpenJDK anymore. Let's see if such a boomer like me is correct.

Download the tar.gz file of Oracle Java SE 1.8.0 from here: Oracle Java and don't worry about the license, the FAQ explicitly states this:



Oracle Java SE8 updates, which includes the Oracle JRE with Java Web Start, continue to be free for personal use, development, testing, prototyping, demonstrating and some other important uses explained in this FAQ under the OTN License Agreement for Java SE. Personal users can continue downloading the Oracle Java SE 8 JRE at java.com.

Unzip the package in the .sdkman/candidates/java folder. You should be able to check the OracleJDK by typing:



 $ sdk list java

Sdkman should list the Oracle Java SE as follow:



Unclassified  |     | 8.0.281      | none    | local only | 8.0.281-oracle

Nice! Therefore we can now compile our project:



 $ sdk use java 8.0.281-oracle
Using java version 8.0.281-oracle in this shell.
 $ mvn clean compile exec:java

and if everything is fine:

Nice twice! However, we don't want to have a headache for Oracle licenses/subscriptions/etc., so we now change the OracleJDK with an Open one and compile again.



 $ sdk install java 8.0.272.hs-adpt
... install process by sdkman ...
 $ sdk use java 8.0.272.hs-adpt
Using java version 8.0.272.hs-adpt in this shell.
 $ mvn clean compile exec:java

and tragically:



[ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.
[ERROR] Re-run Maven using the -X switch to enable full debug logging.
[ERROR] 
[ERROR] For more information about the errors and possible solutions, please read the following articles:
[ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoFailureException

Errors, errors everywhere! TreeFX can't find any package of the framework JavaFX! How to solve this? JavaFX is now a set of self-contained dependencies, so let's modify the pom.xml file by defining the right dependencies.



<dependencies>
  <dependency>
    <groupId>org.openjfx</groupId>
    <artifactId>javafx-controls</artifactId>
    <version>16</version>
  </dependency>
  <dependency>
    <groupId>org.openjfx</groupId>
    <artifactId>javafx-media</artifactId>
    <version>16</version>
  </dependency>
</dependencies>

and then compile again. But... again...



[ERROR] Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.8.1:compile (default-compile) on project treefx: Compilation failure
[ERROR] /home/.../treefx/src/main/java/io/trydent/treefx/Flower.java:[7,20] cannot access javafx.scene.Group
[ERROR]   bad class file: /home/.../.m2/repository/org/openjfx/javafx-graphics/16/javafx-graphics-16-linux.jar(javafx/scene/Group.class)
[ERROR]     class file has wrong version 55.0, should be 52.0
[ERROR]     Please remove or make sure it appears in the correct subdirectory of the classpath.

No more errors about missing packages, but class file version! What does it mean? Every time the Java Compiler compiles your code, it annotates your classes with a number to identify the target JDK release. In this case, the class version 55.0 targets JDK 11, but we're using JDK 8, therefore the class version number should be at a maximum of 52.0.

Unfortunately, there is no way to downgrade JavaFX to a version compatible with JDK 8. Sure, as someone already suggested to me, we can download Liberica OpenJDK 8 with JavaFX (or the Azul one) however, we will be forever stuck to JavaFX 8. Being locked to JavaFX 8 doesn't allow us to benefit from having bug fixes and other kinds of improvements.

It's time to upgrade our OpenJDK to 11.

OpenJDK 11

Once we set our JavaFX dependencies in the pom.xml, we can try to switch to an OpenJDK 11:



 $ sdk use java 11.0.10.hs-adpt
Using java version 11.0.10.hs-adpt in this shell.
 $ mvn clean compile exec:java

And everything will be fine. No more errors about missing packages and class versions.

Because our application is ready, we can package it into a self-executable-jar and release it to our customers/friends/pets (well, if they are all interested in Tree animations, of course). By doing so, we face an old-fashioned issue: we depend on the customer/friend/pet's installed JDK. We can't upgrade our development JDK (let's think about new API's and runtime improvements, etc.) until our customers don't upgrade their own. Yes, we could package the application with a JRE, but this would lead to having a pretty big sized ZIP for just one animation.

JLink, the Java Linker

As explained in JEP 282, JLink is a tool that can assemble modules and their dependencies into a custom Java runtime image. Since the JRE, JDK and some libraries/frameworks are now restructured in modules, we can create a custom JRE image for our wonderful JavaFX application!

JavaFX is indeed modularized, so we can try!

Before starting, though, I have to inform you that I skipped the official Maven JLink Plugin usage. In a draft of this article, I explained how to work with the JLink plugin and JavaFX, but I realized it was far too long and not very useful in practice (for this example, at least). Instead of the Maven JLink Plugin, we'll use the official JavaFX JLink Plugin provided by the JavaFX development team to simplify the process.

Let us modify the pom.xml and append a Maven plugin for JavaFX and JLink:



<plugin>
  <groupId>org.openjfx</groupId>
  <artifactId>javafx-maven-plugin</artifactId>
  <version>0.0.5</version>
  <configuration>
    <compress>2</compress>
    <noHeaderFiles>true</noHeaderFiles>
    <stripDebug>true</stripDebug>
    <noManPages>true</noManPages>
    <launcher>treefx</launcher>
    <mainClass>treefx/com.acme.treefx.TreeFX</mainClass>
    <jlinkImageName>treefx</jlinkImageName>
    <jlinkZipName>treefx</jlinkZipName>
  </configuration>
</plugin>

The configuration is not quite self-explanatory, here some hints:

compress enables compression of resources, we set to 2 for compressing to ZIP
noHeaderFiles excludes C header files for supporting native-code programming (we don't need that for our custom JRE)
stripDebug excludes JRE debug information (yeah we don't need this too)
noManPages excludes the JDK docs
jlinkImageName the runtime final name
jlinkZipName the ZIP final name
launcher the executable name
mainClass what main needs to be launched by specifying module, package and class.

We set all such exclusions for shrinking our custom JRE to a smaller size than the usual one.
However, if we try the new plugin command:



 $ mvn clean compile javafx:jlink

Something is still missing:



[ERROR] Failed to execute goal org.openjfx:javafx-maven-plugin:0.0.5:jlink (default-cli) on project treefx: Error: jlink requires a module descriptor

As I said before if JDK, JRE and JavaFX are now modularized, this implies that we need to modularize our application too. Create a new file named module-info.java in the folder src/main/java:



module treefx {
  requires javafx.controls;
  requires javafx.media;

  opens com.acme.treefx to javafx.graphics;
}

The file defines our application as a module named treefx and our module requires some modules as well in order to work, javafx.controls and javafx.media.
I don't want to explain too much about it, but keep in mind that modules are a new way to organize your applications and libraries and enable strong encapsulation and implementation/reflection hiding. Knowing this, what about that last statement opens in module-info?
There are some cases where you have to enable reflection (because of modules, reflection is inaccessible by default) for your classes, because of frameworks for instance. In our case we have to enable the reflection of our package com.acme.treefx to the module javafx.graphics to let JavaFX work.

Then once again:



 $ mvn clean compile javafx:jlink

... finally ...



[INFO] Building zip: /home/.../treefx/target/treefx.zip

Yeeeah our application is ready to be propagated to Knowhere (cit.)!

Let's take a look at what actually happened in the target folder:



 $ tree target/treefx



target/treefx
├── bin
│   ├── java
│   ├── keytool
│   └── treefx
├── conf
│   ├── [...]
├── legal
│   ├── [...]
├── lib
│   ├── [...]
└── release

What do we have in the generated treefx.zip? As we can see, it's just a JRE with our application encapsulated inside. In folder bin, we can find the launcher treefx.

And the size of the ZIP file? Just 47Mb! The best part? If our application only needs the modules we declared in module-info, the size will increase only if our application grows in terms of compiled classes.

Nice trice, right? But we want more! For the sake of aesthetics, we prefer to distribute our application with an official deb package how can we do this?

It's time to upgrade our OpenJDK to 16!

OpenJDK 16 and JPackage

OpenJDK 16 has been released on 16th March. However, this is not the right post to talk about all the new shining features and tools, but it is the right post to talk about a new particular tool: JPackage.

JPackage was introduced as an incubating tool from JDK 14 to JDK 15 and is now production-ready for packaging self-contained Java applications.

As above we want to rely on Maven and fortunately, we got a well-packed plugin (JPackage Plugin) on Maven Central that we can add:



<plugin>
  <groupId>org.panteleyev</groupId>
  <artifactId>jpackage-maven-plugin</artifactId>
  <version>1.4.0</version>
  <configuration>
    <name>TreeFX</name>
    <appVersion>1.0.0</appVersion>
    <vendor>com.acme</vendor>
    <destination>target/dist</destination>
    <module>treefx/com.acme.treefx.TreeFX</module>
    <runtimeImage>target/treefx</runtimeImage>
    <linuxShortcut>true</linuxShortcut>
    <linuxPackageName>treefx</linuxPackageName>
    <linuxAppCategory>Utilities</linuxAppCategory>
    <linuxMenuGroup>Utilities</linuxMenuGroup>
    <icon>${project.basedir}/duke.png</icon>
    <javaOptions>
      <option>-Dfile.encoding=UTF-8</option>
    </javaOptions>
  </configuration>
</plugin>

I have to spend some words about the configuration because it's a little bit tricky on some points.

Easy part

name sets the application name (you don't say!)
appVersion as any other application, our TreeFX has a release version too!
vendor sets the vendor or our Acme company
destination sets the destination folder for the generated package

Tricky part

module we set the module and the main class of our application since JPackage will generate a new executable for our package
runtimeImage is usually set to our development OpenJDK, because JPackge could work with JLink to create the custom JRE, but because of some issues I found with JavaFX and JLink (the process is not straightforward), we trick JPackage and we tell it to take the previous custom JRE created with the OpenJFX plugin

Linux part (I put it just for the sake of completeness)

linuxShortcut creates a shortcut after the application installation process
linuxPackageName sets the final deb package name
linuxAppCategory sets the Linux application category
linuxMenuGroup sets the Linux menu-group

Maybe I have to spend more words on the tricky part. As mentioned above, JPackage is a tool for packaging a self-contained Java application, any kind of Java application, modularized and non-modularized applications. But for this example, we want to distribute a simple animation and it would be quite weird to distribute a whole JRE image for the sake of one animation. Therefore, always think about what you really need for your application and let the right tool works for you.

About linux part. JPackage allows you to package not only for Linux, but also for Windows and MacOS. However, JPackage depends on the operating system in order to work, so you can't package a Linux app, if you're using Window or MacOS, in a nutshell, no, cross-compiling is not an option. We will try to package our TreeFX app for Windows in an addendum to this article (maybe next week).

Enough chatting. Let's see it work:



 $ mvn clean compile javafx:jlink jpackage:jpackage

Wait a while and then:



 $ ls target/dist/
treefx_1.0.0-1_amd64.deb

Nice fourice! We now have our package for AMD64 arch (yeah that's mine). If we still have some curiosity about how the deb package is organized, we can unpack it:



 $ dpkg-deb -R target/dist/treefx_1.0.0-1_amd64.deb target/unpacked
 $ tree target/unpacked



target/unpacked
├── DEBIAN
│   ├── control
│   ├── postinst
│   ├── postrm
│   ├── preinst
│   └── prerm
└── opt
    └── treefx
        ├── bin
        │   └── TreeFX
        ├── lib
        │   ├── app
        │   │   └── TreeFX.cfg
        │   ├── runtime
        │   │   ├── [...]
        │   ├── TreeFX.png
        │   └── treefx-TreeFX.desktop
        └── share
            └── doc
                └── copyright

We could optimize things and surf into other JPackage configurations, but so far so good (as an old Bryan Adams' album and song said).

Conclusions

We finally reached the bottom of this tutorial, a lot more could be said about JPMS, JLink and JPackage, this is indeed an introduction to tickle your curiosity, and as you can see we can still build nice stuff with JavaFX and Java technology.

Dig into https://gitlab.com/lucaguada/treefx repository and look at the different branches to have a clearer vision of the above steps.

See you, space cowboy!

«A chance to begin again in a golden land of opportunity and adventure!»

CQRS and Event Sourcing

atipiJ — Mon, 21 Dec 2020 12:57:00 +0000

Introduction

The purpose of this article is to go through some basic concepts of Command Query Responsibility Segregation (CQRS) and Event Sourcing design patterns.
There's going to be an analysis of these patterns and then we will implement these to the use case.
We will apply CQRS to the use case and we will apply Event Sourcing to fix some shortcomings of CQRS.

Repository containing the Kata that we solved to implement the patterns: CherryChain bank account kata

CQRS

CQRS is a pattern used for modeling the domain of software. Its name stands for Command Query Responsibility Segregation and, as you can guess, its goal is to separate the writing part of the data from the reading part.
Often in projects, these two concepts are enclosed in a single domain model.
However, as the application begins to grow, it could easily become difficult to manage and expensive to maintain.
On the contrary, by separating these two concepts of commands and queries, it is possible to maintain a low complexity even in very large projects.

Event Sourcing

Event Sourcing is an event-centric technique for persisting the Application State. In this approach, the state is stored as a series of events, in an event store. When the state is updated, events are emitted and get inserted into the event store.

Loading the state consists of iterating through all the saved events and summing them to reconstruct the state in its entirety. Whenever the state gets modified, an event is emitted. This means that any consumer that is listening to these events can handle them accordingly, using the data inside the event. A consumer is responsible for listening and consuming events.

Event Sourcing has some inherent benefits when used as an application architecture, for example, it makes write operations much faster since it only has to generate and save a lightweight event.

Using Event Sourcing alongside CQRS brings some advantages. In CQRS there is an application write repository which normally saves as-is when commands arrive. By substituting this part of the application with Event Sourcing, writing operations will become faster, and it will also be possible to have application state snapshots at any point in time.

Use case

The scenario in which we're going to apply these patterns is the Bank account kata using the Kotlin programming language.
The goal is to create a service that allows basic "banking" operations such as deposit and withdrawal. Of course everything
has to be programmed following the patterns accordingly.

Here is a list of features that we are going to develop:

The bank can register a customer
The bank can create an account for a customer
The bank can deposit money to an account
The bank can withdraw money from an account

The bank can register a customer

The first step was to create a class representing the bank and allowing it to register a customer.
Following the TDD (Test Driven Development) methodology we started by creating the following test and then proceeding with the implementation of the feature:

  @Test
  internal fun `it should register a customer`() {
    val customerId = bank.registerCustomer("Firstname Lastname")

    assertThat(bank.getCustomer(customerId).fullName).isEqualTo("Firstname Lastname")
  }

So what we need is a function to register a customer and another one to read a customer, given his identifier.
From now we can already begin to think about the CQRS pattern, which allows us to keep the concepts of writing and reading data isolated.
When we want to register a customer we will have to launch a command, while when we want to read a customer we will have to make a query.

At the moment we can focus on the registration function, in which the customer's full name is passed as input. The function then will generate an id (the id will be then returned by the function) and launch a command to create the new user.

  fun registerCustomer(fullName: String): UUID {
    val id = UUID.randomUUID()
    commandBus.publish(RegisterCustomer(id, fullName))
    return id
  }

In our kata, to implement the registration of a new customer, we created the RegisterCustomer command:

sealed class Command

data class RegisterCustomer(val id: UUID, val fullName: String): Command()

and the DefaultCommandBus communication channel which is used to publish the commands that reach a class (which extends the CommandHandler interface) that knows how to handle them:

interface CommandBus {
  fun publish(command: Command)
}

class DefaultCommandBus(private val handler: CommandHandler) : CommandBus {
  override fun publish(command: Command) {
    handler.handle(command)
  }
}

We have called the class used to manage these commands DefaultCommandHandler, which, upon receiving RegisterCustomer command, creates a new customer, registers it, and then saves it in the repository:

interface CommandHandler {
  fun handle(command: Command)
}

class DefaultCommandHandler(private val customerRepository: Repository<Customer>) : CommandHandler {
  override fun handle(command: Command) {
    when (command) {
      is RegisterCustomer -> registerCustomer(command)
    }
  }

  private fun registerCustomer(command: RegisterCustomer) {
    val customer = Customer()
    customer.register(command.id, command.fullName)
    customerRepository.put(customer)
  }
}

The Customer state contains information regarding the bank's customer. There are an identification code and full name. Since we have used the Event Sourcing pattern to make sure our client's story is not lost, we don't update the information directly. We save it as a list of events. As we can see, the function used for registration adds the CustomerRegistered event
to the customer state's changes list:

class Customer {
  lateinit var id: UUID
  val changes = mutableListOf<Event>()

  fun register(id: UUID, fullName: String) {
    changes.add(CustomerRegistered(id, fullName))
  }
}

interface Event { val id: UUID }

data class CustomerRegistered(override val id: UUID, val fullName: String) : Event

The Repository where our new customer is stored is made up of an EventStore and an EventBus. When the function to save the customer is called, each of the events regarding that particular customer (identified by an id) are saved in the Store and then sent via the dedicated Bus.

interface Repository<T> {
  fun put(entity: T)
}

class InMemoryCustomerRepository(private val eventStore: EventStore, private val eventBus: EventBus): Repository<Customer> {
  override fun put(customer: Customer) {
    customer.changes.forEach {
      eventStore.append(it)
      eventBus.emit(it)
    }
  }
}

InMemoryEventStore is our class that takes care of keeping events in memory and then being subsequently interrogated when necessary. (It doesn't necessarily need to be an in-memory implementation, the EventStore interface can be implemented as any one wishes).

interface EventStore {
  fun append(event: Event)
}

class InMemoryEventStore : EventStore {
  private val list = mutableListOf<Event>()

  override fun append(event: Event) {
    list.add(event)
  }
}

DefaultEventBus is the communication channel where all the events are sent which are then managed by the EventHandler:

interface EventBus {
  fun emit(event: Event)
}

class DefaultEventBus(private vararg val handlers: EventHandler) : EventBus {
  override fun emit(event: Event) {
      handlers.forEach { it.handle(event) }
  }
}

CustomerViewsEventHandler is a class that handles customer events and is has a collaborator called CustomerViews. When this handler receives a CustomerRegistered event, it creates a CustomerView with the data it receives and projects it in the view:

interface EventHandler {
  fun handle(event: Event)
}

class CustomerViewsEventHandler(private val customerViews: CustomerViews) : EventHandler {
  override fun handle(event: Event) {
    when(event) {
      is CustomerRegistered -> handle(event)
    }
  }

  fun handle(event: CustomerRegistered) {
    customerViews.add(CustomerView(event.id, event.fullName))
  }
}

data class CustomerView(val id: UUID, val fullName: String)

InMemoryCustomerViews is the view where all customer information is projected into. The views can then be queried by the bank using the get function. This keeps the command and query logic separate as indicated by the CQRS pattern:

interface CustomerViews {
  fun add(customerView: CustomerView)
  fun get(customerId: UUID): CustomerView
}

class InMemoryCustomerViews: CustomerViews {
  private val data = mutableMapOf<UUID, CustomerView>()

  override fun add(customerView: CustomerView) {
    data[customerView.id] = customerView
  }

  override fun get(customerId: UUID): CustomerView {
    return data[customerId]!!
  }
}

In this way, the first point of the kata was completed following the CQRS and Event Sourcing pattern.

The bank can create an account for a customer

Now we want to make sure that the bank can create one or more accounts for each customer where the balance will be stored.
After creating the test we went on to write a function to create an account on a specific customer, in which an id is generated to identify the account and then publish the CreateAccount command on the command bus:

  fun createAccount(customerId: UUID): UUID {
    val id = UUID.randomUUID()
    commandBus.publish(CreateAccount(id, customerId));
    return id;
  }

As we saw in the previous chapter, the class that will handle the command we published is DefaultCommandHandler.
So in addition to the management of the "RegisterCustomer" command, we have added the management of the new CreateAccount command.
A new account is created with the id and customer id passed by the command and finally stored in the repository in the same way we saw in the previous chapter (using EventStore and EventBus).

class DefaultCommandHandler(private val accountRepository: Repository<Account>, private val customerRepository: Repository<Customer>) : CommandHandler {

  override fun handle(command: Command) {
    when (command) {
      is RegisterCustomer -> registerCustomer(command)
      is CreateAccount -> create(command)
    }
  }

  private fun registerCustomer(command: RegisterCustomer) {
    val customer = Customer()
    customer.register(command.id, command.fullName, command.type)
    customerRepository.put(customer)
  }

  private fun create(createAccount: CreateAccount) {
    val currentAccount = Account()
    currentAccount.create(createAccount.id, createAccount.customerId)
    accountRepository.put(currentAccount)
  }
}

The Account class is composed as follows: an identifier, a customer id, and a list of events.
When the creation method is used, the AccountCreated event is produced which will then be published by the Repository on the event bus.

class Account() {
  lateinit var id: UUID
  lateinit var customerId: UUID
  val change = mutableListOf<Event>()

  fun create(id: UUID, customerId: UUID) {
    change.add(AccountCreated(id, customerId))
  }
}

At this point, the AccountViewsEventHandler class will receive the AccountCreated event with which the AccountViews projection will be populated. It will then be possible to make queries to find out the balance of each customer.
As you can see from the code, in the phase of creating the balance it is initialized to zero.

class AccountViewsEventHandler(private val accountViews: AccountViews) : EventHandler {
  override fun handle(event: Event) {
    when (event) {
      is AccountCreated -> handle(event)
    }
  }

  private fun handle(accountCreated: AccountCreated) {
    accountViews.insert(AccountView(
      id = accountCreated.id,
      customerId = accountCreated.customerId,
      balance = 0
    ))
  }
}

Also in this new feature that we added to the bank, we were able to follow the CQRS and ES pattern.

The bank can deposit money to an account

To allow the bank to deposit money to an account, we allowed our application to handle the DepositMoney command. The handler of this command will update the Account state by increasing its balance based on the amount deposited. Before doing this update operation, it first has to load the existing account information. This is the function (present in the DefaultCommandHandler) that handles the command:

private fun deposit(depositMoney: DepositMoney) {
    val currentAccount = accountRepository.get(id = depositMoney.accountId)

    if (currentAccount!=null) {
      currentAccount.deposit(depositMoney.amount)
      accountRepository.put(currentAccount)
    } else {
      throw BankException("account does not exist")
    }
  }

We know that this information is stored as a series of events in our event store. So before doing any update operation, the application first reads all the events, saved in the event store, relative to the specific account in which the money will be deposited.
We then create an empty instance of Account. These events are then used in the hydrate function present in the empty Account instance, which iterates through the events and creates the state information from scratch.

class InMemoryAccountRepository(private val eventStore: EventStore, private val eventBus: EventBus) : Repository<Account> {
  override fun get(id: UUID): Account? {
    val events = eventStore.getEventsBy(id = id)
    return if (events.isEmpty()) {
      null
    } else {
      val account = Account()
      account.hydrate(events)
      account
    }
  }
}

The hydrate() function works as follows:

  fun hydrate(events: List<Event>) {
    events.forEach {event ->
      when (event) {
        is AccountCreated -> {
          this.id = event.id
          this.customerId = event.customerId
        }
        is MoneyDeposited -> this.balance += event.amount
      }
    }
  }

This is a key step in Event Sourcing. As we can see, all the events are processed and based on their type, an update operation is applied to the Account instance. This means that the first event must be an AccountCreated event, which will set the account's id and customerId. Every other MoneyDeposited event that might have been emitted in the past will be iterated through, and will update the current Account instance. After the hydration is done, we will have in our hands the instance of the account that had been stored as a series of events in the event store. We update the changes of this instance by putting in our new MoneyDeposited event, which is then emitted to the event bus and appended to the event store.

class InMemoryAccountRepository(private val eventStore: EventStore, private val eventBus: EventBus) : Repository<Account> {
  override fun put(account: Account) {
    account.change.forEach { event ->
      eventStore.append(event)
      eventBus.emit(event)
    }
  }
}

It is intercepted by the AccountViewsEventHandler. This handler is responsible for keeping the Account Views synchronized with the actual state. The Views are the part of the application that are visible externally. The views don't necessarily store the state as-is the state can receive any desired transformation before being exposed externally.

class AccountViewsEventHandler(private val accountViews: AccountViews) : EventHandler {
  private fun handle(moneyDeposited: MoneyDeposited) {
    val accountView = accountViews.get(moneyDeposited.id)
    val updatedView = accountView.copy(balance = accountView.balance + moneyDeposited.amount)
    accountViews.delete(moneyDeposited.id)
    accountViews.insert(updatedView)
  }
}

With this feature, it is possible to deposit money into an account. We also encountered the hydrate() function, which helps us to update an existing account by iterating through the events present in the event store.

The bank can withdraw money from an account

To allow the bank to withdraw money from an account, the application handles the WithdrawMoney command. This operation is analogous to the depositing of money. This function is present in the DefaultCommandHandler:

private fun withdraw(withdrawMoney: WithdrawMoney) {
  val currentAccount = accountRepository.get(id = withdrawMoney.accountId)

  if (currentAccount.hasEnough(withdrawMoney.amount)) {
    currentAccount.withdraw(withdrawMoney.amount)
    accountRepository.put(currentAccount)
  } else {
    throw BankException("not enough money")
  }

It updates the Account by decreasing its balance based on the amount withdrawn. The withdraw process first loads the state information by iterating through its stored events (hydrate() function):

fun hydrate(events: List<Event>) {
  events.forEach {event ->
    when (event) {
      is AccountCreated -> {
        this.id = event.id
        this.customerId = event.customerId
      }
      is MoneyDeposited -> this.balance += event.amount
      is MoneyWithdrawn -> this.balance -= event.amount
    }
  }
}

After the events have been iterated through completely, then we have the Account instance that we wanted and we can proceed with updating it. At this point a MoneyWithdrawn event is created, containing the necessary information of the withdrawal it is then appended to the changes of the account:

fun withdraw(amount: Int, fee: Int) {
  change.add(MoneyWithdrawn(id, amount, fee))
}

This event is then saved to the event store, emitted to the event bus and intercepted by the AccountViewsEventHandler, which projects the changes to the AccountViews:

private fun handle(moneyWithdrawn: MoneyWithdrawn) {
  val accountView = accountViews.get(moneyWithdrawn.id)
  val updatedView = accountView.copy(balance = accountView.balance - moneyWithdrawn.amount)
  accountViews.delete(moneyWithdrawn.id)
  accountViews.insert(updatedView)
}

With this feature, it is possible to withdraw money from an account.

Conclusions

As we have seen, by using the CQRS and ES pattern, we managed to solve this kata.
Some pros of this pattern are:

There is a single source of truth
Views can be generated with high customization
The separation of actions that change the state and queries can bring high optimization

However there are some obstacles (or opportunities!) that need to be conquered when working with this pattern:

Kickstarting a project can prove difficult
There is a necessity to have a good model of the domain of the project
Events might lose meaning over time if the domain isn't modeled properly

Zero Downtime Deployment with Docker Swarm

ndr_brt — Mon, 21 Dec 2020 12:54:27 +0000

If you are a software developer that in the past has dealt with production software, you are certainly familiar with this struggle:

deployment time

I hope you have already put into effect measures to trust your deployment by taking advantage of great practices such as Continuous Integration, Automated Testing, Continuous Delivery and so on, these should be carried out before tackling the thing I’m going to talk about. This is not going to be a silver bullet.

Today we’ll talk about zero downtime.

the problem

Imagine you have already packaged your application into a docker image, published it into a docker registry and have it up and running in production by the following command:

docker run …

Nothing wrong, your application is good to go.

But… how can we update it?

The easiest solution is to stop the old one and start the new version

docker stop <running container id>

docker run <new version>

There’s a problem with this approach: from the time you stop the old container and the complete bootstrap of the new version, your application will not respond.

the solution

This problem is pretty common and can be solved in many ways. We discovered an almost “effort free” way using Docker Swarm.

what is Docker Swarm?

Swarm is a Docker “mode” already included in your Docker installation. It’s a powerful cluster engine that will help you scale your application.

Also, it will solve your downtime problems.

how?

The idea is to transform your docker instance into a single node swarm cluster:

docker swarm init

This command should return something like

Swarm initialized: current node (<node_id>) is now a manager.

To add a worker to this swarm, run the following command:

docker swarm join — token <swarm_token> <node_addess+port>

To add a manager to this swarm, run ‘docker swarm join-token manager’ and follow the instructions.

These are instructions for adding nodes to our cluster. But you don’t need that now.

Now you need to deploy the stack.

In order to achieve that, you need to define a docker-compose file like this:

version: ‘3.7’

networks:
  my-network:
    external: false

services:
  my-server:
    image: ${IMAGE}
    hostname: my-server
    container_name: my-server
    ports:
    - ‘8080:8080’
    networks:
    - my-network
    healthcheck:
      test: [“CMD”, “curl”, “-i”, "http://localhost:8080/health"]
    deploy:
      mode: replicated
      replicas: 2
      update_config:
        order: start-first
        failure_action: rollback
        delay: 5s

In this example there is just one container, but you can deploy as many as you need.

Some concepts:

— image: a variable that represents the image name and version

— network: the stack network, with this various services can communicate with each other

— healthcheck: definitions of commands that verify the service status (up or down). The commands should return error code = 0 when the service status is good and error code != 0 when the service has not started yet, stopped, paused, etc..

— replicas: the numbers of parallel containers of the service that will be deployed

You can start your service by the following command:

export IMAGE=${IMAGE_NAME}:${IMAGE_VERSION}

docker stack deploy -c docker-compose.yml <stack_name> — with-registry-auth

Once you are ready to update your service, you can use the same command, with a different value of IMAGE_VERSION, and there will be no downtime since Swarm will take care of starting the new containers and, once started correctly, it will also take care of stopping the old ones.

Throttle HTTP requests on paged resources with Vert.x

ndr_brt — Mon, 21 Dec 2020 12:53:40 +0000

Hey Java developers, I know that you are interested in asynchronous programming and I know you are struggling with such paradigm.
I know that because I’m one of you!

Let us see how to deal with rate limiting issues in the magical async world in Java with Vert.x.

For the unaware a little recap:

What is Vert.x?

Vert.x is a polyglot event-driven and non-blocking application framework.

What is “Rate limiting”?

Rate limiting is a set of policies implemented by a web server used to force client to “throttle” requests from being overloading.

What are paged resources?

Paged resources are resources split in various pages to prevent overloading.

We will work with a real implementation case:

We are music fanatics (as I am), we are users of Discogs.com, and we want to fetch its data in our non-blocking application.

Taking a look at the discogs.com API reference we will notice that — aside from the API specifications — there are some information about data format, headers, versioning and so on.
We are now interested in two sections: Rate Limiting and Pagination.
Those are the problems to deal with.

A sketch of our architecture

Every arrow from left to right is a method call, every arrow from right to left is a Future.

We will start from the bottom and go up the river.

Rate Limiting

Discogs permits, according with documentation, 25 request per minute, so, 25 will be our rate limit, and one minute is the size of the rate limit window.

The idea is to have an interface with a method called *execute *which will accept a Request (a data object that represents the request) as parameter and return a Future (because we think async!) of Buffer, that is the response’s body.
Trivial and neat.

interface Requests {
  Future<Buffer> **execute**(Request request);
}

We need to put a delay between requests, hence inside the RequestExecutor implementation we will use a queue to decouple the execute invocations from the throttled request execution:

class ThrottledRequests implements Requests {

  ...

  @Override
  public Future<Buffer> execute(Request request) {
    try {
      Future<Buffer> future = Future.future();
      queue.put(new RequestContext(request, future));
      return future;        
    } catch (Exception e) {
      return Future.failedFuture(e);
    }
  }

  ...

}

RequestContext is a simple data object that allow us to bind request to its future.

Then, the request should be consumed from the queue and executed, we will do this with a periodic task by starting it every N milliseconds, where N is a non-fixed value computed from the maximum number of requests we are allowed to do in the rate limit window.

At first we will start with an optimistic 1, therefore here’s our ThrottledRequests constructor:

public class ThrottledRequests implements Requests {
  private static final int RATE_LIMIT_WINDOW_SIZE = 60000;
  private final Logger log = getLogger(getClass());
  private final BlockingQueue<RequestContext> queue = new LinkedBlockingQueue<>();
  private final AtomicLong executorId = new AtomicLong(0);
  private final AtomicLong actualDelay = new AtomicLong(100);
  private final HttpClient http;
  private final Vertx vertx;

  ThrottledRequests(Vertx vertx) {
    this.vertx = vertx;
    this.http = vertx.createHttpClient();
    long id = vertx.setPeriodic(actualDelay.get(), executor());
    this.executorId.set(id);
  }

  ...
}

Some explanations:

vertx: you need the vertx instance to launch a periodic executor.
actualDelay: is an AtomicLong that represent the actual delay, so, as we said before, it’s starting value is 1.
executor(): core method of the class, it insantiate a function that will perform the throttled requests.
executorId: we need to track the executor task id to stop it when the actualDelay will change. Is an AtomicLong too

Let us see how to handle async request executions:

private Handler<Long> executor() {
  return timerId -> {
    RequestContext context = queue.poll();
    if (context != null) {
      Request inputRequest = context.request;
      http.request(inputRequest.method(), inputRequest.options())
        .putHeader("User-Agent", "YourApp/0.1")
        .setFollowRedirects(true)
        .handler(response -> {
          response.bodyHandler(context.future::complete);
          checkAndUpdateRateLimit(response);
        })
        .end();
    }
  };
}

Easy, uh?
On every request responded we complete the future and…checkAndUpdateRateLimit():
according to the API documentation, there is a header in every response we will get from discogs called X-Discogs-Ratelimit, that tells us how many requests we can do in a rate limit window of 1 minute. Cool.

private void checkAndUpdateRateLimit(HttpClientResponse response) {
  Optional.ofNullable(response.getHeader("X-Discogs-Ratelimit"))
    .map(Long::parseLong)
    .map(rateLimit -> rateLimit - 1)
    .map(reqPerMinute -> RATE_LIMIT_WINDOW_SIZE / reqPerMinute)
    .ifPresent(throttleDelay -> {
      if (throttleDelay != actualDelay.getAndSet(throttleDelay)) {
        vertx.cancelTimer(executorId.get());
        long id = vertx.setPeriodic(throttleDelay, executor());
        executorId.set(id);
      }
    });
}

The calculation is pretty simple:

Check the rate limit header
Subtract one, just to avoid to fill the window
Calculate the delay you should take between requests to do less than rateLimit requests in the time window (one minute)
If the delay is different from the current one, we remove the old executor’s timer and define another one with a new delay time.

Pagination

Handling pagination is a little trickier than rate limit, since there is some work to do with futures, you know, the total page count will be known only after fetching the first page.

The idea is to divide responsibilities, we will implement three components:
(the example is on the Inventory resource, which retrieves Listing entities)

GetListingPage: one implementation for every resource, such class uses Requests as collaborator, knows the url to invoke and knows how to deserialize the single fetched page. So its responsability is the Page.
Pages uses GetResourcePage to fetch first page and eventually the others, join them. Its responsibility is the Pages.
Client obtains all the pages from Pages and extract the entities from those.

Let’s start from GetListingPage, is a class that implement an interface with one method: apply

@Override
Future<ListingPage> apply(String userId, Integer pageNumber) {
  log.info("Request {} inventory page {}", userId, pageNumber);
  String path = String.format("/users/%s/inventory?page=%d", userId, pageNumber);
  Request request = Request.get(path);

  Future<ListingPage> future = Future.future();
  executor.execute(request).setHandler(async -> {
    if (async.succeeded()) {
      String json = async.result().toString();
      ListingPage page = Json.fromJson(json, ListingPage.class);
      future.complete(page);
    } else {
      future.fail(async.cause());
    }
  });

  return future;
}

Nothing special, execute request and deserialize JSON output to ListingPage class (a class that represent the ListingPage structure).

Maybe the more interesting class is Pages. The public method is getFor:

public Future<List<T>> getFor(String userId) {
  Future<List<T>> result = Future.*future*();

  getPage.apply(userId, 1).setHandler(async -> {
    if (async.succeeded()) {
      T firstPage = async.result();
      int totalPages = firstPage.pages();
      log.info("Total pages: {}", totalPages);

      List<Future> futures = IntStream
        .range(2, totalPages + 1)
        .mapToObj(page -> getPage.apply(userId, page))
        .collect(Collectors.toList());

      CompositeFuture.all(futures)
        .setHandler(joinPages(result,firstPage));
    }
  });

  return result;
}

We will ask for the first page and then, knowing the total pages count, we’ll ask for every other page, combining all the Futures into a CompositeFuture.

When every future will be completed, we will joinPages:

private Handler<AsyncResult<CompositeFuture>> joinPages(
  Future<List<T>> future, T firstPage
) {
  return async -> {
    if (async.succeeded()) {
      CompositeFuture remnants = async.result();
      Collection remnantPages = IntStream
        .range(0, remnants.size())
        .mapToObj(remnants::resultAt)
        .map(Page.class::cast)
        .collect(Collectors.toList());

      List<T> total = new ArrayList<>();
      total.add(firstPage);
      total.addAll(remnantPages);

      future.complete(total);
    } else {
      future.fail(async.cause());
    }
  };
}

Now, at the higher level (Client) we can just get pages and extract entities:

listingPages.getFor(user).setHandler(async -> {
  if (async.succeeded()) {
    List<Listing> listings = async.result().stream()
      .map(ListingPage::listings)
      .flatMap(Collection::stream)
      .collect(Collectors.*toList*()));

    ... do your logic ...

  }
});

Pretty cool, huh?

Source code:

TDD in an event driven application

ndr_brt — Mon, 21 Dec 2020 12:52:53 +0000

You (should) know, Test Driven Development is one of the greatest practice in software development: it gives you confidence in what you implement, it gives you courage to refactor or change behaviors, it forces you to think about what you are about to implement.

For sure not every piece of code deserves to be written test-first, but at least you should start thinking: “Can I write a test before?”.

Passing to an event driven world the process remains the same:
there’s a SUT, a trigger action, and some behavior verification:

Consider implementing a feature for an hypothetical Continuous Integration application, on tests executed should check the results and notify to the application. This application in based on Vert.x framework and is coded in Kotlin language.

So our first test will be: “on tests executed, when they passed, should emit tests passed”:

@ExtendWith(VertxExtension::class)
internal class CheckTestsExecutionTest {

  @Test
  @Timeout(3, timeUnit = SECONDS)
  internal fun `on tests executed, when they passed, should emit tests passed`(vertx: Vertx, context: VertxTestContext) {

    CheckTestsExecution(vertx.eventBus()).start()

    vertx.eventBus().consumer<JsonObject>("TestsPassed") {
      context.completeNow()
    }

    vertx.eventBus()
      .publish("TestsExecuted", JsonObject.mapFrom(
        TestsExecuted(listOf(
          TestResult("simple test", passed = true)
        )))
      )
  }
}

There’s a lot of stuff in there. The VertxExtension *class is a junit5 extension provided by the library *vertx-junit5 that gives some facilities to write Vert.x unit tests.

But, the very important thing here is that there’s no assertion. The reason is simple: we are testing an asynchronous behavior, so we will get no synchronous output from our .

What we need here is to set an expectation. To do that we create a consumer on the event we expecting to be launched from our SUT. To make test pass then we call the completeNow method on context object (is a Vert.x junit5 extension utility to handle asynchronous tests, as well as @Timeout annotation).
The test will pass only when completeNow method is called before the timeout. Simple.

The last test row emits an event that should trigger our desired behavior.
Obviously, after fixing compilation errors, the test will fail. Red bar, cool.

Let’s write only enough code to make test pass:

class CheckTestsExecution(private val eventBus: EventBus) {

  fun start() {
    eventBus.consumer<JsonObject>("TestsExecuted") {
      eventBus.publish("TestsPassed", JsonObject())
    }
  } 
}

Green bar. Go on.

Now, we need to describe the case where one test fails and so we will expect to receive TestsFailed event. We should also expect not to receive TestsPasses.

  @Test
  @Timeout(3, timeUnit = SECONDS)
  internal fun `on tests executed, when they failed, should emit tests passed`(vertx: Vertx, context: VertxTestContext) {
    CheckTestsExecution(vertx.eventBus()).start()

    vertx.eventBus().consumer<JsonObject>("TestsPassed") {
      context.failNow(RuntimeException("No TestPassed event should be emitted"))
    }

    vertx.eventBus().consumer<JsonObject>("TestsFailed") {
      context.completeNow()
    }

    val testResults = listOf(
      TestResult("simple test", passed = false)
    )
    vertx.eventBus()
      .publish("TestsExecuted", JsonObject.mapFrom(TestsExecuted(testResults)))
  }

Red bar again, so, make it green ASAP!

class CheckTestsExecution(private val eventBus: EventBus) {

  fun start() {
    eventBus.consumer<JsonObject>("TestsExecuted") { message ->
      val testExecuted = message.body().mapTo(TestsExecuted::class.java)
      if (testExecuted.tests.none { !it.passed }) {
        eventBus.publish("TestsPassed", JsonObject())
      } else {
        eventBus.publish("TestsFailed", JsonObject())
      }
    }
  }

}

Perfect, you just implemented an event driven component with TDD on Vert.x in Kotlin. Refactoring is up to you.

As you see, no great difference from the classic TDD approach: think, write a test, do something and expect behaviors, just pay attention to asynchronous manners and use expectations.

Handle backpressure between Kafka and a database with Vert.x

ndr_brt — Thu, 06 Aug 2020 06:52:39 +0000

As we already discussed in the past, asynchronous programming brings many pros in developing reactive and responsive applications. However, it also carries cons and challenges, one of the main ones is the backpressure problem.

What is backpressure?

In physics

it is a resistance or force opposing the desired flow of fluid through pipes (wikipedia)

We can translate the problem on a known scenario: persistence of messages polled from a bus, where there are a huge amount of messages on a bus that our application is polling really fast, but the database underneath is really slow.

How can be funnel overflow be avoided?

In a synchronous scenario, there's no backpressure's issue, the sync nature of the computation blocks the polling from the bus until the current message is processed.
But, in the async world, the polling is executed without clues on what's happening on the database. So if the database can't handle all the messages coming from the bus, the messages will remain "in between", which means in the memory of our service.
This can lead to failures or, at worst, to service fault.

Let's try to develop an application that persists messages in a database, and make it evolve to handle backpressure

Automatic polling

At first, our verticle will do those operations:

initialize the JDBC client
initialize the Kafka client
subscribe to the topic
persist the records

The code is quite simple, and it works well with small amounts of messages.
When the load gets bigger and bigger a problem appears: using the handler of Vertx Kafka consumer means that there's no control on the message ratio, so it will poll continuously without considering the persistance rate, causing memory overload.



public class MainVerticle extends AbstractVerticle {

@Override

  public void start(Promise<Void> startPromise) throws Exception {

    JDBCClient jdbc = JDBCClient.create(vertx, datasourceConfiguration());

    KafkaConsumer

      .<String, String>create(vertx, kafkaConsumerConfiguration())

      .subscribe("topic.name", startPromise)

      .handler(record -> {

        persist(jdbc, record)

          .onSuccess(result -> System.out.println("Message persisted"))

          .onFailure(cause -> System.err.println("Message not persisted " + cause));

      });

  }

private Map<String, String> kafkaConsumerConfiguration() {

    final Map<String, String> config = new HashMap<>();

    config.put(BOOTSTRAP_SERVERS_CONFIG, "kafkahost:9092");

    config.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    config.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    return config;

  }

private Future<UpdateResult> persist(JDBCClient jdbc, KafkaConsumerRecord<String, String> record) {

    Promise<UpdateResult> promise = Promise.promise();

    JsonArray params = toParams(record);

    jdbc.updateWithParams("insert or update query to persist record", params, promise);

    return promise.future();

  }

private JsonObject datasourceConfiguration() {

    // TODO datasource configuration

    return null;

  }

private JsonArray toParams(KafkaConsumerRecord<String, String> record) {

    // TODO: convert the record into params for the sql command

    return null;

  }

}

Explicit polling

To handle the backpressure, explicit polling shall be used, and this can be done by avoiding the kafka consumer's handler setting and by calling poll manually (in the following case, every 100ms).
By using this approach, it can be made so that every poll gets performed only when the batch of previously polled messages are persisted.
This behaviour can be achieved by handling every message's persist future and collecting all of them with the CompositeFuture.all, that will succeed only when all the messages are completed, and only in this case the next polling can be made.
If at least one of the future fails, everything will fail, and the polling will stop.
There are various solutions that can be adopted to make the service handle the failure, e.g. sending the message to a Dead Letter Queue, but we will not cover this case.

The problem with this code is that if a message fails, we will lose it, because the consumer is set to auto-commit, so, it's vertx that commits the topic offset.



public class MainVerticle extends AbstractVerticle {

@Override

  public void start(Promise<Void> startPromise) throws Exception {

    JDBCClient jdbc = JDBCClient.create(vertx, datasourceConfiguration());

    KafkaConsumer<String, String> consumer = KafkaConsumer

      .<String, String>create(vertx, kafkaConsumerConfiguration())

      .subscribe("topic.name", startPromise);

<span class="n">poll</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">consumer</span><span class="o">);</span>



}

private void poll(JDBCClient jdbc, KafkaConsumer<String, String> consumer) {

    Promise<KafkaConsumerRecords<String, String>> pollPromise = Promise.promise();

    consumer.poll(100, pollPromise);

<span class="n">pollPromise</span><span class="o">.</span><span class="na">future</span><span class="o">()</span>
  <span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">records</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">List</span><span class="o">&lt;</span><span class="nc">Future</span><span class="o">&lt;</span><span class="nc">UpdateResult</span><span class="o">&gt;&gt;</span> <span class="n">futures</span> <span class="o">=</span> <span class="nc">IntStream</span><span class="o">.</span><span class="na">range</span><span class="o">(</span><span class="mi">0</span><span class="o">,</span> <span class="n">records</span><span class="o">.</span><span class="na">size</span><span class="o">())</span>
      <span class="o">.</span><span class="na">mapToObj</span><span class="o">(</span><span class="nl">records:</span><span class="o">:</span><span class="n">recordAt</span><span class="o">)</span>
      <span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="n">record</span> <span class="o">-&gt;</span> <span class="n">persist</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">record</span><span class="o">))</span>
      <span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="n">toList</span><span class="o">());</span>

    <span class="k">return</span> <span class="nc">CompositeFuture</span><span class="o">.</span><span class="na">all</span><span class="o">(</span><span class="k">new</span> <span class="nc">ArrayList</span><span class="o">&lt;&gt;(</span><span class="n">futures</span><span class="o">));</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onSuccess</span><span class="o">(</span><span class="n">composite</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">System</span><span class="o">.</span><span class="na">out</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"All messages persisted"</span><span class="o">);</span>
    <span class="n">poll</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">consumer</span><span class="o">);</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onFailure</span><span class="o">(</span><span class="n">cause</span> <span class="o">-&gt;</span> <span class="nc">System</span><span class="o">.</span><span class="na">err</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"Error persisting messages: "</span> <span class="o">+</span> <span class="n">cause</span><span class="o">))</span>
<span class="o">;</span>



}

private Map<String, String> kafkaConsumerConfiguration() {

    final Map<String, String> config = new HashMap<>();

    config.put(BOOTSTRAP_SERVERS_CONFIG, "kafkahost:9092");

    config.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    config.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    return config;

  }

...

}

Manual commit

Setting the ENABLE_AUTO_COMMIT_CONFIG properties to false, the service takes ownership of the topic offset commit.
The commit will be performed only when every message will be persisted, with this trick the at least once delivery is achieved.



public class MainVerticle extends AbstractVerticle {

@Override

  public void start(Promise<Void> startPromise) throws Exception {

    JDBCClient jdbc = JDBCClient.create(vertx, datasourceConfiguration());

    KafkaConsumer<String, String> consumer = KafkaConsumer

      .<String, String>create(vertx, kafkaConsumerConfiguration())

      .subscribe("topic.name", startPromise);

<span class="n">poll</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">consumer</span><span class="o">);</span>



}

private void poll(JDBCClient jdbc, KafkaConsumer<String, String> consumer) {

    Promise<KafkaConsumerRecords<String, String>> pollPromise = Promise.promise();

    consumer.poll(100, pollPromise);

<span class="n">pollPromise</span><span class="o">.</span><span class="na">future</span><span class="o">()</span>
  <span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">records</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">List</span><span class="o">&lt;</span><span class="nc">Future</span><span class="o">&lt;</span><span class="nc">UpdateResult</span><span class="o">&gt;&gt;</span> <span class="n">futures</span> <span class="o">=</span> <span class="nc">IntStream</span><span class="o">.</span><span class="na">range</span><span class="o">(</span><span class="mi">0</span><span class="o">,</span> <span class="n">records</span><span class="o">.</span><span class="na">size</span><span class="o">())</span>
      <span class="o">.</span><span class="na">mapToObj</span><span class="o">(</span><span class="nl">records:</span><span class="o">:</span><span class="n">recordAt</span><span class="o">)</span>
      <span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="n">record</span> <span class="o">-&gt;</span> <span class="n">persist</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">record</span><span class="o">))</span>
      <span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="n">toList</span><span class="o">());</span>

    <span class="k">return</span> <span class="nc">CompositeFuture</span><span class="o">.</span><span class="na">all</span><span class="o">(</span><span class="k">new</span> <span class="nc">ArrayList</span><span class="o">&lt;&gt;(</span><span class="n">futures</span><span class="o">));</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">composite</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">Promise</span><span class="o">&lt;</span><span class="nc">Void</span><span class="o">&gt;</span> <span class="n">commitPromise</span> <span class="o">=</span> <span class="nc">Promise</span><span class="o">.</span><span class="na">promise</span><span class="o">();</span>
    <span class="n">consumer</span><span class="o">.</span><span class="na">commit</span><span class="o">(</span><span class="n">commitPromise</span><span class="o">);</span>
    <span class="k">return</span> <span class="n">commitPromise</span><span class="o">.</span><span class="na">future</span><span class="o">();</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onSuccess</span><span class="o">(</span><span class="n">any</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">System</span><span class="o">.</span><span class="na">out</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"All messages persisted and committed"</span><span class="o">);</span>
    <span class="n">poll</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">consumer</span><span class="o">);</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onFailure</span><span class="o">(</span><span class="n">cause</span> <span class="o">-&gt;</span> <span class="nc">System</span><span class="o">.</span><span class="na">err</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"Error persisting and committing messages: "</span> <span class="o">+</span> <span class="n">cause</span><span class="o">))</span>
<span class="o">;</span>



}

private Map<String, String> kafkaConsumerConfiguration() {

    final Map<String, String> config = new HashMap<>();

    config.put(BOOTSTRAP_SERVERS_CONFIG, "kafkahost:9092");

    config.put(ENABLE_AUTO_COMMIT_CONFIG, "false");

    config.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    config.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    return config;

  }

  ...

}

Bonus feature: achieve ordering

With a little effort it's possible to achieve ordering:
future composition allows forcing every persist operation to wait the completion of its precedent.
It's achievable by chaining the async computations one to another, so every one will be executed when the precedent future succeeds.
This is a smart pattern to be used when serialization is needed.



public class MainVerticle extends AbstractVerticle {

  ...

  private void poll(JDBCClient jdbc, KafkaConsumer<String, String> consumer) {

    Promise<KafkaConsumerRecords<String, String>> pollPromise = Promise.promise();

    consumer.poll(100, pollPromise);

<span class="n">pollPromise</span><span class="o">.</span><span class="na">future</span><span class="o">()</span>
  <span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">records</span> <span class="o">-&gt;</span> <span class="nc">IntStream</span><span class="o">.</span><span class="na">range</span><span class="o">(</span><span class="mi">0</span><span class="o">,</span> <span class="n">records</span><span class="o">.</span><span class="na">size</span><span class="o">())</span>
    <span class="o">.</span><span class="na">mapToObj</span><span class="o">(</span><span class="nl">records:</span><span class="o">:</span><span class="n">recordAt</span><span class="o">)</span>
    <span class="o">.</span><span class="na">reduce</span><span class="o">(</span><span class="nc">Future</span><span class="o">.&lt;</span><span class="nc">UpdateResult</span><span class="o">&gt;</span><span class="n">succeededFuture</span><span class="o">(),</span>
      <span class="o">(</span><span class="n">acc</span><span class="o">,</span> <span class="n">record</span><span class="o">)</span> <span class="o">-&gt;</span> <span class="n">acc</span><span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">it</span> <span class="o">-&gt;</span> <span class="n">persist</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">record</span><span class="o">)),</span>
      <span class="o">(</span><span class="n">a</span><span class="o">,</span><span class="n">b</span><span class="o">)</span> <span class="o">-&gt;</span> <span class="n">a</span>
    <span class="o">)</span>
  <span class="o">)</span>
  <span class="o">.</span><span class="na">compose</span><span class="o">(</span><span class="n">composite</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">Promise</span><span class="o">&lt;</span><span class="nc">Void</span><span class="o">&gt;</span> <span class="n">commitPromise</span> <span class="o">=</span> <span class="nc">Promise</span><span class="o">.</span><span class="na">promise</span><span class="o">();</span>
    <span class="n">consumer</span><span class="o">.</span><span class="na">commit</span><span class="o">(</span><span class="n">commitPromise</span><span class="o">);</span>
    <span class="k">return</span> <span class="n">commitPromise</span><span class="o">.</span><span class="na">future</span><span class="o">();</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onSuccess</span><span class="o">(</span><span class="n">any</span> <span class="o">-&gt;</span> <span class="o">{</span>
    <span class="nc">System</span><span class="o">.</span><span class="na">out</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"All messages persisted and committed"</span><span class="o">);</span>
    <span class="n">poll</span><span class="o">(</span><span class="n">jdbc</span><span class="o">,</span> <span class="n">consumer</span><span class="o">);</span>
  <span class="o">})</span>
  <span class="o">.</span><span class="na">onFailure</span><span class="o">(</span><span class="n">cause</span> <span class="o">-&gt;</span> <span class="nc">System</span><span class="o">.</span><span class="na">err</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="s">"Error persisting and committing messages: "</span> <span class="o">+</span> <span class="n">cause</span><span class="o">));</span>



}

  ...

}

Conclusion

Backpressure is a fundamental topic to cover when working with async programming.
It does not come for free out of the vert.x box, but it can be acheived with some simple tricks.

Introduction to Vert.x

Piero — Fri, 19 Jun 2020 07:56:35 +0000

What is Vert.x? What is it for?

Vert.x is a multi-language toolkit based on the JVM and used in software applications to implement
asynchronous and event driven structures.
However, to better understand its use, we must take a step back and explain what asynchronous programming is.

Asynchronous Programming

Usually a program executes instructions sequentially on a single thread and while a function is running,
it waits for it to finish before continuing with the next instruction, this is known as blocking execution.

Task B blocking execution:

 Main Thread:        |_Task A_| --> |______Task B______| --> |_Task C_|

Fortunately, modern computers are equipped with processors that are made of multiple cores that can execute instructions
in parallel (multithreading). A program can therefore proceed normally on the main thread while one of its
functions executes its own code on another thread in parallel.
However, the process is still synchronous, because if the main thread at some point needs the result of the
function on the other thread, and the calculation is still ongoing, then an exception is thrown.

Task C ends after the start of task D then an exception is thrown (task D needs the result of task C):

 Main Thread:        |_Task A_| --> |_Task B_| --> |_Task D_| --> exception
 Secondary Thread:              --> |______Task C______|

To solve this problem, asynchronous programming comes into play with the use of special APIs, which allow you to define actions to be performed as soon as the result of a specific function becomes available.

Task C generates the promise of a value while Task D knows that it must wait for this asynchronous result before processing its code:

 Main Thread:        |_Task A_| --> |_Task B_| --> async-result --> |_Task D_|
 Promise:                       --> |______Task C______|

One tool that allows you to use the asynchronous programming is Vert.x

Vert.x is

highly modular and consists of several components that can be used according to one's needs, in fact it cannot be defined as a framework, rather it should be considered a set of libraries.
light (the core is about 650kB), high-performance and simple to use. Therefore, ideal for the development of microservices and more ...
polyglot, so it supports several languages: Java, Kotlin, JavaScript, Groovy, Ruby, Scala, Ceylon.

What does it offer?

The most used modules:

Core: basic functionality with support for HTTP, TCP, file system access and various other features.
Web: tool-kit to write sophisticated web applications and HTTP microservices.
Data Access: asynchronous clients that can be used to access your PostgreSql, MySql, MongoDb, ...
Testing: integration and support of unit tests on asynchronous code.
Authentication and Authorisation: provides API for authentication and authorization with JWT, OAuth 2, JDBC auth, Mongo auth, ...
Devops: offers several components to monitoring the app while it runs in production (metrics, health checks, ...)

Other modules:

Reactive: add-ons to make an application even more responsive.
Microservices: components to build applications based on microservices.
MQTT: provides clients and servers which are able to open, manage and close a communication through MQTT.
Messaging: provides clients and servers which are able to communicate via AMQP, STOMP or RabbitMQ protocol.
Integration: provides clients to interact with Apache Kafka, Consul or send SMTP emails.
Event Bus Bridge: offers several "bridges" to extend the use of the Event Bus outside the app via TCP ports.
Clustering: supports different types of clustering (Hazelcast, Infinispan, Apache Ignite, Apache Zookeeper).
Services: useful services for encapsulating reusable functionality in other places, which are distributed using an identifier.
Cloud: tool to build an OpenShift platform for cloud applications.
Advanced: advanced tools to be used in particular projects but which are not normally used.

Asynchronous Programming in Vert.x

To understand how Vert.x allows implementing asynchronous programming, two key concepts must be clarified: what is a Verticle and what are Event Buses for?

Verticle is the deployment unit in Vert.x, which processes incoming events using an Event Loop on a specific Thread. By default, two Event Loops are connected to each Thread Core of the CPU. A Verticle can be instantiated several times.

On the other hand, when we talk about Events we mean, for example, the reception of "network buffers", "timing events" or the reception of messages from other Verticles.

The Event Bus is the communication channel that Verticles used to send messages asynchronously. The type of data exchanged can be of any type, but it is preferable to use the JSON format since it can be understood by all languages.

The supported communication patterns are: Point-to-Point messaging (direct message), Request / Response messaging and Publish / Subscribe messaging (to send messages in broadcast).

The Event Bus can also be distributed, meaning Verticles that are not necessarily running on the same machine can communicate with each other. In addition, thanks to the "bridges" it can also communicate with generic messaging protocols such as AMQP and STOMP.

After understanding the architecture behind Vert.x, now let's try to understand the two main constructs of asynchronous programming: Promise and Future.

Promise and Future

In abstract terms, Future and Promise can be thought of as values that at a certain point in time become available and therefore require a certain amount of time to be calculated or retrieved. For this reason, two states in which a Future / Promise can be found are generally recognized: Completed / Determined when the value is available or Incomplete / Indeterminate when the value has not yet been calculated.

In Vert.x there are Promise and Future constructs that refer to these abstractions:

Future: reference, read-only, at a value yet to be calculated
Promise: variable, assignable only once, to which Future refers

In other words, a Future shows the value previously written in a Promise in read-only mode.

NOTE: do not confuse Vert.x futures with Java ones!

Conclusion

So in conclusion Vert.x is a very useful tool that allows us to use asynchronous programming in different areas and
which however carries some advantages and disadvantages.

Strengths:

scalability, being modular allows it to be used only where it is deemed necessary
microservices, being light but at the same time performing it is ideal for the development of microservices
versatility, by supporting different languages you are not bound to use one that you don't know or don't like

Weaknesses:

learning curve, it takes time to fully understand the mechanisms behind the Vert.x modules
backpressure, which must be managed to avoid memory problems
delays, if an event is not properly handled it can generate unexpected delays

Authors:

dev.to/atipij
dev.to/piero90

Future Composition in Vert.x

ndr_brt — Wed, 20 May 2020 11:42:45 +0000

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)