feat(sdk-android): Pauser with frame-callback-gated extractors

Worker thread calls pauseAndSnapshot with an extractor closure;
the Pauser posts a frame callback that runs extractors on the
main thread and then blocks that thread on a semaphore until
release() fires. FrameCallbackPoster abstracts Choreographer so
JVM unit tests drive it with a single-thread executor, while the
Android runtime uses ChoreographerPoster.
This commit is contained in:
pj committed 2026-04-17 22:55:06 +07:00
1 parent ac49818283
commit 52944b1aa6
3 files changed
+193

No files matched your search

@@ -0,0 +1,9 @@
package dev.uatu.sdk
import android.view.Choreographer
class ChoreographerPoster : FrameCallbackPoster {
override fun postFrameCallback(callback: () -> Unit) {
Choreographer.getInstance().postFrameCallback { callback() }
}
}
@@ -0,0 +1,51 @@
package dev.uatu.sdk
import java.util.concurrent.CountDownLatch
import java.util.concurrent.Semaphore
import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException
import java.util.concurrent.atomic.AtomicReference
fun interface FrameCallbackPoster {
fun postFrameCallback(callback: () -> Unit)
}
class Pauser(
private val poster: FrameCallbackPoster,
private val pauseTimeoutMillis: Long = 5_000L,
) {
@Volatile private var currentGate: Semaphore? = null
/**
* Schedules extractors to run on the frame-callback thread (the SDK's
* "main thread" analogue) and blocks that thread after they complete
* until release() is called or pauseTimeoutMillis elapses.
* Returns the extractor output. Must be called from a worker thread.
*/
@Throws(TimeoutException::class)
fun pauseAndSnapshot(extractors: () -> Map<String, Any?>): Map<String, Any?> {
val gate = Semaphore(0)
val ready = CountDownLatch(1)
val captured = AtomicReference<Result<Map<String, Any?>>>()
poster.postFrameCallback {
captured.set(runCatching { extractors() })
ready.countDown()
try {
gate.tryAcquire(pauseTimeoutMillis, TimeUnit.MILLISECONDS)
} catch (_: InterruptedException) {
Thread.currentThread().interrupt()
}
}
currentGate = gate
if (!ready.await(pauseTimeoutMillis, TimeUnit.MILLISECONDS)) {
throw TimeoutException("extractors did not run within ${pauseTimeoutMillis}ms")
}
return captured.get().getOrThrow()
}
fun release() {
currentGate?.release()
}
}
@@ -0,0 +1,133 @@
package dev.uatu.sdk
import java.util.concurrent.CountDownLatch
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicReference
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertTrue
import org.junit.Assert.fail
import org.junit.Test
class PauserTest {
// Runs posted callbacks on a dedicated "main thread" executor, modelling
// Choreographer's contract that callbacks fire off-thread from the caller.
class FakeFrameThread : FrameCallbackPoster {
val executor = Executors.newSingleThreadExecutor { runnable -> Thread(runnable, "fake-frame-thread") }
val threadRef = AtomicReference<Thread>()
override fun postFrameCallback(callback: () -> Unit) {
executor.submit {
threadRef.compareAndSet(null, Thread.currentThread())
callback()
}
}
fun shutdown() { executor.shutdownNow() }
}
private lateinit var frameThread: FakeFrameThread
@After fun tearDown() {
if (::frameThread.isInitialized) frameThread.shutdown()
}
@Test fun extractorsRunOnFrameThreadAndSnapshotReturns() {
frameThread = FakeFrameThread()
val pauser = Pauser(frameThread, pauseTimeoutMillis = 2_000L)
val extractorThread = AtomicReference<Thread>()
// Run on a worker thread so we can observe the separation.
val snapshot = runOnWorker {
pauser.pauseAndSnapshot {
extractorThread.set(Thread.currentThread())
mapOf("screen" to "home", "count" to 3)
}.also { snapshot ->
// Immediately release so the frame thread can exit the callback.
pauser.release()
snapshot
}
}
assertEquals("home", snapshot["screen"])
assertEquals(3, snapshot["count"])
assertNotNull("extractor must have run", extractorThread.get())
assertEquals("fake-frame-thread", extractorThread.get().name)
}
@Test fun frameThreadStaysBlockedUntilRelease() {
frameThread = FakeFrameThread()
val pauser = Pauser(frameThread, pauseTimeoutMillis = 2_000L)
val releasedMarker = AtomicBoolean(false)
val latch = CountDownLatch(1)
val worker = Thread {
pauser.pauseAndSnapshot { emptyMap() }
// Now the frame thread is blocked inside the callback. Verify by
// posting another callback and checking it does NOT run until we release.
val secondCallbackRan = CountDownLatch(1)
frameThread.postFrameCallback { secondCallbackRan.countDown() }
assertFalse("second callback should be queued, not run",
secondCallbackRan.await(200, TimeUnit.MILLISECONDS))
pauser.release()
assertTrue("second callback should run after release",
secondCallbackRan.await(2, TimeUnit.SECONDS))
releasedMarker.set(true)
latch.countDown()
}
worker.start()
assertTrue("worker must finish", latch.await(5, TimeUnit.SECONDS))
assertTrue("release must have happened", releasedMarker.get())
}
@Test fun timeoutPropagatesWhenFrameThreadNeverRuns() {
val stuck = FrameCallbackPoster { /* never invokes callback */ }
val pauser = Pauser(stuck, pauseTimeoutMillis = 150L)
try {
pauser.pauseAndSnapshot { emptyMap() }
fail("expected TimeoutException")
} catch (_: TimeoutException) {
// pass
}
}
@Test fun extractorExceptionBubbles() {
frameThread = FakeFrameThread()
val pauser = Pauser(frameThread, pauseTimeoutMillis = 2_000L)
try {
runOnWorker {
pauser.pauseAndSnapshot {
pauser.release() // drop the lock before throwing so main can unwind
throw IllegalStateException("extractor boom")
}
}
fail("expected IllegalStateException")
} catch (e: IllegalStateException) {
assertEquals("extractor boom", e.message)
}
}
@Test fun releaseWithoutActivePauseIsNoOp() {
frameThread = FakeFrameThread()
val pauser = Pauser(frameThread, pauseTimeoutMillis = 2_000L)
pauser.release() // should not throw
}
private fun <T> runOnWorker(block: () -> T): T {
val result = AtomicReference<Result<T>>()
val thread = Thread { result.set(runCatching { block() }) }
thread.start()
thread.join(5_000L)
return result.get().getOrThrow()
}
}