Vert.x #

Vert.x adalah toolkit untuk membangun aplikasi reaktif dan event-driven di JVM. Berbeda dari framework konvensional yang berbasis thread-per-request, Vert.x menggunakan model event loop — satu thread menangani ribuan koneksi secara non-blocking, mirip dengan Node.js tapi di JVM. Ini menjadikan Vert.x sangat efisien untuk use case dengan koneksi simultan dalam jumlah sangat besar — API gateway, proxy, atau backend game real-time. Vert.x mendukung Kotlin secara kelas satu lewat ekstensi dan coroutine. Artikel ini membahas arsitektur Vert.x, membangun HTTP server, event bus untuk komunikasi antar komponen, akses database reaktif, dan integrasi coroutine.

Arsitektur Vert.x — 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(thread terpisah)"]
    WP --> EL

Konsep	Penjelasan
Event Loop	Thread non-blocking yang memproses event secara bergantian
Verticle	Unit deployment Vert.x, setara dengan microservice kecil
Event Bus	Message bus internal untuk komunikasi antar Verticle
Handler	Fungsi callback yang dipanggil ketika event terjadi
Future/Promise	Abstraksi untuk operasi asinkron
Worker Verticle	Verticle yang berjalan di thread pool, untuk operasi blocking

Aturan terpenting: jangan pernah memblokir event loop thread. Operasi blocking (I/O file, query database synchronous, Thread.sleep()) harus dijalankan di Worker Pool atau menggunakan library reaktif.

Kapan Menggunakan Vert.x #

PILIH Vert.x jika:
  ✓ Butuh throughput sangat tinggi (ribuan koneksi simultan)
  ✓ API gateway atau reverse proxy
  ✓ Real-time application (game server, trading platform)
  ✓ Ingin toolkit yang fleksibel (bukan opinionated framework)
  ✓ Tim sudah familiar dengan model reaktif (RxJava, Reactor)

PERTIMBANGKAN lain jika:
  ✗ Tim baru dengan pemrograman reaktif — kurva belajar tinggi
  ✗ Aplikasi CRUD sederhana — Spring Boot atau Ktor lebih produktif
  ✗ Butuh ekosistem library yang sangat kaya — Spring lebih baik

Setup dan Dependensi #

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

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

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

    // Database reaktif
    implementation("io.vertx:vertx-pg-client:$vertxVersion")          // PostgreSQL reaktif
    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 menggunakan Jackson secara internal)
    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")
}

Verticle — Unit Deployment #

Verticle adalah unit utama kode dalam Vert.x. Setiap Verticle berjalan di 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>) {
        // Semua kode di sini berjalan di event loop
        // JANGAN blokir di sini!

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

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

    override fun stop(stopPromise: Promise<Void>) {
        println("Verticle berhenti")
        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 berhasil di-deploy: ${result.result()}")
        } else {
            println("Gagal deploy: ${result.cause().message}")
        }
    }
}

HTTP Server dan Router #

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)

    // Middleware global
    router.route().handler(BodyHandler.create())  // parse 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 tidak ditemukan").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", "Terjadi kesalahan internal").encode())
    }

    return router
}

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

        // Data placeholder
        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 produk 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", "ID tidak valid").encode()
            )
            return@handler
        }

        // Simulasi: produk tidak ditemukan jika id > 100
        if (id > 100) {
            ctx.response().setStatusCode(404).end(
                JsonObject().put("error", "Produk $id tidak ditemukan").encode()
            )
            return@handler
        }

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

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

        val nama = body.getString("nama")
        if (nama.isNullOrBlank()) {
            ctx.response().setStatusCode(400).end(
                JsonObject().put("error", "Nama tidak boleh kosong").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 produk
    router.put("/produk/:id").handler { ctx ->
        val id = ctx.pathParam("id").toIntOrNull()
            ?: return@handler ctx.response().setStatusCode(400).end("ID tidak valid")

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

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

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

Event Bus — Komunikasi Antar Verticle #

Event Bus adalah sistem messaging internal Vert.x yang memungkinkan Verticle berkomunikasi tanpa tight coupling:

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

// Verticle pengirim
class ProducerVerticle : AbstractVerticle() {
    override fun start() {
        // Kirim pesan setiap 2 detik
        vertx.setPeriodic(2000) {
            val pesan = JsonObject()
                .put("aksi", "REFRESH_CACHE")
                .put("waktu", System.currentTimeMillis())

            // Kirim tanpa menunggu respons (fire-and-forget)
            vertx.eventBus().publish("cache.events", pesan)

            // Kirim dan tunggu respons (request-reply)
            vertx.eventBus().request<JsonObject>("data.service", JsonObject().put("id", 42)) { reply ->
                if (reply.succeeded()) {
                    println("Respons diterima: ${reply.result().body()}")
                } else {
                    println("Gagal: ${reply.cause().message}")
                }
            }
        }
    }
}

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

        // Subscribe ke topic — bisa diterima semua subscriber (publish)
        eventBus.consumer<JsonObject>("cache.events") { pesan: Message<JsonObject> ->
            val aksi = pesan.body().getString("aksi")
            println("Event diterima: $aksi")
        }

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

            // Kirim respons balik
            pesan.reply(JsonObject().put("id", id).put("nama", "Data $id").put("berhasil", true))
        }
    }
}

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

    // Deploy semua verticle
    vertx.deployVerticle(ConsumerVerticle())
    vertx.deployVerticle(ProducerVerticle())
}

Coroutine dengan vertx-lang-kotlin #

Vert.x menyediakan ekstensi coroutine yang mengubah callback API menjadi fungsi suspend:

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() {
        // Buat server — coAwait mengubah callback menjadi suspend
        val server = vertx.createHttpServer()
        val router = Router.router(vertx)

        router.get("/async").handler { ctx ->
            // Launch coroutine di dalam 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 hingga server siap
        println("Coroutine server berjalan di port 8080")
    }

    private suspend fun operasiAsync(): String {
        // Simulasi operasi asinkron
        kotlinx.coroutines.delay(100)
        return "selesai dalam coroutine"
    }

    // Database reaktif dengan coroutine
    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"))
        }
    }
}

Database Reaktif — PostgreSQL Client #

Vert.x menyediakan reactive SQL client yang non-blocking:

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

fun buatPgPool(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 dengan 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 gagal: ${result.cause().message}")
                handler(emptyList())
            }
        }
}

// Query dengan parameter (mencegah 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())
        }
    }
}

// Integrasi dalam 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("Nama wajib diisi")
        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())
            }
        )
    }
}

WebSocket di Vert.x #

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

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

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

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

// Server-side event dengan 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 ke event bus
        val consumer = vertx.eventBus().consumer<String>("sse.events") { msg ->
            response.write("data: ${msg.body()}\n\n")
        }

        // Cleanup saat koneksi ditutup
        response.closeHandler {
            consumer.unregister()
        }
    }
}

Testing dengan 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 mengembalikan 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 produk mengembalikan list`(testContext: VertxTestContext) {
        client.get(8080, "localhost", "/api/v1/produk")
            .send { result ->
                testContext.verify {
                    val respons = result.result()
                    assert(respons.statusCode() == 200)
                    val body = respons.bodyAsJsonObject()
                    assert(body.containsKey("data"))
                }
                testContext.completeNow()
            }
    }

    @Test
    fun `POST produk baru mengembalikan 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 Breaker — Resiliensi #

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

val circuitBreaker = CircuitBreaker.create("layanan-eksternal", vertx,
    CircuitBreakerOptions()
        .setMaxFailures(5)           // buka circuit setelah 5 kegagalan
        .setTimeout(2000)            // timeout per operasi: 2 detik
        .setResetTimeout(10000)      // coba tutup circuit setelah 10 detik
        .setFallbackOnFailure(true)  // panggil fallback saat gagal
)

fun panggilLayananEksternal(url: String, ctx: RoutingContext) {
    circuitBreaker.execute<JsonObject> { promise ->
        // Operasi yang mungkin gagal
        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: kembalikan data default
            ctx.response().setStatusCode(503).end(
                JsonObject().put("error", "Layanan sementara tidak tersedia").encode()
            )
        }
    }
}

Ringkasan #

Jangan pernah memblokir event loop — ini adalah aturan terpenting Vert.x. Operasi blocking (I/O synchronous, Thread.sleep, query JDBC) harus dijalankan di Worker Verticle atau menggunakan library reaktif.
Verticle sebagai unit isolasi — setiap Verticle memiliki event loop sendiri. Gunakan ini untuk memisahkan tanggung jawab: satu Verticle untuk HTTP, satu untuk database, satu untuk cache.
Event Bus untuk decoupling — publish untuk broadcast (banyak receiver), send untuk unicast, request untuk request-reply. Event Bus bisa di-cluster untuk komunikasi antar node.
Gunakan coAwait() untuk coroutine — ekstensi coAwait() mengubah callback API Vert.x menjadi fungsi suspend. Gunakan CoroutineVerticle sebagai base class untuk coroutine-friendly Verticle.
Reactive SQL Client untuk database — vertx-pg-client, vertx-mysql-client adalah driver non-blocking yang tidak memblokir event loop. Hindari JDBC biasa di dalam Verticle standar.
Router untuk HTTP — Router.router(vertx) menyediakan routing yang kaya: path params, query params, middleware (BodyHandler, CorsHandler, AuthenticationHandler), sub-router.
Circuit Breaker untuk resiliensi — bungkus panggilan ke layanan eksternal dengan Circuit Breaker. Ini mencegah cascade failure saat layanan eksternal lambat atau tidak responsif.
Vert.x untuk throughput tinggi — satu instance Vert.x bisa menangani puluhan ribu koneksi simultan dengan resource yang jauh lebih kecil dari model thread-per-request tradisional.

← Sebelumnya: Ktor Berikutnya: Kotlinx.html →