@@ -474,6 +474,111 @@ class GatewaySessionInvokeTest {
474474 }
475475 }
476476477+ @Test
478+fun sendNodeEventDetailed_sendsPresenceAlivePayloadAndReturnsStructuredResponse() = runBlocking {
479+val json = testJson()
480+val connected = CompletableDeferred<Unit>()
481+val nodeEventParams = CompletableDeferred<JsonObject>()
482+val lastDisconnect = AtomicReference("")
483+val server =
484+ startGatewayServer(json) { webSocket, id, method, frame ->
485+when (method) {
486+"connect" -> {
487+ webSocket.send(connectResponseFrame(id))
488+ }
489+"node.event" -> {
490+if (!nodeEventParams.isCompleted) {
491+ nodeEventParams.complete(frame["params"]?.jsonObject ?: JsonObject(emptyMap()))
492+ }
493+val payload =
494+"""{"ok":true,"event":"node.presence.alive","handled":true,"reason":"persisted"}"""
495+ webSocket.send(
496+"""{"type":"res","id":"$id","ok":true,"payload":$payload}""",
497+ )
498+ webSocket.close(1000, "done")
499+ }
500+ }
501+ }
502+503+val harness =
504+ createNodeHarness(
505+ connected = connected,
506+ lastDisconnect = lastDisconnect,
507+ ) { GatewaySession.InvokeResult.ok("""{"handled":true}""") }
508+509+try {
510+ connectNodeSession(harness.session, server.port)
511+ awaitConnectedOrThrow(connected, lastDisconnect, server)
512+513+val result =
514+ harness.session.sendNodeEventDetailed(
515+ event = "node.presence.alive",
516+ payloadJson = """{"trigger":"connect","sentAtMs":123}""",
517+ timeoutMs = TEST_TIMEOUT_MS,
518+ )
519+val params = withTimeout(TEST_TIMEOUT_MS) { nodeEventParams.await() }
520+val response = json.parseToJsonElement(result.payloadJson.orEmpty()).jsonObject
521+val payload = json.parseToJsonElement(params["payloadJSON"]?.jsonPrimitive?.content.orEmpty()).jsonObject
522+523+ assertEquals(true, result.ok)
524+ assertEquals("node.presence.alive", params["event"]?.jsonPrimitive?.content)
525+ assertEquals("connect", payload["trigger"]?.jsonPrimitive?.content)
526+ assertEquals("123", payload["sentAtMs"]?.jsonPrimitive?.content)
527+ assertEquals(true, response["handled"]?.jsonPrimitive?.content?.toBooleanStrict())
528+ assertEquals("persisted", response["reason"]?.jsonPrimitive?.content)
529+ } finally {
530+ shutdownHarness(harness, server)
531+ }
532+ }
533+534+ @Test
535+fun sendNodeEvent_preservesCompletedRpcAsSuccessWhenGatewayReturnsError() = runBlocking {
536+val json = testJson()
537+val connected = CompletableDeferred<Unit>()
538+val nodeEventParams = CompletableDeferred<JsonObject>()
539+val lastDisconnect = AtomicReference("")
540+val server =
541+ startGatewayServer(json) { webSocket, id, method, frame ->
542+when (method) {
543+"connect" -> {
544+ webSocket.send(connectResponseFrame(id))
545+ }
546+"node.event" -> {
547+if (!nodeEventParams.isCompleted) {
548+ nodeEventParams.complete(frame["params"]?.jsonObject ?: JsonObject(emptyMap()))
549+ }
550+ webSocket.send(
551+"""{"type":"res","id":"$id","ok":false,"error":{"code":"RATE_LIMITED","message":"slow down"}}""",
552+ )
553+ webSocket.close(1000, "done")
554+ }
555+ }
556+ }
557+558+val harness =
559+ createNodeHarness(
560+ connected = connected,
561+ lastDisconnect = lastDisconnect,
562+ ) { GatewaySession.InvokeResult.ok("""{"handled":true}""") }
563+564+try {
565+ connectNodeSession(harness.session, server.port)
566+ awaitConnectedOrThrow(connected, lastDisconnect, server)
567+568+val sent =
569+ harness.session.sendNodeEvent(
570+ event = "agent.request",
571+ payloadJson = """{"message":"restore"}""",
572+ )
573+val params = withTimeout(TEST_TIMEOUT_MS) { nodeEventParams.await() }
574+575+ assertEquals(true, sent)
576+ assertEquals("agent.request", params["event"]?.jsonPrimitive?.content)
577+ } finally {
578+ shutdownHarness(harness, server)
579+ }
580+ }
581+477582private fun testJson(): Json = Json { ignoreUnknownKeys = true }
478583479584private fun createNodeHarness(
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。