From ac6d8eb41725d67696d43944cfd80b44ed274291 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 16 Apr 2026 14:48:15 +0000 Subject: [PATCH] refactor: replace CallManager mutable callbacks with SharedFlow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Convert all 5 mutable `var` callback fields on CallManager (onAnswerReceived, onIceCandidateReceived, onNewPeerInGroupCall, onMidCallOfferReceived, onPeerLeft) into a single `sessionEvents: SharedFlow`. Benefits: - Eliminates the stale-callback window between Activity destroy and recreate. No more nulling callbacks in onDestroy — the `closed` flag on CallSession is sufficient. - No mutable state shared between Activity and background code. - CallActivity no longer wires or tears down callbacks — CallSession subscribes to the flow in its init block. - Type-safe sealed interface (CallSessionEvent) replaces 5 separate lambda types. The renegotiationEvents SharedFlow remains separate because renegotiation has its own glare-resolution logic in CallSession. Updated all 11 test sites in CallManagerTest to collect from sessionEvents instead of setting callback vars. https://claude.ai/code/session_019yNnDjGKmJb19gadmojq54 --- .../amethyst/ui/call/CallActivity.kt | 20 +--- .../amethyst/ui/call/session/CallSession.kt | 25 +++- .../amethyst/commons/call/CallManager.kt | 69 +++++------ .../amethyst/commons/call/CallSessionEvent.kt | 47 ++++++++ .../amethyst/commons/call/CallManagerTest.kt | 108 ++++++++++-------- 5 files changed, 171 insertions(+), 98 deletions(-) create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallSessionEvent.kt diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/CallActivity.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/CallActivity.kt index 16bdcc963..f6c8d2bb6 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/CallActivity.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/CallActivity.kt @@ -126,12 +126,9 @@ class CallActivity : AppCompatActivity() { ) session = callSession - // Wire CallManager callbacks to this session's handlers. - callManager.onAnswerReceived = { event -> callSession.onCallAnswerReceived(event.pubKey, event.sdpAnswer()) } - callManager.onIceCandidateReceived = { event -> callSession.onIceCandidateReceived(event) } - callManager.onNewPeerInGroupCall = { peerPubKey -> callSession.onNewPeerInGroupCall(peerPubKey) } - callManager.onMidCallOfferReceived = { peerPubKey, sdpOffer -> callSession.onMidCallOfferReceived(peerPubKey, sdpOffer) } - callManager.onPeerLeft = { peerPubKey -> callSession.disposePeerSession(peerPubKey) } + // No callback wiring needed — CallSession collects from + // callManager.sessionEvents and renegotiationEvents SharedFlows + // in its init block. registerPipReceiver() observeVideoStateForPip(callSession) @@ -240,15 +237,8 @@ class CallActivity : AppCompatActivity() { session?.close() session = null - // Null out callbacks so signaling events arriving after this - // Activity is gone don't route to a dead CallSession. - if (manager != null) { - manager.onAnswerReceived = null - manager.onIceCandidateReceived = null - manager.onNewPeerInGroupCall = null - manager.onMidCallOfferReceived = null - manager.onPeerLeft = null - } + // No callback nulling needed — CallSession collects from + // SharedFlows; the `closed` flag prevents processing after close(). // Publish reject/hangup so the remote side stops ringing. // Use goAsync-style: the hangup is best-effort. If the process diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/session/CallSession.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/session/CallSession.kt index bdd3f81df..bde72db6f 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/session/CallSession.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/session/CallSession.kt @@ -141,9 +141,28 @@ class CallSession( // ---- Initialization ---- init { - // Subscribe to renegotiation events via SharedFlow instead of a - // mutable callback, so we never miss an event across Activity - // restarts. + // Collect all session-level events (answers, ICE candidates, + // peer joins/leaves, mid-call offers) from the single SharedFlow. + scope.launch { + callManager.sessionEvents.collect { event -> + if (closed) return@collect + when (event) { + is com.vitorpamplona.amethyst.commons.call.CallSessionEvent.AnswerReceived -> + onCallAnswerReceived(event.event.pubKey, event.event.sdpAnswer()) + is com.vitorpamplona.amethyst.commons.call.CallSessionEvent.IceCandidateReceived -> + onIceCandidateReceived(event.event) + is com.vitorpamplona.amethyst.commons.call.CallSessionEvent.NewPeerInGroupCall -> + onNewPeerInGroupCall(event.peerPubKey) + is com.vitorpamplona.amethyst.commons.call.CallSessionEvent.MidCallOfferReceived -> + onMidCallOfferReceived(event.peerPubKey, event.sdpOffer) + is com.vitorpamplona.amethyst.commons.call.CallSessionEvent.PeerLeft -> + disposePeerSession(event.peerPubKey) + } + } + } + + // Renegotiation events have their own flow because they go + // through dedicated glare-resolution logic. scope.launch { callManager.renegotiationEvents.collect { event -> if (!closed) onRenegotiationOfferReceived(event) diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt index 99a16b5f3..d2bd72186 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt @@ -76,29 +76,30 @@ class CallManager( private val _state = MutableStateFlow(CallState.Idle) val state: StateFlow = _state.asStateFlow() - /** Called for every answer that should be applied to a PeerConnection (includes peer pubkey). */ - var onAnswerReceived: ((CallAnswerEvent) -> Unit)? = null - var onIceCandidateReceived: ((CallIceCandidateEvent) -> Unit)? = null + /** + * All events that the Activity-scoped `CallSession` must handle are + * emitted through this single flow. Using a SharedFlow instead of + * mutable `var` callbacks means: + * - No stale references to a dead CallSession between Activity + * destroy and recreate. + * - The collector can (re)subscribe via `repeatOnLifecycle` without + * missing buffered events. + * - No null-check on every emit site. + * + * Buffer size 16 covers bursts of ICE candidates + answers arriving + * in quick succession during call setup. + */ + private val _sessionEvents = MutableSharedFlow(extraBufferCapacity = 16) + val sessionEvents: SharedFlow = _sessionEvents.asSharedFlow() /** - * Renegotiation offers from remote peers are emitted as flow events so the - * Activity-scoped `CallSession` can (re)subscribe via `repeatOnLifecycle` - * without losing messages across Activity restarts. Buffered so a short - * gap between Activity teardown and re-collection cannot drop a - * renegotiation. + * Renegotiation offers from remote peers. Separate from [sessionEvents] + * because renegotiation has its own glare-resolution logic in + * CallSession and benefits from a dedicated collector. */ private val _renegotiationEvents = MutableSharedFlow(extraBufferCapacity = 8) val renegotiationEvents: SharedFlow = _renegotiationEvents.asSharedFlow() - /** Called when a new peer joins the group call (callee-to-callee mesh setup). */ - var onNewPeerInGroupCall: ((peerPubKey: HexKey) -> Unit)? = null - - /** Called when a mid-call offer is received from another callee in a group call. */ - var onMidCallOfferReceived: ((peerPubKey: HexKey, sdpOffer: String) -> Unit)? = null - - /** Called when a peer leaves the call (hangup) but the call continues with remaining peers. */ - var onPeerLeft: ((peerPubKey: HexKey) -> Unit)? = null - private var timeoutJob: Job? = null private var resetJob: Job? = null private val processedEventIds = LinkedHashSet() @@ -349,7 +350,7 @@ class CallManager( else -> {} } - onMidCallOfferReceived?.invoke(callerPubKey, event.sdpOffer()) + _sessionEvents.tryEmit(CallSessionEvent.MidCallOfferReceived(callerPubKey, event.sdpOffer())) return } } @@ -450,7 +451,7 @@ class CallManager( if (discovered.isNotEmpty()) { Log.d("CallManager") { "acceptCall: triggering mesh setup with ${discovered.size} discovered peers: ${discovered.map { it.take(8) }}" } for (peer in discovered) { - onNewPeerInGroupCall?.invoke(peer) + _sessionEvents.tryEmit(CallSessionEvent.NewPeerInGroupCall(peer)) } } } @@ -517,7 +518,7 @@ class CallManager( disarmRingingWatchdog() armConnectingWatchdog(current.callId) Log.d("CallManager") { "onCallAnswered: Offering -> Connecting, forwarding answer to CallSession" } - onAnswerReceived?.invoke(event) + _sessionEvents.tryEmit(CallSessionEvent.AnswerReceived(event)) } is CallState.Connecting -> { @@ -546,7 +547,7 @@ class CallManager( // Forward to CallController — it routes to the correct PeerSession // and internally triggers callee-to-callee mesh setup if needed. Log.d("CallManager") { "onCallAnswered: additional peer ${answeringPeer.take(8)} in Connecting, forwarding to CallController" } - onAnswerReceived?.invoke(event) + _sessionEvents.tryEmit(CallSessionEvent.AnswerReceived(event)) } is CallState.IncomingCall -> { @@ -585,7 +586,7 @@ class CallManager( // Forward to CallController — it routes to the correct PeerSession // and internally triggers callee-to-callee mesh setup if needed. Log.d("CallManager") { "onCallAnswered: peer ${answeringPeer.take(8)} answer in Connected, forwarding to CallController" } - onAnswerReceived?.invoke(event) + _sessionEvents.tryEmit(CallSessionEvent.AnswerReceived(event)) } else -> { @@ -603,7 +604,7 @@ class CallManager( // still ringing (IncomingCall). In every other state it is our own // echo from a relay after a sibling device rejected a call that we // have already accepted and are now actively in — treating it as a - // peer rejection would drop ourselves from the call (onPeerLeft + // peer rejection would drop ourselves from the call (PeerLeft // would dispose our own PeerConnection, killing the live audio on // the device that picked up). Just drop the echo. if (rejectingPeer == signer.pubKey) { @@ -625,7 +626,7 @@ class CallManager( transitionToEnded(current.callId, current.peerPubKeys, EndReason.PEER_REJECTED) } else { _state.value = current.copy(peerPubKeys = remaining) - onPeerLeft?.invoke(rejectingPeer) + _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(rejectingPeer)) } } @@ -634,7 +635,7 @@ class CallManager( cancelPeerTimeout(rejectingPeer) _state.value = current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - rejectingPeer) - onPeerLeft?.invoke(rejectingPeer) + _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(rejectingPeer)) } is CallState.Connected -> { @@ -642,7 +643,7 @@ class CallManager( cancelPeerTimeout(rejectingPeer) _state.value = current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - rejectingPeer) - onPeerLeft?.invoke(rejectingPeer) + _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(rejectingPeer)) } is CallState.IncomingCall -> { @@ -669,7 +670,7 @@ class CallManager( } private fun onIceCandidate(event: CallIceCandidateEvent) { - onIceCandidateReceived?.invoke(event) + _sessionEvents.tryEmit(CallSessionEvent.IceCandidateReceived(event)) } private fun onRenegotiate(event: CallRenegotiateEvent) { @@ -831,7 +832,7 @@ class CallManager( peerPubKeys = connectedRemaining, pendingPeerPubKeys = pendingRemaining, ) - onPeerLeft?.invoke(leavingPeer) + _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(leavingPeer)) } } @@ -856,7 +857,7 @@ class CallManager( peerPubKeys = connectedRemaining, pendingPeerPubKeys = pendingRemaining, ) - onPeerLeft?.invoke(leavingPeer) + _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(leavingPeer)) } } @@ -868,7 +869,7 @@ class CallManager( transitionToEnded(callId, current.peerPubKeys, EndReason.PEER_HANGUP) } else { _state.value = current.copy(peerPubKeys = remaining) - onPeerLeft?.invoke(leavingPeer) + _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(leavingPeer)) } } @@ -1160,7 +1161,7 @@ class CallManager( * at least one other peer is connected by definition, so the call * always continues. * - * Fires [onPeerLeft] so the CallController disposes the per-peer + * Emits [CallSessionEvent.PeerLeft] so the CallSession disposes the per-peer * PeerConnection (and any pending ICE buffers) for the dropped peer. */ private suspend fun handlePeerTimeout( @@ -1186,7 +1187,7 @@ class CallManager( transitionToEnded(current.callId, current.peerPubKeys, EndReason.TIMEOUT) } else { _state.value = current.copy(peerPubKeys = remaining) - onPeerLeft?.invoke(peerPubKey) + _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(peerPubKey)) } } @@ -1204,7 +1205,7 @@ class CallManager( ) } else { _state.value = current.copy(pendingPeerPubKeys = newPending) - onPeerLeft?.invoke(peerPubKey) + _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(peerPubKey)) } } @@ -1214,7 +1215,7 @@ class CallManager( Log.d("CallManager") { "Per-peer timeout: dropping ${peerPubKey.take(8)} from Connected (invitedByUs=$wasInvitedByUs)" } shouldPublishHangup = wasInvitedByUs _state.value = current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - peerPubKey) - onPeerLeft?.invoke(peerPubKey) + _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(peerPubKey)) } else -> { diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallSessionEvent.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallSessionEvent.kt new file mode 100644 index 000000000..69488a250 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallSessionEvent.kt @@ -0,0 +1,47 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.amethyst.commons.call + +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nipACWebRtcCalls.events.CallAnswerEvent +import com.vitorpamplona.quartz.nipACWebRtcCalls.events.CallIceCandidateEvent + +/** + * Events emitted by [CallManager] that the Activity-scoped `CallSession` + * must handle. Delivered via `SharedFlow` so the collector can + * (re)subscribe across Activity restarts without missing buffered events. + */ +sealed interface CallSessionEvent { + /** A peer answered our offer (or we received an answer in a group call). */ + data class AnswerReceived(val event: CallAnswerEvent) : CallSessionEvent + + /** An ICE candidate arrived from a peer. */ + data class IceCandidateReceived(val event: CallIceCandidateEvent) : CallSessionEvent + + /** A new peer joined the group call and needs callee-to-callee mesh setup. */ + data class NewPeerInGroupCall(val peerPubKey: HexKey) : CallSessionEvent + + /** A mid-call offer arrived from another callee in a group call. */ + data class MidCallOfferReceived(val peerPubKey: HexKey, val sdpOffer: String) : CallSessionEvent + + /** A peer left the call (hangup/reject/timeout) but the call continues. */ + data class PeerLeft(val peerPubKey: HexKey) : CallSessionEvent +} diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/CallManagerTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/CallManagerTest.kt index b86369b87..222cd68b8 100644 --- a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/CallManagerTest.kt +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/CallManagerTest.kt @@ -33,6 +33,8 @@ import com.vitorpamplona.quartz.nipACWebRtcCalls.events.CallRenegotiateEvent import com.vitorpamplona.quartz.nipACWebRtcCalls.tags.CallType import com.vitorpamplona.quartz.utils.TimeUtils import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.Job +import kotlinx.coroutines.launch import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.test.advanceTimeBy import kotlinx.coroutines.test.advanceUntilIdle @@ -100,6 +102,16 @@ class CallManagerTest { return manager to published } + /** + * Starts collecting [CallManager.sessionEvents] into a list. + * Returns (events, job) — cancel the job when done. + */ + private fun TestScope.collectSessionEvents(manager: CallManager): Pair, Job> { + val events = mutableListOf() + val job = launch { manager.sessionEvents.collect { events.add(it) } } + return events to job + } + // ---- Event construction helpers ---- // These use the real event builders from quartz so that tag structures // stay in sync with the production code. The builder returns an @@ -323,15 +335,15 @@ class CallManagerTest { manager.initiateCall(bob, CallType.VOICE, callId, sdpOffer) assertIs(manager.state.value) - var answerReceived = false - manager.onAnswerReceived = { answerReceived = true } + val (events, job) = collectSessionEvents(manager) val answer = makeAnswer(from = bob, to = alice) manager.onSignalingEvent(answer) val state = manager.state.value assertIs(state) - assertTrue(answerReceived, "onAnswerReceived callback should fire") + assertTrue(events.any { it is CallSessionEvent.AnswerReceived }, "AnswerReceived event should be emitted") + job.cancel() } // ======================================================================== @@ -461,14 +473,14 @@ class CallManagerTest { manager.onSignalingEvent(makeOffer(from = alice, to = bob)) manager.acceptCall(sdpAnswer) - var iceCalled = false - manager.onIceCandidateReceived = { iceCalled = true } + val (events, job) = collectSessionEvents(manager) // ICE candidate from self (echoed back by relay) val selfIce = makeIceCandidate(from = bob, to = alice) manager.onSignalingEvent(selfIce) - assertTrue(!iceCalled, "Self ICE candidates MUST be ignored") + assertTrue(events.none { it is CallSessionEvent.IceCandidateReceived }, "Self ICE candidates MUST be ignored") + job.cancel() } @Test @@ -533,13 +545,14 @@ class CallManagerTest { manager.onSignalingEvent(makeOffer(from = alice, to = bob)) manager.acceptCall(sdpAnswer) - var receivedIce: CallIceCandidateEvent? = null - manager.onIceCandidateReceived = { receivedIce = it } + val (events, job) = collectSessionEvents(manager) val ice = makeIceCandidate(from = alice, to = bob) manager.onSignalingEvent(ice) - assertNotNull(receivedIce, "ICE candidate should be forwarded via callback") + val iceEvent = events.filterIsInstance().firstOrNull() + assertNotNull(iceEvent, "ICE candidate should be emitted via sessionEvents") + job.cancel() } // ======================================================================== @@ -794,15 +807,16 @@ class CallManagerTest { // Bob still ringing assertIs(manager.state.value) - // Track mesh setup callback - val newPeers = mutableListOf() - manager.onNewPeerInGroupCall = { newPeers.add(it) } + // Track mesh setup events + val (events, job) = collectSessionEvents(manager) // Bob accepts manager.acceptCall(sdpAnswer) // Should trigger callee-to-callee mesh setup with carol + val newPeers = events.filterIsInstance().map { it.peerPubKey } assertTrue(carol in newPeers, "Should discover carol for mesh setup after accepting") + job.cancel() } // ======================================================================== @@ -819,19 +833,17 @@ class CallManagerTest { manager.onPeerConnected() assertIs(manager.state.value) - var midCallPeer: HexKey? = null - var midCallSdp: String? = null - manager.onMidCallOfferReceived = { peer, sdp -> - midCallPeer = peer - midCallSdp = sdp - } + val (events, job) = collectSessionEvents(manager) // Carol sends a mid-call offer (callee-to-callee mesh) val carolOffer = makeOffer(from = carol, to = bob, callId = callId, sdp = "carol-sdp") manager.onSignalingEvent(carolOffer) - assertEquals(carol, midCallPeer) - assertEquals("carol-sdp", midCallSdp) + val midCallEvent = events.filterIsInstance().firstOrNull() + assertNotNull(midCallEvent) + assertEquals(carol, midCallEvent.peerPubKey) + assertEquals("carol-sdp", midCallEvent.sdpOffer) + job.cancel() } // ======================================================================== @@ -882,12 +894,13 @@ class CallManagerTest { manager.onPeerConnected() manager.onSignalingEvent(makeAnswer(from = carol, to = alice)) - val leftPeers = mutableListOf() - manager.onPeerLeft = { leftPeers.add(it) } + val (events, job) = collectSessionEvents(manager) manager.onSignalingEvent(makeHangup(from = bob, to = alice)) - assertTrue(bob in leftPeers, "onPeerLeft should fire when a peer hangs up") + val leftPeers = events.filterIsInstance().map { it.peerPubKey } + assertTrue(bob in leftPeers, "PeerLeft should be emitted when a peer hangs up") + job.cancel() } // ======================================================================== @@ -1128,8 +1141,8 @@ class CallManagerTest { * Bob is already in a Connected group call with Alice. Alice invites Carol. * Carol's broadcast CallAnswer reaches Bob. Bob's state must expand to * include Carol in [CallState.Connected.peerPubKeys] and the answer must - * still be forwarded via [CallManager.onAnswerReceived] so the caller-side - * [CallController] can unconditionally initiate a mesh offer to Carol. + * still be emitted via [CallManager.sessionEvents] so the caller-side + * CallSession can unconditionally initiate a mesh offer to Carol. */ @Test fun midCallInviteAnswerFromUnknownPeerInConnectedExpandsMembership() = @@ -1142,8 +1155,7 @@ class CallManagerTest { manager.onPeerConnected() assertIs(manager.state.value) - var forwardedPeer: HexKey? = null - manager.onAnswerReceived = { event -> forwardedPeer = event.pubKey } + val (events, job) = collectSessionEvents(manager) // Alice invited Carol mid-call; Carol broadcasts her acceptance. // Bob sees a CallAnswer from Carol (unknown peer, same call-id) @@ -1155,7 +1167,9 @@ class CallManagerTest { assertIs(state) assertTrue(carol in state.peerPubKeys, "Mid-call joiner must be added to peerPubKeys") assertTrue(alice in state.peerPubKeys, "Existing peer must still be present") - assertEquals(carol, forwardedPeer, "Answer must still be forwarded to CallController") + val forwardedPeer = events.filterIsInstance().firstOrNull()?.event?.pubKey + assertEquals(carol, forwardedPeer, "Answer must still be emitted to CallSession") + job.cancel() } /** @@ -1296,16 +1310,17 @@ class CallManagerTest { // Step 6: Bob receives Carol's answer broadcast. Bob's state must // expand to include Carol (mid-call join), and the answer must be - // forwarded so Bob's CallController can initiate a mesh offer. - var bobForwardedAnswer: HexKey? = null - bobManager.onAnswerReceived = { event -> bobForwardedAnswer = event.pubKey } + // emitted so Bob's CallSession can initiate a mesh offer. + val (bobEvents, bobJob) = collectSessionEvents(bobManager) bobManager.onSignalingEvent( makeGroupAnswer(from = carol, members = setOf(alice, bob, carol), sdp = "carol-answer-sdp"), ) val bobAfterCarolAnswer = bobManager.state.value assertIs(bobAfterCarolAnswer) assertTrue(carol in bobAfterCarolAnswer.peerPubKeys, "Bob should add Carol to his membership") - assertEquals(carol, bobForwardedAnswer, "Bob must forward Carol's answer to his CallController") + val bobForwardedAnswer = bobEvents.filterIsInstance().firstOrNull()?.event?.pubKey + assertEquals(carol, bobForwardedAnswer, "Bob must emit Carol's answer to his CallSession") + bobJob.cancel() } // ======================================================================== @@ -1528,15 +1543,16 @@ class CallManagerTest { assertTrue(carol in connecting.pendingPeerPubKeys) // Carol publishes a mesh offer to Bob (same call-id). - var midCallOfferPeer: HexKey? = null - manager.onMidCallOfferReceived = { peer, _ -> midCallOfferPeer = peer } + val (events, job) = collectSessionEvents(manager) manager.onSignalingEvent(makeGroupOffer(from = carol, members = setOf(alice, bob, carol))) val after = manager.state.value assertIs(after) assertTrue(carol in after.peerPubKeys, "Carol should move into peerPubKeys") assertTrue(carol !in after.pendingPeerPubKeys, "Carol should leave pending") - assertEquals(carol, midCallOfferPeer, "CallController must still receive the mid-call offer") + val midCallOfferPeer = events.filterIsInstance().firstOrNull()?.peerPubKey + assertEquals(carol, midCallOfferPeer, "CallSession must still receive the mid-call offer") + job.cancel() // And her watchdog should be cancelled — advancing past the // 30 s budget must not further alter state. @@ -1929,7 +1945,7 @@ class CallManagerTest { /** * Regression for the subtle bug where a self-reject echo would trigger - * onPeerLeft(signer.pubKey) on a device that had already accepted, + * PeerLeft(signer.pubKey) on a device that had already accepted, * disposing its own PeerSession and killing the live call. */ @Test @@ -1937,8 +1953,7 @@ class CallManagerTest { runTest { val (manager, _) = createManager(localPubKey = bob, followedKeys = setOf(alice)) - var peerLeftCalled = false - manager.onPeerLeft = { peerLeftCalled = true } + val (events, job) = collectSessionEvents(manager) manager.onSignalingEvent(makeOffer(from = alice, to = bob)) manager.acceptCall(sdpAnswer) @@ -1950,29 +1965,29 @@ class CallManagerTest { val siblingReject = makeReject(from = bob, to = alice) manager.onSignalingEvent(siblingReject) - // Our live call must survive — the echo must not fire - // onPeerLeft (which would dispose our own PeerSession) and must + // Our live call must survive — the echo must not emit + // PeerLeft (which would dispose our own PeerSession) and must // not change state. assertIs(manager.state.value) assertTrue( - !peerLeftCalled, - "onPeerLeft must not fire for a self-reject echo — the UI would " + + events.none { it is CallSessionEvent.PeerLeft }, + "PeerLeft must not be emitted for a self-reject echo — the UI would " + "tear down our own PeerSession, killing the live call audio.", ) + job.cancel() } /** * Same guard for the Connecting state: a self-reject arriving during * the connection handshake must not drop us from our own pending set - * and must not invoke onPeerLeft(self). + * and must not emit PeerLeft(self). */ @Test fun selfRejectEchoDuringConnectingDoesNotFireOnPeerLeft() = runTest { val (manager, _) = createManager(localPubKey = bob, followedKeys = setOf(alice)) - var peerLeftCalled = false - manager.onPeerLeft = { peerLeftCalled = true } + val (events, job) = collectSessionEvents(manager) manager.onSignalingEvent(makeOffer(from = alice, to = bob)) manager.acceptCall(sdpAnswer) @@ -1982,7 +1997,8 @@ class CallManagerTest { manager.onSignalingEvent(siblingReject) assertIs(manager.state.value) - assertTrue(!peerLeftCalled, "onPeerLeft must not fire for a self-reject echo during Connecting") + assertTrue(events.none { it is CallSessionEvent.PeerLeft }, "PeerLeft must not be emitted for a self-reject echo during Connecting") + job.cancel() } /**