Vert.x #

Vert.x is a toolkit for building reactive and event-driven applications on the JVM. Unlike conventional frameworks based on thread-per-request, Vert.x uses the event loop model — one thread handles thousands of connections non-blocking, similar to Node.js but on the JVM. This makes Vert.x extremely efficient for use cases with very large numbers of simultaneous connections — API gateways, proxies, or real-time game backends. Vert.x supports Kotlin as a first-class citizen through extensions and coroutines. This article covers Vert.x’s architecture, building an HTTP server, the event bus for inter-component communication, reactive database access, and coroutine integration.

Vert.x Architecture — The Event Loop #

flowchart LR
    R1[Request 1] --> EL["Event Loop Thread\n(non-blocking)"]
    R2[Request 2] --> EL
    R3[Request 3] --> EL
    EL --> H1[Handler 1]
    EL --> H2[Handler 2]
    EL --> H3[Handler 3]
    H1 --> |"Blocking I/O\n(DB, file)"| WP["Worker Pool\n(separate thread)"]
    WP --> EL

Concept	Explanation
Event Loop	A non-blocking thread that processes events in turn
Verticle	Vert.x’s deployment unit, equivalent to a small microservice
Event Bus	An internal message bus for communication between Verticles
Handler	A callback function invoked when an event occurs
Future/Promise	An abstraction for asynchronous operations
Worker Verticle	A Verticle running on a thread pool, for blocking operations

The most important rule: never block the event loop thread. Blocking operations (file I/O, synchronous database queries, Thread.sleep()) must run on a Worker Pool or use reactive libraries.

When to Use Vert.x #

CHOOSE Vert.x if:
  ✓ You need very high throughput (thousands of simultaneous connections)
  ✓ API gateways or reverse proxies
  ✓ Real-time applications (game servers, trading platforms)
  ✓ You want a flexible toolkit (not an opinionated framework)
  ✓ The team is already familiar with the reactive model (RxJava, Reactor)

CONSIDER something else if:
  ✗ A new team with reactive programming — steep learning curve
  ✗ Simple CRUD applications — Spring Boot or Ktor are more productive
  ✗ You need a very rich library ecosystem — Spring is better

Setup and Dependencies #

// build.gradle.kts
val vertxVersion = "4.5.7"

dependencies {
    // Core
    implementation("io.vertx:vertx-core:$vertxVersion")
    implementation("io.vertx:vertx-web:$vertxVersion")

    // Kotlin extensions and coroutine support
    implementation("io.vertx:vertx-lang-kotlin:$vertxVersion")
    implementation("io.vertx:vertx-lang-kotlin-coroutines:$vertxVersion")

    // Reactive database
    implementation("io.vertx:vertx-pg-client:$vertxVersion")          // reactive PostgreSQL
    implementation("io.vertx:vertx-sql-client-templates:$vertxVersion")

    // Event bus codecs
    implementation("io.vertx:vertx-web-client:$vertxVersion")          // HTTP client
    implementation("io.vertx:vertx-circuit-breaker:$vertxVersion")     // circuit breaker

    // JSON (Vert.x uses Jackson internally)
    implementation("io.vertx:vertx-json-schema:$vertxVersion")

    // Testing
    testImplementation("io.vertx:vertx-junit5:$vertxVersion")
    testImplementation("io.vertx:vertx-web-client:$vertxVersion")
    testImplementation("org.junit.jupiter:junit-jupiter:5.10.2")
}

Verticles — Deployment Units #

A Verticle is Vert.x’s main unit of code. Each Verticle runs on an event loop:

import io.vertx.core.AbstractVerticle
import io.vertx.core.Promise
import io.vertx.core.Vertx

class ServerVerticle : AbstractVerticle() {

    override fun start(startPromise: Promise<Void>) {
        // All code here runs on the event loop
        // DON'T block here!

        val server = vertx.createHttpServer()
        val router = createRouter()

        server.requestHandler(router)
            .listen(8080) { result ->
                if (result.succeeded()) {
                    println("Server running on port 8080")
                    startPromise.complete()
                } else {
                    startPromise.fail(result.cause())
                }
            }
    }

    override fun stop(stopPromise: Promise<Void>) {
        println("Verticle stopping")
        stopPromise.complete()
    }

    private fun createRouter() = io.vertx.ext.web.Router.router(vertx)
}

fun main() {
    val vertx = Vertx.vertx()

    vertx.deployVerticle(ServerVerticle()) { result ->
        if (result.succeeded()) {
            println("Verticle deployed successfully: ${result.result()}")
        } else {
            println("Deploy failed: ${result.cause().message}")
        }
    }
}

HTTP Servers and Routers #

import io.vertx.core.json.JsonObject
import io.vertx.ext.web.Router
import io.vertx.ext.web.RoutingContext
import io.vertx.ext.web.handler.BodyHandler

fun Application.buatRouter(vertx: io.vertx.core.Vertx): Router {
    val router = Router.router(vertx)

    // Global middleware
    router.route().handler(BodyHandler.create())  // parse the request body
    router.route().handler(LoggerHandler.create()) // logging

    // CORS
    router.route().handler(
        io.vertx.ext.web.handler.CorsHandler.create()
            .allowedMethod(io.vertx.core.http.HttpMethod.GET)
            .allowedMethod(io.vertx.core.http.HttpMethod.POST)
            .allowedMethod(io.vertx.core.http.HttpMethod.PUT)
            .allowedMethod(io.vertx.core.http.HttpMethod.DELETE)
            .allowedHeader("Content-Type")
            .allowedHeader("Authorization")
    )

    // Health check
    router.get("/health").handler { ctx ->
        ctx.json(JsonObject().put("status", "ok").put("waktu", System.currentTimeMillis()))
    }

    // API routes
    val apiRouter = Router.router(vertx)
    daftarkanProdukRoute(apiRouter, vertx)

    router.mountSubRouter("/api/v1", apiRouter)

    // Error handler
    router.errorHandler(404) { ctx ->
        ctx.response()
            .setStatusCode(404)
            .putHeader("Content-Type", "application/json")
            .end(JsonObject().put("error", "Endpoint not found").encode())
    }

    router.errorHandler(500) { ctx ->
        ctx.failure()?.let { println("Error: ${it.message}") }
        ctx.response()
            .setStatusCode(500)
            .putHeader("Content-Type", "application/json")
            .end(JsonObject().put("error", "Internal server error").encode())
    }

    return router
}

fun daftarkanProdukRoute(router: Router, vertx: io.vertx.core.Vertx) {
    // GET all products
    router.get("/produk").handler { ctx ->
        val halaman = ctx.queryParam("halaman").firstOrNull()?.toIntOrNull() ?: 1
        val ukuran = ctx.queryParam("ukuran").firstOrNull()?.toIntOrNull() ?: 20

        // Placeholder data
        val produk = listOf(
            JsonObject().put("id", 1).put("nama", "Laptop").put("harga", 15_000_000),
            JsonObject().put("id", 2).put("nama", "Mouse").put("harga", 250_000)
        )

        ctx.json(JsonObject()
            .put("data", io.vertx.core.json.JsonArray(produk))
            .put("halaman", halaman)
            .put("ukuran", ukuran)
        )
    }

    // GET a product by ID
    router.get("/produk/:id").handler { ctx ->
        val id = ctx.pathParam("id").toIntOrNull()
        if (id == null) {
            ctx.response().setStatusCode(400).end(
                JsonObject().put("error", "Invalid ID").encode()
            )
            return@handler
        }

        // Simulate: product not found if id > 100
        if (id > 100) {
            ctx.response().setStatusCode(404).end(
                JsonObject().put("error", "Produk $id not found").encode()
            )
            return@handler
        }

        ctx.json(JsonObject().put("id", id).put("nama", "Produk $id").put("harga", id * 10_000))
    }

    // POST a new product
    router.post("/produk").handler { ctx ->
        val body = runCatching { ctx.body().asJsonObject() }.getOrElse {
            ctx.response().setStatusCode(400).end(
                JsonObject().put("error", "Body is not valid JSON").encode()
            )
            return@handler
        }

        val nama = body.getString("nama")
        if (nama.isNullOrBlank()) {
            ctx.response().setStatusCode(400).end(
                JsonObject().put("error", "Name must not be empty").encode()
            )
            return@handler
        }

        val produkBaru = body.copy().put("id", (Math.random() * 1000).toInt())
        ctx.response()
            .setStatusCode(201)
            .putHeader("Content-Type", "application/json")
            .end(produkBaru.encode())
    }

    // PUT update a product
    router.put("/produk/:id").handler { ctx ->
        val id = ctx.pathParam("id").toIntOrNull()
            ?: return@handler ctx.response().setStatusCode(400).end("Invalid ID")

        val body = ctx.body().asJsonObject()
        val diperbarui = body.copy().put("id", id)
        ctx.json(diperbarui)
    }

    // DELETE a product
    router.delete("/produk/:id").handler { ctx ->
        val id = ctx.pathParam("id")
        ctx.response().setStatusCode(204).end()
    }
}

// Extension for ctx.json()
fun RoutingContext.json(obj: JsonObject) {
    response()
        .putHeader("Content-Type", "application/json")
        .end(obj.encode())
}

The Event Bus — Communication Between Verticles #

The Event Bus is Vert.x’s internal messaging system that lets Verticles communicate without tight coupling:

import io.vertx.core.AbstractVerticle
import io.vertx.core.eventbus.Message

// The sending Verticle
class ProducerVerticle : AbstractVerticle() {
    override fun start() {
        // Send a message every 2 seconds
        vertx.setPeriodic(2000) {
            val pesan = JsonObject()
                .put("aksi", "REFRESH_CACHE")
                .put("waktu", System.currentTimeMillis())

            // Send without waiting for a response (fire-and-forget)
            vertx.eventBus().publish("cache.events", pesan)

            // Send and wait for a response (request-reply)
            vertx.eventBus().request<JsonObject>("data.service", JsonObject().put("id", 42)) { reply ->
                if (reply.succeeded()) {
                    println("Response received: ${reply.result().body()}")
                } else {
                    println("Failed: ${reply.cause().message}")
                }
            }
        }
    }
}

// The receiving Verticle
class ConsumerVerticle : AbstractVerticle() {
    override fun start() {
        val eventBus = vertx.eventBus()

        // Subscribe to a topic — all subscribers receive it (publish)
        eventBus.consumer<JsonObject>("cache.events") { pesan: Message<JsonObject> ->
            val aksi = pesan.body().getString("aksi")
            println("Event received: $aksi")
        }

        // Handler for request-reply
        eventBus.consumer<JsonObject>("data.service") { pesan: Message<JsonObject> ->
            val id = pesan.body().getInteger("id")
            println("Request received for ID: $id")

            // Send the reply back
            pesan.reply(JsonObject().put("id", id).put("nama", "Data $id").put("berhasil", true))
        }
    }
}

fun main() {
    val vertx = Vertx.vertx()

    // Deploy all verticles
    vertx.deployVerticle(ConsumerVerticle())
    vertx.deployVerticle(ProducerVerticle())
}

Coroutines with vertx-lang-kotlin #

Vert.x provides coroutine extensions that turn callback APIs into suspend functions:

import io.vertx.kotlin.coroutines.CoroutineVerticle
import io.vertx.kotlin.coroutines.coAwait
import io.vertx.kotlin.coroutines.dispatcher

class CoroutineVerticleContoh : CoroutineVerticle() {

    override suspend fun start() {
        // Create the server — coAwait turns callbacks into suspend
        val server = vertx.createHttpServer()
        val router = Router.router(vertx)

        router.get("/async").handler { ctx ->
            // Launch a coroutine inside the handler
            io.vertx.kotlin.coroutines.launch(vertx.dispatcher()) {
                val hasil = operasiAsync()
                ctx.json(JsonObject().put("hasil", hasil))
            }
        }

        server.requestHandler(router)
        server.listen(8080).coAwait()  // suspend until the server is ready
        println("Coroutine server running on port 8080")
    }

    private suspend fun operasiAsync(): String {
        // Simulate an async operation
        kotlinx.coroutines.delay(100)
        return "done in coroutine"
    }

    // Reactive database with coroutines
    suspend fun queryDenganCoroutine(pool: io.vertx.pgclient.PgPool): List<JsonObject> {
        val result = pool.query("SELECT * FROM produk WHERE aktif = true").execute().coAwait()
        return result.map { row ->
            JsonObject()
                .put("id", row.getLong("id"))
                .put("nama", row.getString("nama"))
                .put("harga", row.getDouble("harga"))
        }
    }
}

Reactive Database — The PostgreSQL Client #

Vert.x provides a non-blocking reactive SQL client:

import io.vertx.pgclient.PgConnectOptions
import io.vertx.pgclient.PgPool
import io.vertx.sqlclient.PoolOptions
import io.vertx.sqlclient.Tuple

fun createPgPool(vertx: io.vertx.core.Vertx): PgPool {
    val connectOptions = PgConnectOptions().apply {
        host = System.getenv("DB_HOST") ?: "localhost"
        port = System.getenv("DB_PORT")?.toInt() ?: 5432
        database = System.getenv("DB_NAME") ?: "myapp"
        user = System.getenv("DB_USER") ?: "postgres"
        password = System.getenv("DB_PASSWORD") ?: "postgres"
    }

    val poolOptions = PoolOptions().apply {
        maxSize = 10
    }

    return PgPool.pool(vertx, connectOptions, poolOptions)
}

// Query with the callback API
fun queryProduk(pool: PgPool, handler: (List<JsonObject>) -> Unit) {
    pool.query("SELECT id, nama, harga, stok FROM produk WHERE aktif = true")
        .execute { result ->
            if (result.succeeded()) {
                val produk = result.result().map { row ->
                    JsonObject()
                        .put("id", row.getLong("id"))
                        .put("nama", row.getString("nama"))
                        .put("harga", row.getDouble("harga"))
                        .put("stok", row.getInteger("stok"))
                }
                handler(produk)
            } else {
                println("Query failed: ${result.cause().message}")
                handler(emptyList())
            }
        }
}

// Query with parameters (prevents SQL injection)
fun simpanProduk(pool: PgPool, nama: String, harga: Double, stok: Int,
                  onSuccess: (Long) -> Unit, onError: (Throwable) -> Unit) {
    pool.preparedQuery(
        "INSERT INTO produk (nama, harga, stok) VALUES ($1, $2, $3) RETURNING id"
    ).execute(Tuple.of(nama, harga, stok)) { result ->
        if (result.succeeded()) {
            val id = result.result().first().getLong("id")
            onSuccess(id)
        } else {
            onError(result.cause())
        }
    }
}

// Integration in a route
fun daftarkanProdukRouteWithDb(router: Router, pool: PgPool) {
    router.get("/produk").handler { ctx ->
        queryProduk(pool) { produk ->
            ctx.response()
                .putHeader("Content-Type", "application/json")
                .end(io.vertx.core.json.JsonArray(produk).encode())
        }
    }

    router.post("/produk").handler { ctx ->
        val body = ctx.body().asJsonObject()
        val nama = body.getString("nama") ?: return@handler ctx.response()
            .setStatusCode(400).end("Name is required")
        val harga = body.getDouble("harga") ?: 0.0
        val stok = body.getInteger("stok") ?: 0

        simpanProduk(pool, nama, harga, stok,
            onSuccess = { id ->
                ctx.response().setStatusCode(201)
                    .putHeader("Content-Type", "application/json")
                    .end(body.put("id", id).encode())
            },
            onError = { err ->
                ctx.response().setStatusCode(500)
                    .end(JsonObject().put("error", err.message).encode())
            }
        )
    }
}

WebSockets in Vert.x #

fun setupWebSocket(router: Router) {
    // WebSocket handler
    router.route("/ws").handler { ctx ->
        val ws = ctx.request().upgrade()  // upgrade HTTP to WebSocket

        ws.textMessageHandler { pesan ->
            println("Received: $pesan")
            ws.writeTextMessage("Echo: $pesan")  // echo back
        }

        ws.closeHandler {
            println("WebSocket closed")
        }

        ws.exceptionHandler { err ->
            println("WebSocket error: ${err.message}")
        }
    }
}

// Server-side events with the event bus
fun setupSse(router: Router, vertx: io.vertx.core.Vertx) {
    router.get("/sse").handler { ctx ->
        val response = ctx.response().apply {
            putHeader("Content-Type", "text/event-stream")
            putHeader("Cache-Control", "no-cache")
            putHeader("Connection", "keep-alive")
            isChunked = true
        }

        // Subscribe to the event bus
        val consumer = vertx.eventBus().consumer<String>("sse.events") { msg ->
            response.write("data: ${msg.body()}\n\n")
        }

        // Cleanup when the connection closes
        response.closeHandler {
            consumer.unregister()
        }
    }
}

Testing with Vert.x JUnit 5 #

import io.vertx.core.Vertx
import io.vertx.ext.web.client.WebClient
import io.vertx.junit5.VertxExtension
import io.vertx.junit5.VertxTestContext
import org.junit.jupiter.api.AfterEach
import org.junit.jupiter.api.BeforeEach
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.extension.ExtendWith

@ExtendWith(VertxExtension::class)
class ServerVerticleTest {

    private lateinit var client: WebClient

    @BeforeEach
    fun setup(vertx: Vertx, testContext: VertxTestContext) {
        client = WebClient.create(vertx)

        vertx.deployVerticle(ServerVerticle()) { result ->
            if (result.succeeded()) testContext.completeNow()
            else testContext.failNow(result.cause())
        }
    }

    @AfterEach
    fun teardown(vertx: Vertx) {
        client.close()
    }

    @Test
    fun `GET health returns 200`(vertx: Vertx, testContext: VertxTestContext) {
        client.get(8080, "localhost", "/health")
            .send { result ->
                testContext.verify {
                    assert(result.succeeded())
                    assert(result.result().statusCode() == 200)
                }
                testContext.completeNow()
            }
    }

    @Test
    fun `GET products returns a list`(testContext: VertxTestContext) {
        client.get(8080, "localhost", "/api/v1/produk")
            .send { result ->
                testContext.verify {
                    val response = result.result()
                    assert(response.statusCode() == 200)
                    val body = response.bodyAsJsonObject()
                    assert(body.containsKey("data"))
                }
                testContext.completeNow()
            }
    }

    @Test
    fun `POST a new product returns 201`(testContext: VertxTestContext) {
        val payload = JsonObject()
            .put("nama", "Laptop Test")
            .put("harga", 15_000_000.0)
            .put("stok", 5)

        client.post(8080, "localhost", "/api/v1/produk")
            .putHeader("Content-Type", "application/json")
            .sendJsonObject(payload) { result ->
                testContext.verify {
                    assert(result.result().statusCode() == 201)
                }
                testContext.completeNow()
            }
    }
}

Circuit Breakers — Resilience #

import io.vertx.circuitbreaker.CircuitBreaker
import io.vertx.circuitbreaker.CircuitBreakerOptions

val circuitBreaker = CircuitBreaker.create("layanan-eksternal", vertx,
    CircuitBreakerOptions()
        .setMaxFailures(5)           // open the circuit after 5 failures
        .setTimeout(2000)            // per-operation timeout: 2 seconds
        .setResetTimeout(10000)      // try closing the circuit after 10 seconds
        .setFallbackOnFailure(true)  // call the fallback on failure
)

fun panggilLayananEksternal(url: String, ctx: RoutingContext) {
    circuitBreaker.execute<JsonObject> { promise ->
        // The operation that may fail
        webClient.getAbs(url).send { result ->
            if (result.succeeded()) {
                promise.complete(result.result().bodyAsJsonObject())
            } else {
                promise.fail(result.cause())
            }
        }
    }.onComplete { result ->
        if (result.succeeded()) {
            ctx.json(result.result())
        } else {
            // Fallback: return default data
            ctx.response().setStatusCode(503).end(
                JsonObject().put("error", "Service temporarily unavailable").encode()
            )
        }
    }
}

Summary #

Never block the event loop — this is Vert.x’s most important rule. Blocking operations (synchronous I/O, Thread.sleep, JDBC queries) must run in Worker Verticles or use reactive libraries.
Verticles as isolation units — each Verticle has its own event loop. Use this to separate responsibilities: one Verticle for HTTP, one for the database, one for caching.
The Event Bus for decoupling — publish for broadcast (many receivers), send for unicast, request for request-reply. The Event Bus can be clustered for inter-node communication.
Use coAwait() for coroutines — the coAwait() extension turns Vert.x callback APIs into suspend functions. Use CoroutineVerticle as the base class for coroutine-friendly Verticles.
Reactive SQL Clients for databases — vertx-pg-client, vertx-mysql-client are non-blocking drivers that don’t block the event loop. Avoid plain JDBC inside standard Verticles.
Routers for HTTP — Router.router(vertx) provides rich routing: path params, query params, middleware (BodyHandler, CorsHandler, AuthenticationHandler), sub-routers.
Circuit Breakers for resilience — wrap external service calls with a Circuit Breaker. This prevents cascade failures when an external service is slow or unresponsive.
Vert.x for high throughput — one Vert.x instance can handle tens of thousands of simultaneous connections with far fewer resources than the traditional thread-per-request model.

← Previous: Ktor Next: Kotlinx.html →