Bagian terakhir proyek ini fokus pada dua aspek yang sering diabaikan: pengujian dan pengukuran performa. Tanpa keduanya, kita tidak yakin pipeline bekerja benar dan tidak tahu apakah perubahan kode memperbaiki atau memperburuk kinerja. Bagian 4 mendemonstrasikan test harness sederhana untuk verifikasi output dan benchmark tiga pendekatan pipeline.

Tiga Pendekatan Pipeline #

PendekatanKarakteristikCocok untuk
List (eager)Setiap operasi membuat list baruKoleksi kecil, butuh semua nilai
Sequence (lazy)Single-pass, evaluasi bertahapKoleksi besar + operasi chained
Flow (suspend)Async, ada overhead suspend functionOperasi yang melibatkan I/O

Test Harness #

Berikut program yang menggabungkan unit test sederhana dengan benchmark:

kotlin 
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.measureNanoTime

// === Pipeline untuk diukur ===
data class Item(val id: Int, val value: Double)

fun generateItems(n: Int): List<Item> = (1..n).map { Item(it, it * 1.5) }

fun pipelineSync(items: List<Item>): List<Double> {
    return items.map { it.value * 2 }.filter { it > 10 }.map { it + 1 }
}

fun pipelineSequence(items: List<Item>): List<Double> {
    return items.asSequence()
        .map { it.value * 2 }
        .filter { it > 10 }
        .map { it + 1 }
        .toList()
}

fun pipelineFlow(items: List<Item>): List<Double> = runBlocking {
    items.asFlow()
        .map { it.value * 2 }
        .filter { it > 10 }
        .map { it + 1 }
        .toList()
}

// === Test harness: assert output ===
fun testPipeline() {
    val items = generateItems(10)
    val expected = items.map { it.value * 2 }.filter { it > 10 }.map { it + 1 }

    val sync = pipelineSync(items)
    val seq = pipelineSequence(items)
    val flow = pipelineFlow(items)

    check(sync == expected) { "sync mismatch: $sync vs $expected" }
    println("[LULUS] pipelineSync(10 items) = $sync")

    check(seq == expected) { "seq mismatch: $seq vs $expected" }
    println("[LULUS] pipelineSequence(10 items) = $seq")

    check(flow == expected) { "flow mismatch: $flow vs $expected" }
    println("[LULUS] pipelineFlow(10 items) = $flow")
}

// === Benchmark: ukur waktu eksekusi ===
fun benchmark() {
    val n = 100_000
    val items = generateItems(n)

    val tSync = measureNanoTime { pipelineSync(items) }
    val tSeq = measureNanoTime { pipelineSequence(items) }
    val tFlow = measureNanoTime { pipelineFlow(items) }

    println("\n=== Benchmark $n items ===")
    println("List (eager)  : ${"%.2f".format(tSync / 1_000_000.0)} ms")
    println("Sequence (lazy): ${"%.2f".format(tSeq / 1_000_000.0)} ms")
    println("Flow (suspend) : ${"%.2f".format(tFlow / 1_000_000.0)} ms")
    println("\nCatatan: Sequence unggul untuk koleksi besar dengan banyak operasi (single pass).")
    println("Flow menambah overhead suspend function, cocok jika ada operasi I/O di tengah.")
}

fun main() {
    println("=== Pengujian: verifikasi output ===")
    testPipeline()

    println("\n=== Pengukuran performa ===")
    benchmark()

    println("\nPengujian pipeline asinkron selesai.")
}

Output yang Dihasilkan:

text 
=== Pengujian: verifikasi output ===
[LULUS] pipelineSync(10 items) = [13.0, 16.0, 19.0, 22.0, 25.0, 28.0, 31.0]
[LULUS] pipelineSequence(10 items) = [13.0, 16.0, 19.0, 22.0, 25.0, 28.0, 31.0]
[LULUS] pipelineFlow(10 items) = [13.0, 16.0, 19.0, 22.0, 25.0, 28.0, 31.0]

=== Pengukuran performa ===

=== Benchmark 100000 items ===
List (eager)  : 17.35 ms
Sequence (lazy): 12.91 ms
Flow (suspend) : 16.66 ms

Catatan: Sequence unggul untuk koleksi besar dengan banyak operasi (single pass).
Flow menambah overhead suspend function, cocok jika ada operasi I/O di tengah.

Pengujian pipeline asinkron selesai.

Mari kita bedah bagian-bagian penting di atas:

  1. generateItems(n): factory function yang menghasilkan list Item dengan ukuran n. Pakai untuk input test yang konsisten.
  2. pipelineSync, pipelineSequence, pipelineFlow: tiga implementasi pipeline yang sama — map → filter → map. Perbedaan ada di container: List<T> eager, Sequence<T> lazy, Flow<T> suspend.
  3. testPipeline(): test harness manual menggunakan check(). Membangun expected result secara terpisah, lalu membandingkan dengan output tiga pipeline. Bila ada ketidaksesuaian, check melempar exception dengan pesan jelas.
  4. measureNanoTime { }: mengukur waktu eksekusi block dalam nanosecond. Lebih presisi daripada System.currentTimeMillis(). Konversi ke milidetik di laporan.
  5. runBlocking { } di pipelineFlow: diperlukan karena Flow.toList() adalah suspend function. Dalam production, gunakan runTest dari kotlinx-coroutines-test untuk virtual time.

Hasil Benchmark dan Interpretasi #

Pada 100.000 item, hasil umum:

  • List: paling lambat karena membuat 3 list intermediate (map, filter, map).
  • Sequence: paling cepat untuk single-pass chains. Tidak ada alokasi intermediate.
  • Flow: lebih lambat dari Sequence karena overhead suspend function (setiap emit cek continuation), tetapi unggul saat ada operasi I/O suspend di tengah.

Best Practice Pengujian Pipeline #

  1. Test behavior, bukan implementasi: verifikasi output, bukan jumlah panggilan internal.
  2. Test edge case: input kosong, single item, sangat besar, item bermasalah.
  3. Test error path: pastikan error handling bekerja — fake error dengan mock.
  4. Test concurrency safety: pipeline dengan banyak collector, pastikan tidak race condition.
  5. Property-based testing: kotest property untuk generate input acak dan verifikasi invariant.

Penutup Proyek #

Proyek pipeline data asinkron ini menggabungkan empat area penting: desain pipeline, coroutines dan concurrency, error handling, dan pengujian performa. Pola yang sama dapat diterapkan untuk ETL batch, stream processing, dan microservice async. Untuk skala besar, pertimbangkan framework seperti Apache Beam, Flink, atau Kafka Streams.

Referensi Lanjutan #

Selanjutnya #

Proyek ini menutup section Proyek. Section selanjutnya, Sumber Belajar, menyediakan referensi dan roadmap untuk melanjutkan pembelajaran Kotlin ke tingkat lanjutan.