feat(sdk-android): Uatu public API over SocketClient + Pauser

Uatu singleton owns an internal UatuRuntime that wires the socket
client, pauser, and extractor registry. start() sends HELLO on
connect, responds to PAUSE by running extractors inside the pause
window and replying with STATE, and releases the gate on RESUME.

LinkedHashMap preserves extractor registration order so trace
output is deterministic. Failed extractors log a warning and
record null rather than aborting the whole snapshot.
This commit is contained in:
pj committed 2026-04-17 22:57:12 +07:00
1 parent 8560d32cfe
commit 1700160077
3 files changed
+260 -10

No files matched your search

@@ -2,17 +2,39 @@ package dev.uatu.sdk
import android.app.Application import android.app.Application
object Uatu {
fun start(application: Application, configuration: Configuration = Configuration()) {
// stub — real implementation in tasks #6–8
}
fun extract(name: String, function: () -> Any?) {
// stub — registry lands in task #8
}
}
data class Configuration( data class Configuration(
val socketName: String = "uatu-agent", val socketName: String = "uatu-agent",
val pauseTimeoutMillis: Long = 5_000L, val pauseTimeoutMillis: Long = 5_000L,
) )
object Uatu {
const val VERSION: String = "0.0.1"
@Volatile private var runtime: UatuRuntime? = null
@Synchronized
fun start(application: Application, configuration: Configuration = Configuration()) {
if (runtime != null) return
val newRuntime = UatuRuntime(
transport = LocalAbstractTransport(configuration.socketName),
pauser = Pauser(ChoreographerPoster(), configuration.pauseTimeoutMillis),
version = VERSION,
platform = "android",
appPackage = application.packageName,
)
newRuntime.start()
runtime = newRuntime
}
fun extract(name: String, function: () -> Any?) {
val activeRuntime = runtime
?: throw IllegalStateException("Uatu.start must be called before registering extractors")
activeRuntime.register(name, function)
}
@Synchronized
internal fun stopForTest() {
runtime?.stop()
runtime = null
}
}
@@ -0,0 +1,82 @@
package dev.uatu.sdk
import android.util.Log
internal class UatuRuntime(
transport: AgentTransport,
private val pauser: Pauser,
private val version: String,
private val platform: String,
private val appPackage: String,
) {
private val extractors = LinkedHashMap<String, () -> Any?>()
@Volatile private var sender: SocketClient.MessageSender? = null
private val socketClient = SocketClient(transport, AgentHandler())
fun start() {
socketClient.start()
}
fun stop() {
socketClient.stop()
}
fun register(name: String, extractor: () -> Any?) {
synchronized(extractors) { extractors[name] = extractor }
}
internal fun snapshot(): Map<String, Any?> {
val drained = synchronized(extractors) { LinkedHashMap(extractors) }
val result = LinkedHashMap<String, Any?>(drained.size)
for ((name, extractor) in drained) {
result[name] = runCatching { extractor() }
.onFailure { cause -> Log.w(LOG_TAG, "extractor $name threw: $cause") }
.getOrNull()
}
return result
}
private inner class AgentHandler : SocketClient.Handler {
override fun onConnected(sender: SocketClient.MessageSender) {
this@UatuRuntime.sender = sender
try {
sender.send(Message.hello(version, platform, appPackage))
} catch (cause: Exception) {
Log.w(LOG_TAG, "failed to send HELLO: $cause")
}
}
override fun onMessage(message: Message) {
when (message.type) {
MessageType.PAUSE -> handlePause(message.id)
MessageType.RESUME -> pauser.release()
MessageType.GOODBYE -> socketClient.stop()
else -> Log.w(LOG_TAG, "unexpected message type ${message.type} from host")
}
}
override fun onDisconnected(cause: Throwable?) {
sender = null
pauser.release()
}
private fun handlePause(id: Long) {
val snapshots = try {
pauser.pauseAndSnapshot { snapshot() }
} catch (cause: Exception) {
Log.w(LOG_TAG, "snapshot failed: $cause")
emptyMap()
}
val activeSender = sender ?: return
try {
activeSender.send(Message.state(id, snapshots))
} catch (cause: Exception) {
Log.w(LOG_TAG, "failed to send STATE: $cause")
}
}
}
companion object {
private const val LOG_TAG = "Uatu"
}
}
@@ -0,0 +1,146 @@
package dev.uatu.sdk
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.CountDownLatch
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
import org.junit.After
import org.junit.Assert.assertArrayEquals
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertTrue
import org.junit.Test
class UatuRuntimeTest {
private lateinit var transport: SocketClientTest.FakeTransport
private lateinit var frameThread: PauserTest.FakeFrameThread
private lateinit var runtime: UatuRuntime
private fun newRuntime(): UatuRuntime {
transport = SocketClientTest.FakeTransport()
frameThread = PauserTest.FakeFrameThread()
val pauser = Pauser(frameThread, pauseTimeoutMillis = 2_000L)
return UatuRuntime(
transport = transport,
pauser = pauser,
version = "0.0.1",
platform = "android",
appPackage = "com.example.uatu_test",
).also { runtime = it }
}
@After fun tearDown() {
if (::runtime.isInitialized) runtime.stop()
if (::frameThread.isInitialized) frameThread.shutdown()
}
@Test fun startSendsHelloWithSdkMetadata() {
newRuntime().start()
val server = transport.nextServerEndpoint()
val hello = Protocol.read(server.input)
assertEquals(MessageType.HELLO, hello.type)
assertEquals("0.0.1", hello.version)
assertEquals("android", hello.platform)
assertEquals("com.example.uatu_test", hello.appPackage)
}
@Test fun pauseTriggersExtractorsAndReturnsState() {
newRuntime().start()
val server = transport.nextServerEndpoint()
Protocol.read(server.input) // drain HELLO
runtime.register("screen") { "customer_ledger" }
runtime.register("ledger.balance") { 1500 }
Protocol.write(server.output, Message.pause(7))
val state = Protocol.read(server.input)
assertEquals(MessageType.STATE, state.type)
assertEquals(7L, state.id)
val snapshots = state.snapshots ?: error("state.snapshots must not be null")
assertEquals("customer_ledger", snapshots["screen"])
assertEquals(1500, snapshots["ledger.balance"])
Protocol.write(server.output, Message.resume(7))
// After resume, the frame thread can accept subsequent callbacks.
Protocol.write(server.output, Message.pause(8))
val nextState = Protocol.read(server.input)
assertEquals(MessageType.STATE, nextState.type)
assertEquals(8L, nextState.id)
}
@Test fun extractorInvocationOrderMatchesRegistration() {
val runtime = newRuntime()
val order = CopyOnWriteArrayList<String>()
runtime.register("first") { order += "first"; 1 }
runtime.register("second") { order += "second"; 2 }
runtime.register("third") { order += "third"; 3 }
val snapshot = runtime.snapshot()
assertEquals(listOf("first", "second", "third"), order)
assertEquals(listOf("first", "second", "third"), snapshot.keys.toList())
assertArrayEquals(arrayOf(1, 2, 3), snapshot.values.toList().toTypedArray())
}
@Test fun extractorThrowIsIsolatedAndReportsNull() {
val runtime = newRuntime()
runtime.register("ok") { "value" }
runtime.register("boom") { throw IllegalStateException("oops") }
runtime.register("later") { 42 }
val snapshot = runtime.snapshot()
assertEquals("value", snapshot["ok"])
assertEquals(null, snapshot["boom"])
assertEquals(42, snapshot["later"])
}
@Test fun concurrentExtractorRegistrationIsSafe() {
val runtime = newRuntime()
val registrations = 500
val pool = Executors.newFixedThreadPool(8)
val latch = CountDownLatch(registrations)
val index = AtomicInteger(0)
repeat(registrations) {
pool.submit {
val id = index.getAndIncrement()
runtime.register("ext-$id") { id }
latch.countDown()
}
}
assertTrue(latch.await(5, TimeUnit.SECONDS))
pool.shutdown()
val snapshot = runtime.snapshot()
assertEquals(registrations, snapshot.size)
}
@Test fun helloIsSentOnReconnect() {
val runtime = newRuntime()
val shortBackoff = Backoff(initialDelayMillis = 10L, maxDelayMillis = 10L, multiplier = 1.0)
// Swap in a client with short backoff by re-creating the runtime's client indirectly.
// We'll re-use the existing runtime instead and just close the first connection.
runtime.start()
val firstServer = transport.nextServerEndpoint()
Protocol.read(firstServer.input) // drain first HELLO
firstServer.close()
val secondServer = transport.nextServerEndpoint(timeoutMillis = 3_000L)
val secondHello = Protocol.read(secondServer.input)
assertEquals(MessageType.HELLO, secondHello.type)
}
@Test fun registerBeforeStartQueuesCorrectlyOnceStarted() {
val runtime = newRuntime()
runtime.register("early") { 1 }
runtime.register("middle") { 2 }
runtime.start()
runtime.register("late") { 3 }
val snapshot = runtime.snapshot()
assertEquals(3, snapshot.size)
assertEquals(1, snapshot["early"])
assertEquals(2, snapshot["middle"])
assertEquals(3, snapshot["late"])
}
}