DEV Community: Selim Dincer

Idiomatic Kotlin Abstractions for the Vert.x EventBus

Selim Dincer — Mon, 31 May 2021 06:15:23 +0000

The EventBus plays an important role in Vert.x applications. It is a convenient way for modules to communicate with each other using various messaging patterns.

Making use of the EventBus is fairly easy. Talking to a service and handling the response could look like this:

vertx.eventBus()
  .localConsumer<Void>(address)
  .handler { message ->
    message.reply(jsonObjectOf("message_of_the_day" to "Hello World"))
  }

val response = vertx.eventBus()
  .request<JsonObject>(address, null)
  .await()
  .body()

println(response.getString("message_of_the_day"))

The data types that we can transfer are fairly limited by default, but extensible via custom codecs.

If you make extensive use of the EventBus however, a few things might be reasons for concern:

What happens when we have a typo in the address?
What happens if the receiving side expects a String, but we send an Int?
How do we handle exceptions?

Sadly, the answer to the first two is runtime errors. So is there a way to avoid them, or even catch them at compile-time? Can we additionally improve exception handling and reduce boilerplate?

In order to answer these questions, let's take a step back and think about what more or less idiomatic kotlin code would look like. Ideally, we would expose a simple interface with the functionality that we would like to provide. The service in the code snippet above could be modelled as follows:

interface MessageOfTheDayService {
  suspend fun getMessageOfTheDay(): Result

  sealed class Result
  data class Success(val message: String): Result()
  data class Failure(val cause: Throwable): Result()
}

Simple, right? With a simple kotlin interface, there's no ambiguity in what type this service needs as input and what types of output it can produce. Additionally, when calling it, we don't need to specify an address, or think about exceptions too much. Wouldn't it be nice if we could define our services like this and still make use of the EventBus while not having to worry about any of the drawbacks?

What if we could get all of the benefits above with a single extra line? Turns out with a little bit of annotation processing and code generation we can!

EventBus-Service is a small library that takes care of this for us. Let's annotate the interface with @EventBusService and then take a look at the generated code:

private const val TOPIC: String = "co.selim.sandbox.messageofthedayservice"

public class MessageOfTheDayServiceImpl(
  private val vertx: Vertx
) : MessageOfTheDayService {
  public override suspend fun getMessageOfTheDay(): MessageOfTheDayService.Result = vertx.eventBus()
    .request<MessageOfTheDayService.Result>(TOPIC + ".getMessageOfTheDay", Unit, deliveryOptions)
    .await()
    .body()
}

This looks quite similar to the code we wrote ourselves but there are two important things to note here:

The address is inferred from the function defined in our service. This leaves no room for typos.
We replaced Void and JsonObject with the types that we're actually interested in.

Now, how would we reply to these requests? The same file also contains an extension property for handling them:

internal val Vertx.getMessageOfTheDayRequests: Flow<EventBusServiceRequest<Unit,
    MessageOfTheDayService.Result>>
  get() = eventBus()
    .localConsumer<Unit>(TOPIC + ".getMessageOfTheDay")
    .toChannel(this)
    .receiveAsFlow()
    .map { EventBusServiceRequestImpl<Unit, MessageOfTheDayService.Result>(it) }

Using these two bits of code, we can avoid all of the problems that were mentioned earlier. Let's rewrite our initial example using the generated code:

vertx.getMessageOfTheDayRequests
  .onEach { (_, reply) -> reply(MessageOfTheDayService.Success("Hello World")) }
  .launchIn(scope)

val motdService = MessageOfTheDayServiceImpl(vertx)
val motd = motdService.getMessageOfTheDay()

when (motd) {
  is MessageOfTheDayService.Failure -> motd.cause.printStackTrace()
  is MessageOfTheDayService.Success -> println(motd.message)
}

That was easy, right? We wrote simpler, more idiomatic code while gaining type safety for free. If you're interested in finding out more or trying this yourself, feel free to check out the GitHub repository for the project.

Writing Vert.x Integration Tests with Kotlin & Testcontainers

Selim Dincer — Tue, 04 May 2021 07:29:39 +0000

Testcontainers are widespread in integration testing where mocking external services such as databases may not be desirable. In this article, we will explore how to use Testcontainers in our Verticle tests.

In the previous article we built a helper class that served as a starting point for all Verticle tests. For our integration tests, we will define a new class. In this example we will be using a Postgres database.

private const val DB_NAME = "integration"
private const val DB_USERNAME = "integration"
private const val DB_PASSWORD = "integration"

@Testcontainers
@ExtendWith(VertxExtension::class)
abstract class IntegrationTest(
  private val vertx: Vertx,
  private val verticleClass: KClass<out Verticle>,
  private val deploymentOptions: DeploymentOptions = DeploymentOptions()
) : AsyncTest(vertx, verticleClass, deploymentOptions) {
  companion object {
    @Container
    private var postgresqlContainer: PostgreSQLContainer<Nothing> = PostgreSQLContainer<Nothing>("postgres:12")
      .apply {
        withDatabaseName(DB_NAME)
        withUsername(DB_USERNAME)
        withPassword(DB_PASSWORD)
      }
  }

  override fun deployVerticle(): String = runBlocking(vertx.dispatcher()) {
    val dbUri = postgresqlContainer.jdbcUrl.substringAfter("jdbc:")
    deploymentOptions.config = jsonObjectOf(
      "db.uri" to dbUri,
      "db.username" to DB_USERNAME,
      "db.password" to DB_PASSWORD,
    )
    vertx.deployVerticle(verticleClass.java, deploymentOptions).await()
  }
}

The @Testcontainers and @Container annotations will let junit take care of the container lifecycle for us. For more details check the official docs.

By overriding the deployVerticle function, we can update the Verticle config and insert our database credentials for the test. Since the Verticle in our example uses the reactive Postgres client, we need to remove the jdbc: prefix of the URI. The Verticle can access this config as follows:

private val pool: Pool by lazy {
  val connectOptions = PgConnectOptions.fromUri(config["db.uri"])
    .setUser(config["db.username"])
    .setPassword(config["db.password"])
  PgPool.pool(vertx, connectOptions, PoolOptions())
}

This means that our Verticle will now be using our testcontainer for all its queries. We can now purely focus on writing tests. The following test will automatically start a database, deploy our Verticle and call an Http endpoint:

@ExtendWith(VertxExtension::class)
class HealthTest(vertx: Vertx) : IntegrationTest(vertx, MainVerticle::class) {
  @Test
  fun dbIsUp() = runTest {
    val request = httpClient.request(HttpMethod.GET, 8080, "localhost", "/db-health").await()
    val response = request.send().await()
    val responseJson = response.body().await().toJsonObject()

    Assertions.assertEquals("up", responseJson.getString("postgres"))
  }
}

Summary

In the last article of the series, we looked at how to test a Vert.x application against a real database using Testcontainers. We were able to hide the complexity of the database lifecycle and Verticle deployment behind a simple helper class.

Source

A full project can be found in this GitHub repo.

Reducing Boilerplate in Vert.x Tests written in Kotlin

Selim Dincer — Tue, 04 May 2021 07:27:42 +0000

In the previous part, we saw how Kotlin can reduce noise in asynchronous tests. In this part we will introduce some syntactic sugar that will help us avoid repetition in our Verticle tests.

When testing a Verticle, we need to tell junit to deploy the Verticle before the test, and undeploy it afterwards. This is fairly straightforward but it can be tedious to add to every single test file:

  private lateinit var deploymentId: String

  @BeforeEach
  fun deployVerticle(): Unit = runBlocking(vertx.dispatcher()) {
    deploymentId = vertx.deployVerticle(verticleClass.java, DeploymentOptions()).await()
  }

  @AfterEach
  fun undeployVerticle(): Unit = runBlocking(vertx.dispatcher()) {
    vertx.undeploy(deploymentId).await()
  }

There's a very simple way we can avoid this boilerplate. If we move this into an abstract class that does this once, we can simply extend it and focus on our tests instead.

We should make sure that this is flexible enough so in this case we want some configuration options. We should be able to:

Specify which Verticle we would like to test
Pass configuration to this Verticle such as a database URI or a temporary directory
Customize the deployment if necessary
Run test code on Vert.x threads

So let's define this helper class:

abstract class AsyncTest(
  private val vertx: Vertx,
  private val verticleClass: KClass<out Verticle>,
  private val deploymentOptions: DeploymentOptions = DeploymentOptions()
) {
  private lateinit var deploymentId: String

  open fun deployVerticle(): String = runBlocking(vertx.dispatcher()) {
    vertx.deployVerticle(verticleClass.java, deploymentOptions).await()
  }

  @BeforeEach
  fun assignDeploymentId() {
    deploymentId = deployVerticle()
  }

  @AfterEach
  fun undeployVerticle(): Unit = runBlocking(vertx.dispatcher()) {
    vertx.undeploy(deploymentId).await()
  }

  protected fun runTest(block: suspend () -> Unit): Unit = runBlocking(vertx.dispatcher()) {
    block()
  }
}

If we have things like Http clients, we can also put them in here and make them accessible to all tests. Now, when we want to test a Verticle, we simply extend this class add tests:

@ExtendWith(VertxExtension::class)
class TestMainVerticle(vertx: Vertx) : AsyncTest(vertx, MainVerticle::class) {

  @Test
  fun `service is healthy when deployed`() = runTest {
    val request = httpClient.request(HttpMethod.GET, 8080, "localhost", "/health").await()
    val response = request.send().await()
    val responseJson = response.body().await().toJsonObject()

    assertEquals("up", responseJson.getString("status"))
  }
}

Our helper class will make sure that:

The verticle is deployed & undeployed for each test
Tests wrapped in runTest run on a Vert.x thread automatically

Summary

In this part, we looked at how to further reduce boilerplate in our asynchronous tests. In the next part, we will see how we can combine everything we've learned so far in order to write integration tests using Testcontainers.

Source

A full project can be found in this GitHub repo.

Writing Async Tests for Vert.x using Kotlin

Selim Dincer — Tue, 04 May 2021 07:27:21 +0000

Testing asynchronous code can be a challenging task. Usually some tooling is required to make it work with testing frameworks like junit.

For Vert.x projects, there's an official junit module to help writing these tests.
The VertxTestContext in that module is comparable to a standard CountdownLatch. The testing framework waits for the latch to unlock and a callback does the unlocking.

Unfortunately this leads to a quite a bit of boilerplate code for simple unit tests. Luckily though, Kotlin's coroutines provide a clean solution to write tests in a more traditional-looking way.

In this article, we're going to explore how this can be achieved by migrating a simple example to use Kotlin's coroutines.

Here's a slightly modified version of the first test:

@Test
@Timeout(value = 5, timeUnit = TimeUnit.SECONDS)
void service_is_healthy(Vertx vertx, VertxTestContext testContext) {
  HttpClient httpClient = vertx.createHttpClient();

  httpClient.request(HttpMethod.GET, 8080, "localhost", "/health")
    .compose(HttpClientRequest::send)
    .compose(HttpClientResponse::body)
    .onComplete(testContext.succeeding(buffer -> {
      JsonObject responseJson = buffer.toJsonObject();

      Assertions.assertEquals("up", responseJson.getString("status"));
      testContext.completeNow();
    }));
}

Now, what are the problems with this code? Notice how the the test context clutters the test code. Wouldn't it be nicer if we didn't have to worry about it?

Let's rewrite this using Kotlin coroutines:

@Test
@Timeout(5, unit = TimeUnit.SECONDS)
fun `service is healthy`(vertx: Vertx): Unit = runBlocking(vertx.dispatcher()) {
  val httpClient = vertx.createHttpClient()

  val request = httpClient.request(HttpMethod.GET, 8080, "localhost", "/health").await()
  val response = request.send().await()
  val responseJson = response.body().await().toJsonObject()

  Assertions.assertEquals("up", responseJson.getString("status"))
}

As you can see, we got rid of almost everything that's unrelated to the actual test. This is the main benefit of using coroutines for tests with Vert.x.

Let's go over the changes that we introduced here. First, we wrapped the test body in runBlocking in order to be able to start coroutines inside the test.

Note: It's important that the coroutines run on the Vert.x threads. Luckily, Vert.x provides a #dispatcher function that we can use for runBlocking.

The #await calls make sure that exceptions make the tests fail. They also force junit to wait for our test, esentially making the VertxTestContext obsolete for our tests.

On top of that, we avoid callbacks and nesting. This makes our tests a little more readable.

Summary

In this short article, we saw that Kotlin's coroutines can vastly reduce the noise in asynchronous tests. Instead of fighting with the asynchronous nature of our code, we can focus on the tests themselves. In the next part, we will see how we can add some helpers to make writing tests even easier.

Source

A full project can be found in this GitHub repo.

Supercharge Your Kotlin Vert.x Application with EventBus Extensions

Selim Dincer — Thu, 12 Nov 2020 18:01:18 +0000

If you have ever worked with the Vert.x EventBus in Kotlin, you might have wondered about ways to make it more type-safe. In this article, I'd like to introduce a tiny library that aims to solve this exact problem. It provides an idiomatic and type-safe way of communicating via the EventBus and in addition avoids unnecessary serialization and deserialization.

Getting Started

Imagine we need to create a division service. We want to give it two doubles and expect it to reply with the result or an error if the operation failed.

If we were to model the input and output for this service in Kotlin we might come up with something like this:

data class DivisionRequest(val dividend: Double, val divisor: Double)

sealed class Division {
  data class Success(val quotient: Double) : Division()
  data class Error(val message: String) : Division()
}

With the data for our service ready, we can start generating the EventBus extensions!

Describing the Service

For the code generation to work, we need to annotate our service Verticle:

@EventBusService(
    topic = "co.selim.sample.division",
    consumes = DivisionRequest::class,
    produces = Division::class,
    propertyName = "divisionRequests",
    functionName = "divide"
)
class DivisionVerticle : CoroutineVerticle() {

Now the library has all it needs to generate type-safe code to interact with our service.

Communicating with the Service

The library will generate two extensions on the Vertx class:

An extension function to call the service:

suspend fun Vertx.divide(request: DivisionRequest): Division

An extension property with the requests that the service can handle:

val Vertx.divisionRequests: Flow<EventBusServiceRequest<DivisionRequest, Division>>

With these two extensions we can communicate with our service while not having to worry about sending in or receiving the wrong type.

Avoiding Serialization

The generated extensions avoid unnecessary serialization and deserialization by making use of a special EventBus codec. This provides better performance and enables us to send and receive Kotlin types.

Using it in Your Project

The code is hosted on GitHub and instructions for adding it to your dependencies can be found in the readme.