DEV Community

Malek
Malek

Posted on

Turn a callback WebSocket feed into a Kotlin Flow with callbackFlow (with a live feed to try it on)

#kotlin #coroutines #flow #tutorial

Most APIs that push data hand you a callback. Kotlin code would rather have a Flow. callbackFlow is the bridge, and
it is short, but two details decide whether your version leaks sockets: where the cleanup goes, and whether the Flow is
cold.

To have something real to wrap, this article uses a public blockchain network that emits a block every 15 seconds over
a WebSocket. It needs no API key and no account, so the code runs as written. You do not need to know anything about
the chain; treat it as a server that pushes one event every 15 seconds.

The project

build.gradle.kts:

plugins {
    kotlin("jvm") version "2.2.0"
    application
}

repositories { mavenCentral() }

dependencies {
    implementation("org.janzeer:janzeer-sdk-client:0.1.7")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.10.2")
}

kotlin { jvmToolchain(17) }
application { mainClass.set("MainKt") }
Enter fullscreen mode Exit fullscreen mode

The callback API

The SDK's WebSocket client looks like most push clients:

val ws = JanzeerRpcWs.connect("https://explorer.janzeer.org")
val subscription = ws.subscribeNewBlocks(NewBlocksParams()) { block -> println(block.height) }
// … later
subscription.unsubscribe()
ws.close()
Enter fullscreen mode Exit fullscreen mode

Three things to manage by hand: the socket, the subscription, and the moment you stop caring.

The Flow 

import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import org.janzeer.sdk.rpc.BlockView
import org.janzeer.sdk.rpc.JanzeerRpcWs
import org.janzeer.sdk.rpc.NewBlocksParams

const val NODE = "https://explorer.janzeer.org"

/** Every newly committed block as a cold Flow: the socket opens on collect and closes when the collector stops. */
fun newBlocks(node: String): Flow<BlockView> = callbackFlow {
    val ws = JanzeerRpcWs.connect(node)
    val subscription = ws.subscribeNewBlocks(NewBlocksParams()) { block -> trySend(block) }
    awaitClose {
        subscription.unsubscribe()
        ws.close()
    }
}

fun main() = runBlocking {
    var previous: Long? = null
    withTimeout(90_000) {
        newBlocks(NODE)
            .map { block ->
                val gap = previous?.let { (block.timestamp - it) / 1000.0 }
                previous = block.timestamp
                "#${block.height}  ${block.hash.take(12)}…  gap=${gap ?: "-"}s"
            }
            .take(3)
            .collect(::println)
    }
}
Enter fullscreen mode Exit fullscreen mode

./gradlew run printed this on a real run:

#239312  126c6f6b7094…  gap=-s
#239313  c93fbe16e199…  gap=15.012s
#239314  02364077a56f…  gap=14.999s
Enter fullscreen mode Exit fullscreen mode

The two details

Cleanup belongs in awaitClose, and nowhere else. awaitClose suspends the builder until the collector goes
away, then runs its block. "Goes away" covers every case at once: take(3) finished, the collecting coroutine was
cancelled, withTimeout fired, or an exception came from downstream. Here it unsubscribes first, so the server is told,
and then closes the socket. If you forget awaitClose, callbackFlow throws at runtime to tell you so, which is one of
the kinder errors in the coroutines library.

Open the resource inside the builder, so the Flow is cold. JanzeerRpcWs.connect is called inside callbackFlow,
not outside it. Calling newBlocks(NODE) therefore opens nothing; the socket exists only while somebody collects, and
two collectors get two sockets. If you want one shared socket for many collectors, add shareIn at the call site, where
the scope that owns the socket is visible:

val shared = newBlocks(NODE).shareIn(scope, SharingStarted.WhileSubscribed(5_000), replay = 1)
Enter fullscreen mode Exit fullscreen mode

WhileSubscribed closes the socket when the last collector leaves, through the same awaitClose.

trySend, not send

The callback runs on the WebSocket's thread, which is not a coroutine, so it cannot call the suspending send.
trySend never blocks that thread. With the default buffer it drops an element only if the collector is 64 events
behind; at one event every 15 seconds that does not happen. For a fast feed, pick the policy on purpose:
.buffer(Channel.CONFLATED) to keep only the newest, or a larger buffer to keep all.

A timeout around the collector

withTimeout(90_000) is there because a push feed can go quiet without closing. Six block intervals without an event
means something is wrong, and a timeout turns "hangs for ever" into an exception you can log. The timeout cancels the
collector, and awaitClose still runs.

Where to go next

Top comments (0)