diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt index e5cbec6e2..fc70a5648 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt @@ -28,15 +28,18 @@ import com.vitorpamplona.nestsclient.transport.WebTransportSession import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.async import kotlinx.coroutines.cancel import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.onSubscription import kotlinx.coroutines.flow.take import kotlinx.coroutines.flow.toList import kotlinx.coroutines.launch @@ -175,19 +178,41 @@ class ReconnectingNestsListenerTest { val handle = reconnecting.subscribeSpeaker("speaker-pubkey") + // Signal when the consumer has actually subscribed to + // the outer frames flow. The wrapper's frames SharedFlow + // is replay=0, so an emit before the consumer registers + // its subscription is silently dropped — observed on + // macOS where the async coroutine isn't scheduled + // before the test thread races into the first emit. + val consumerSubscribed = CompletableDeferred() + + // The wrapper backs handle.objects with a SharedFlow + // (frames.asSharedFlow); the cast lets us use + // SharedFlow.onSubscription, which fires AFTER the + // collector is registered (Flow.onStart fires before). + @Suppress("UNCHECKED_CAST") + val objectsAsShared = handle.objects as SharedFlow val collected = scope.async { withTimeout(5_000L) { - handle.objects.take(2).toList() + objectsAsShared + .onSubscription { consumerSubscribed.complete(Unit) } + .take(2) + .toList() } } - // Wait for the pump to actually subscribe to the first - // session before we emit / fail. With Dispatchers.Default - // the pumpJob's collectLatest doesn't run synchronously - // off the launch site — yielding here lets it advance. + withTimeout(5_000L) { consumerSubscribed.await() } + + // Wait for the pump to actually subscribe to first.frames + // before we emit upstream. first.subscribeCount is + // incremented inside opener(first) — BEFORE the pump + // reaches `handle.objects.collect` — so subscribeCount + // alone doesn't prove the pump is listening. + // subscriptionCount on first.frames flips to 1 only once + // the pump's collect actually registers. withTimeout(5_000L) { - while (first.subscribeCount == 0) kotlinx.coroutines.delay(5) + first.frames.subscriptionCount.first { it > 0 } } first.frames.emit(frame(byteArrayOf(0x01))) @@ -198,11 +223,12 @@ class ReconnectingNestsListenerTest { // Both ScriptedListeners report the same `room` and // `negotiatedMoqVersion`, so we can't differentiate // first.Connected from second.Connected via state alone. - // The actual postcondition we care about — the pump - // re-issued the subscription against the new session — - // is observable directly via second.subscribeCount. + // The pump re-issuing the subscription against the new + // session is observable via second.frames.subscriptionCount + // (same rationale as first.frames above — wait for the + // collect, not just the subscribeSpeaker call). withTimeout(5_000L) { - while (second.subscribeCount == 0) kotlinx.coroutines.delay(5) + second.frames.subscriptionCount.first { it > 0 } } second.frames.emit(frame(byteArrayOf(0x02))) @@ -251,11 +277,18 @@ class ReconnectingNestsListenerTest { val handle = reconnecting.subscribeSpeaker("speaker-pubkey") - // Wait until the pump opened the underlying subscription — - // collectLatest schedules the inner block on a child of - // scope, so it runs after we yield. + // Wait until the pump opened the underlying subscription + // AND committed liveHandleRef. subscribeCount alone + // increments inside opener(first), which runs BEFORE + // the pump sets liveHandleRef + starts collecting from + // first.frames. If we unsubscribe while liveHandleRef + // is still null the inner unsubscribe never fires and + // the assertion below hangs/fails on macOS. + // first.frames.subscriptionCount flips to 1 only once + // the pump's collect actually registers, which is + // strictly after liveHandleRef.set(handle). withTimeout(5_000L) { - while (first.subscribeCount == 0) kotlinx.coroutines.yield() + first.frames.subscriptionCount.first { it > 0 } } assertTrue(first.subscribeCount >= 1, "pump should have opened the underlying sub")