diff --git a/android/app/src/main/cpp/openstream_srt.cpp b/android/app/src/main/cpp/openstream_srt.cpp index da0db019..49adf29f 100644 --- a/android/app/src/main/cpp/openstream_srt.cpp +++ b/android/app/src/main/cpp/openstream_srt.cpp @@ -484,6 +484,11 @@ std::optional parseSrtUrl(const std::string &url) { class NativeSender { public: + static constexpr int kAcceptFailed = -1; + static constexpr int kAcceptPending = 0; + static constexpr int kAcceptConnected = 1; + static constexpr int kAcceptCancelled = 2; + NativeSender() { #if OPENSTREAM_HAVE_LIBSRT srtStarted_ = srt_startup() == 0; @@ -589,7 +594,7 @@ class NativeSender { #endif } - bool listen(const std::string &url, uint64_t expectedLifecycleGeneration) { + bool startListen(const std::string &url, uint64_t expectedLifecycleGeneration) { #if OPENSTREAM_HAVE_LIBSRT disconnect(); const auto parsed = parseSrtUrl(url); @@ -646,29 +651,82 @@ class NativeSender { return false; } + const int epoll = srt_epoll_create(); + if (epoll < 0) { + logError("Could not create SRT listener readiness poller"); + disconnect(); + return false; + } + const int events = SRT_EPOLL_IN | SRT_EPOLL_ERR; + if (srt_epoll_add_usock(epoll, listenerSocket, &events) == SRT_ERROR) { + __android_log_print(ANDROID_LOG_ERROR, kTag, "Could not monitor SRT listener: %s", srt_getlasterror_str()); + srt_epoll_release(epoll); + disconnect(); + return false; + } + { + std::lock_guard lock(socketMutex_); + if (lifecycleGeneration_ != expectedLifecycleGeneration || listener_socket_ != listenerSocket) { + srt_epoll_release(epoll); + return false; + } + listener_epoll_ = epoll; + } + __android_log_print(ANDROID_LOG_INFO, kTag, "Listening for OBS caller on SRT port %s", parsed->port.c_str()); + return true; +#else + (void)url; + (void)expectedLifecycleGeneration; + logError("openstream_srt was built without libsrt. Rebuild with OPENSTREAM_ENABLE_LIBSRT=ON."); + return false; +#endif + } + + // Performs one zero-timeout readiness check and accepts only after SRT reports + // SRT_EPOLL_IN. This keeps the session worker available for disconnect/reconnect. + int acceptPending(uint64_t expectedLifecycleGeneration) { +#if OPENSTREAM_HAVE_LIBSRT + std::lock_guard ioLock(ioMutex_); + SRTSOCKET listenerSocket = SRT_INVALID_SOCK; + int epoll = -1; + { + std::lock_guard lock(socketMutex_); + if (lifecycleGeneration_ != expectedLifecycleGeneration) return kAcceptCancelled; + listenerSocket = listener_socket_; + epoll = listener_epoll_; + } + if (listenerSocket == SRT_INVALID_SOCK || epoll < 0) return kAcceptCancelled; + + SRT_EPOLL_EVENT event{}; + const int ready = srt_epoll_uwait(epoll, &event, 1, 0); + if (ready == SRT_ERROR) { + __android_log_print(ANDROID_LOG_ERROR, kTag, "SRT listener readiness poll failed: %s", srt_getlasterror_str()); + return kAcceptFailed; + } + if (ready == 0) return kAcceptPending; + if (event.fd != listenerSocket || (event.events & SRT_EPOLL_IN) == 0) { + logError("SRT readiness poll returned an unexpected socket"); + return kAcceptFailed; + } + sockaddr_storage peer{}; int peer_len = sizeof(peer); - __android_log_print(ANDROID_LOG_INFO, kTag, "Waiting for OBS caller on SRT port %s", parsed->port.c_str()); const SRTSOCKET acceptedSocket = srt_accept(listenerSocket, reinterpret_cast(&peer), &peer_len); closeListenerSocket(listenerSocket); if (acceptedSocket == SRT_INVALID_SOCK) { __android_log_print(ANDROID_LOG_ERROR, kTag, "SRT accept failed: %s", srt_getlasterror_str()); - disconnect(); - return false; + return kAcceptFailed; } - srt_setsockopt(acceptedSocket, 0, SRTO_SNDTIMEO, &sendTimeoutMs, sizeof sendTimeoutMs); - srt_setsockopt(acceptedSocket, 0, SRTO_TLPKTDROP, &tooLatePacketDrop, sizeof tooLatePacketDrop); - srt_setsockopt(acceptedSocket, 0, SRTO_PEERIDLETIMEO, &peerIdleTimeoutMs, sizeof peerIdleTimeoutMs); if (!setSocketForLifecycle(acceptedSocket, expectedLifecycleGeneration)) { srt_close(acceptedSocket); - return false; + return kAcceptCancelled; } logInfo("OBS connected to Android SRT listener"); - return true; + return kAcceptConnected; #else - (void)url; + (void)expectedLifecycleGeneration; logError("openstream_srt was built without libsrt. Rebuild with OPENSTREAM_ENABLE_LIBSRT=ON."); - return false; + return kAcceptFailed; #endif } @@ -764,8 +822,20 @@ class NativeSender { sendQueueBytes_ = 0; } std::lock_guard ioLock(ioMutex_); - const SRTSOCKET listenerSocket = takeListenerSocket(); + SRTSOCKET listenerSocket = SRT_INVALID_SOCK; + int listenerEpoll = -1; + { + std::lock_guard lock(socketMutex_); + listenerSocket = listener_socket_; + listener_socket_ = SRT_INVALID_SOCK; + listenerEpoll = listener_epoll_; + listener_epoll_ = -1; + } const SRTSOCKET socket = takeSocket(); + if (listenerEpoll >= 0) { + if (listenerSocket != SRT_INVALID_SOCK) srt_epoll_remove_usock(listenerEpoll, listenerSocket); + srt_epoll_release(listenerEpoll); + } if (listenerSocket != SRT_INVALID_SOCK) { srt_close(listenerSocket); } @@ -878,23 +948,23 @@ class NativeSender { return true; } - SRTSOCKET takeListenerSocket() { - std::lock_guard lock(socketMutex_); - const SRTSOCKET socket = listener_socket_; - listener_socket_ = SRT_INVALID_SOCK; - return socket; - } - void closeListenerSocket(SRTSOCKET socket) { bool ownsSocket = false; + int epoll = -1; { std::lock_guard lock(socketMutex_); if (listener_socket_ == socket) { listener_socket_ = SRT_INVALID_SOCK; + epoll = listener_epoll_; + listener_epoll_ = -1; ownsSocket = true; } } if (ownsSocket) { + if (epoll >= 0) { + srt_epoll_remove_usock(epoll, socket); + srt_epoll_release(epoll); + } srt_close(socket); } } @@ -903,6 +973,7 @@ class NativeSender { std::mutex ioMutex_; SRTSOCKET socket_ = SRT_INVALID_SOCK; SRTSOCKET listener_socket_ = SRT_INVALID_SOCK; + int listener_epoll_ = -1; uint64_t lifecycleGeneration_ = 0; bool srtStarted_ = false; #endif @@ -1003,7 +1074,7 @@ Java_dev_openstream_app_stream_SrtNativeBridge_connect( } extern "C" JNIEXPORT jboolean JNICALL -Java_dev_openstream_app_stream_SrtNativeBridge_listen( +Java_dev_openstream_app_stream_SrtNativeBridge_startListen( JNIEnv *env, jobject, jstring url, @@ -1038,21 +1109,47 @@ Java_dev_openstream_app_stream_SrtNativeBridge_listen( } g_state.sender.disconnect(); const bool connected = - g_state.sender.listen(urlString, static_cast(session_generation)); + g_state.sender.startListen(urlString, static_cast(session_generation)); bool published = false; { std::lock_guard lock(g_state.mediaMutex); if (static_cast(session_generation) == g_state.mediaSessionGeneration) { - g_state.connected = connected; + g_state.connected = false; published = connected; } } if (published) { - __android_log_print(ANDROID_LOG_INFO, kTag, "Accepted OBS SRT caller at %s", urlString.c_str()); + __android_log_print(ANDROID_LOG_INFO, kTag, "Listening for OBS SRT caller at %s", urlString.c_str()); } return published ? JNI_TRUE : JNI_FALSE; } +extern "C" JNIEXPORT jint JNICALL +Java_dev_openstream_app_stream_SrtNativeBridge_accept( + JNIEnv *, jobject, jlong session_generation) { + const uint64_t generation = static_cast(session_generation); + { + std::lock_guard lock(g_state.mediaMutex); + if (generation != g_state.mediaSessionGeneration) { + return NativeSender::kAcceptCancelled; + } + } + + const int result = g_state.sender.acceptPending(generation); + { + std::lock_guard lock(g_state.mediaMutex); + if (generation != g_state.mediaSessionGeneration) { + return NativeSender::kAcceptCancelled; + } + if (result == NativeSender::kAcceptConnected) { + g_state.connected = true; + } else if (result == NativeSender::kAcceptFailed) { + g_state.connected = false; + } + } + return result; +} + extern "C" JNIEXPORT jboolean JNICALL Java_dev_openstream_app_stream_SrtNativeBridge_sendVideo( JNIEnv *env, diff --git a/android/app/src/main/java/dev/openstream/app/MainActivity.kt b/android/app/src/main/java/dev/openstream/app/MainActivity.kt index 614f287e..b2f9eb6c 100644 --- a/android/app/src/main/java/dev/openstream/app/MainActivity.kt +++ b/android/app/src/main/java/dev/openstream/app/MainActivity.kt @@ -34,6 +34,8 @@ import dev.openstream.app.discovery.PhoneDiscoveryAdvertiser import dev.openstream.app.encoder.MediaCodecAudioEncoder import dev.openstream.app.encoder.MediaCodecVideoEncoder import dev.openstream.app.stream.ConnectionTarget +import dev.openstream.app.stream.ListenerAcceptCoordinator +import dev.openstream.app.stream.SrtAcceptResult import dev.openstream.app.stream.StreamConfig import dev.openstream.app.stream.SrtStreamClient @@ -68,6 +70,10 @@ class MainActivity : Activity() { private lateinit var phoneAdvertiser: PhoneDiscoveryAdvertiser private lateinit var obsDiscoveryClient: ObsDiscoveryClient private lateinit var controlServer: CameraControlServer + private val sessionWorker = SessionWorker { error -> + Log.e("shin", "Session lifecycle operation failed", error) + } + private val listenerAcceptCoordinator = ListenerAcceptCoordinator(sessionWorker, LISTENER_POLL_MS) private val streamConfig = StreamConfig.Default1080p30 private val mainHandler = Handler(Looper.getMainLooper()) @@ -78,11 +84,12 @@ class MainActivity : Activity() { @Volatile private var phoneConnected = false @Volatile private var selectedObsHost: String? = null @Volatile private var listenerThread: Thread? = null - @Volatile private var callerConnectThread: Thread? = null + @Volatile private var callerConnecting = false @Volatile private var callerGeneration = 0L @Volatile private var pendingListenerStart = false @Volatile private var listenerGeneration = 0L @Volatile private var activityStarted = false + @Volatile private var sessionWorkGeneration = 0L private var keepScreenOn = false private var displayOff = false private var originalBrightness = -1f @@ -185,7 +192,7 @@ class MainActivity : Activity() { onSwitchLens = { lens -> runOnUiThread { selectLens(lens) } }, onToggleTorch = { enabled -> runOnUiThread { torchOn = enabled - camera.setTorch(enabled) + sessionWorker.submit { camera.setTorch(enabled) } setTorchUi(enabled) }}, reservationProvider = { reservedBy }, @@ -208,6 +215,7 @@ class MainActivity : Activity() { cameraPreview.holder.addCallback(object : SurfaceHolder.Callback { override fun surfaceCreated(holder: SurfaceHolder) { + sessionWorkGeneration += 1 initializeLenses() startPreviewIfAllowed() startPhoneServerIfAllowed() @@ -217,10 +225,8 @@ class MainActivity : Activity() { adjustPreviewAspectRatio() } override fun surfaceDestroyed(holder: SurfaceHolder) { - // Close the camera before encoder teardown tries to rebuild a - // preview-only session against this now-invalid surface. - camera.stop() - stopPhoneServer(clearReservation = false, updateStatus = false) + sessionWorkGeneration += 1 + stopPhoneServer(clearReservation = false, updateStatus = false, stopCamera = true) } }) @@ -232,9 +238,11 @@ class MainActivity : Activity() { override fun onStart() { super.onStart() activityStarted = true - phoneAdvertiser.start() - obsDiscoveryClient.start() - controlServer.start() + sessionWorker.submit { + phoneAdvertiser.start() + obsDiscoveryClient.start() + controlServer.start() + } startPreviewIfAllowed() startPhoneServerIfAllowed() } @@ -255,6 +263,7 @@ class MainActivity : Activity() { override fun onStop() { activityStarted = false + sessionWorkGeneration += 1 cancelLensRestart() stopLiveDotAnimation() val session = phoneSessionState.snapshot @@ -263,16 +272,12 @@ class MainActivity : Activity() { ) { phoneSessionState.connectionLost(session.generation) } - // Never block the UI thread on network/thread joins (AGENTS.md 6). - // Flag flips are immediate; heavy teardown runs on a daemon thread. - // stopPhoneServer detects a non-UI thread and runs blocking teardown inline. - Thread({ - runCatching { camera.stop() } - runCatching { stopPhoneServer(clearReservation = false, updateStatus = false) } - runCatching { obsDiscoveryClient.stop() } - runCatching { phoneAdvertiser.stop() } - runCatching { controlServer.stop() } - }, "shinActivityStop").apply { isDaemon = true; start() } + stopPhoneServer( + clearReservation = false, + updateStatus = false, + stopCamera = true, + stopComponents = true, + ) super.onStop() } @@ -281,6 +286,7 @@ class MainActivity : Activity() { identifyHideRunnable?.let(mainHandler::removeCallbacks) identifyHideRunnable = null mainHandler.removeCallbacksAndMessages(null) + sessionWorker.close() super.onDestroy() } @@ -409,11 +415,13 @@ class MainActivity : Activity() { } return@post } - if (activeTargetName == null && callerConnectThread == null) return@post + if (activeTargetName == null && !callerConnecting) return@post phoneSessionState.connectionLost(phoneSessionState.snapshot.generation) stopStream(updateStatus = false) startPreviewIfAllowed() - startPhoneServerIfAllowed() + sessionWorker.submit { + mainHandler.post { startPhoneServerIfAllowed() } + } statusText.text = "Connection lost" statusText.setTextColor(getColor(R.color.os_warning)) statusDetail.text = getString(R.string.status_waiting) @@ -427,8 +435,9 @@ class MainActivity : Activity() { if (activeStreamBitrate == nextBitrate) return activeStreamBitrate = nextBitrate if (activeTargetName == null) { - encoder.stop() + val previousEncoder = encoder encoder = createVideoEncoder(activeStreamBitrate) + sessionWorker.submit { previousEncoder.stop() } } } } @@ -436,8 +445,11 @@ class MainActivity : Activity() { private fun setupGestureDetector() { scaleGestureDetector = ScaleGestureDetector(this, object : ScaleGestureDetector.SimpleOnScaleGestureListener() { override fun onScale(detector: ScaleGestureDetector): Boolean { - val newZoom = camera.scaleZoom(detector.scaleFactor) - showZoomLabel(newZoom) + val newZoom = camera.zoomRatio * detector.scaleFactor + sessionWorker.submit { + val appliedZoom = camera.setZoom(newZoom) + mainHandler.post { showZoomLabel(appliedZoom) } + } return true } }) @@ -453,11 +465,20 @@ class MainActivity : Activity() { private fun initializeLenses() { if (checkSelfPermission(Manifest.permission.CAMERA) != PackageManager.PERMISSION_GRANTED) return - availableLenses = camera.availableLenses() - if (currentLens !in availableLenses) { - currentLens = availableLenses.firstOrNull { it.isBackFacing } ?: availableLenses.first() + val generation = sessionWorkGeneration + sessionWorker.submitIfCurrent(generation, { sessionWorkGeneration }) { + val lenses = camera.availableLenses() + mainHandler.post { + if (sessionWorkGeneration != generation || + checkSelfPermission(Manifest.permission.CAMERA) != PackageManager.PERMISSION_GRANTED + ) return@post + availableLenses = lenses + if (currentLens !in availableLenses) { + currentLens = availableLenses.firstOrNull { it.isBackFacing } ?: availableLenses.first() + } + buildLensButtons() + } } - buildLensButtons() } private fun buildLensButtons() { @@ -493,14 +514,23 @@ class MainActivity : Activity() { } val wasStreaming = activeTargetName != null // Stop the encoder before switching cameras to avoid surface conflicts - if (wasStreaming) { - camera.stopStreaming() - encoder.stop() - } currentLens = lens - runCatching { camera.switchLens(lens) }.onFailure { error -> - Log.e("shin", "Lens switch failed", error) - statusDetail.text = error.message ?: "Lens switch failed" + val lensGeneration = ++sessionWorkGeneration + sessionWorker.submitIfCurrent(lensGeneration, { sessionWorkGeneration }) { + runCatching { + if (wasStreaming) { + camera.stopStreaming() + encoder.stop() + } + camera.switchLens(lens) + }.onFailure { error -> + Log.e("shin", "Lens switch failed", error) + mainHandler.post { + if (sessionWorkGeneration == lensGeneration) { + statusDetail.text = error.message ?: "Lens switch failed" + } + } + } } // If we were streaming, re-create the encoder and re-attach after the camera settles. // MediaCodecList probing is heavy; never run encoder.start() on the UI thread. @@ -509,22 +539,23 @@ class MainActivity : Activity() { val restart = Runnable { lensRestartRunnable = null if (!activityStarted || activeTargetName == null) return@Runnable - Thread({ - val result = runCatching { - // Single-flight with caller/listener paths via callerLifecycleLock. + sessionWorker.submitIfCurrent(lensGeneration, { sessionWorkGeneration }) { + if (!activityStarted || activeTargetName == null) return@submitIfCurrent + runCatching { synchronized(callerLifecycleLock) { encoder.start() camera.startStreaming(encoder.inputSurface()) } - } - result.onFailure { e -> - Log.e("shin", "Failed to restart encoder after lens switch", e) + }.onFailure { error -> + Log.e("shin", "Failed to restart encoder after lens switch", error) mainHandler.post { - statusText.text = "Encoder error" - statusDetail.text = e.message ?: "Unknown" + if (sessionWorkGeneration == lensGeneration) { + statusText.text = "Encoder error" + statusDetail.text = error.message ?: "Unknown" + } } } - }, "shinLensRestart").apply { isDaemon = true; start() } + } } lensRestartRunnable = restart mainHandler.postDelayed(restart, LENS_RESTART_DELAY_MS) @@ -569,7 +600,8 @@ class MainActivity : Activity() { // Only works on back-facing cameras if (currentLens.isFrontFacing) return torchOn = !torchOn - camera.setTorch(torchOn) + val enabled = torchOn + sessionWorker.submit { camera.setTorch(enabled) } setTorchUi(torchOn) } @@ -740,8 +772,11 @@ class MainActivity : Activity() { renderDisconnectVisibility() val generation = callerGeneration + 1 callerGeneration = generation - val thread = Thread({ + callerConnecting = true + val accepted = sessionWorker.submit { try { + // This operation runs after the queued teardown and owns the + // caller socket, codec, and camera transition as one serial unit. streamClient.connect( url = target.toSrtCallerUrl(), codecMime = encoder.codecName, @@ -749,17 +784,19 @@ class MainActivity : Activity() { height = streamConfig.height, fps = streamConfig.fps, ) + check(callerGeneration == generation) { "SRT caller connection was cancelled" } + check(phoneSessionState.snapshot.generation == sessionGeneration) { + "Phone session was superseded" + } synchronized(callerLifecycleLock) { - check(callerGeneration == generation) { "SRT caller connection was cancelled" } - check(phoneSessionState.snapshot.generation == sessionGeneration) { - "Phone session was superseded" - } encoder.start() startAudioIfAllowed() camera.startStreaming(encoder.inputSurface()) } mainHandler.post { - if (callerGeneration != generation) return@post + if (callerGeneration != generation || + phoneSessionState.snapshot.generation != sessionGeneration + ) return@post phoneSessionState.connected(sessionGeneration) activeTargetName = target.name mainHandler.removeCallbacks(statsTicker) @@ -770,7 +807,9 @@ class MainActivity : Activity() { synchronized(callerLifecycleLock) { if (callerGeneration == generation) { streamClient.disconnect() - stopActiveEncoding(updateStatus = false) + camera.stopStreaming() + encoder.stop() + audioEncoder.stop() } } mainHandler.post { @@ -782,15 +821,15 @@ class MainActivity : Activity() { statusDetail.text = error.message ?: "Unknown error" } } finally { - if (callerConnectThread === Thread.currentThread()) { - callerConnectThread = null - } + if (callerGeneration == generation) callerConnecting = false } - }, "shinPhoneSrtCaller").apply { - isDaemon = true } - callerConnectThread = thread - thread.start() + if (!accepted) { + callerConnecting = false + phoneSessionState.failed(sessionGeneration) + statusText.text = "Session busy" + statusDetail.text = "Wait for the current camera transition to finish" + } } private fun startAudioIfAllowed() { @@ -832,16 +871,23 @@ class MainActivity : Activity() { val attemptGeneration = phoneSessionState.snapshot.generation var connectedSessionGeneration: Long? = null val listenResult = runCatching { - streamClient.listen( - url = listenUrl, - codecMime = encoder.codecName, - width = streamConfig.width, - height = streamConfig.height, - fps = streamConfig.fps, + val acceptResult = listenerAcceptCoordinator.awaitAccepted( + expectedGeneration = generation, + currentGeneration = { listenerGeneration }, + startListening = { + streamClient.startListening( + url = listenUrl, + codecMime = encoder.codecName, + width = streamConfig.width, + height = streamConfig.height, + fps = streamConfig.fps, + ) + }, + acceptPending = streamClient::acceptPending, ) - if (!isListenerActive(generation)) { - streamClient.disconnect() - return@runCatching + if (!isListenerActive(generation)) return@runCatching + check(acceptResult == SrtAcceptResult.Connected) { + "SRT listener accept failed: $acceptResult" } phoneConnected = true val sessionGeneration = phoneSessionState.beginConnection() @@ -855,11 +901,30 @@ class MainActivity : Activity() { mainHandler.removeCallbacks(statsTicker) mainHandler.post(statsTicker) } - encoder.start() - startAudioIfAllowed() - camera.startStreaming(encoder.inputSurface()) - check(phoneSessionState.connected(sessionGeneration)) { - "Phone session was superseded" + sessionWorker.submitIfCurrent(generation, { listenerGeneration }) { + runCatching { + encoder.start() + startAudioIfAllowed() + camera.startStreaming(encoder.inputSurface()) + check(phoneSessionState.connected(sessionGeneration)) { + "Phone session was superseded" + } + }.onFailure { error -> + if (isListenerActive(generation)) { + phoneConnected = false + phoneSessionState.failed(sessionGeneration) + mainHandler.post { + if (!isListenerActive(generation)) return@post + statusText.text = "Listener error" + statusDetail.text = error.message ?: "Unknown error" + } + sessionWorker.submitCritical { + camera.stopStreaming() + encoder.stop() + audioEncoder.stop() + } + } + } } while (isListenerActive(generation) && phoneConnected) { Thread.sleep(LISTENER_POLL_MS) @@ -881,7 +946,13 @@ class MainActivity : Activity() { } } - stopActiveEncoding(updateStatus = false) + if (isListenerActive(generation)) { + sessionWorker.submitCritical { + camera.stopStreaming() + encoder.stop() + audioEncoder.stop() + } + } phoneConnected = false activeTargetName = null if (isListenerActive(generation)) { @@ -921,12 +992,13 @@ class MainActivity : Activity() { private fun stopPhoneServer( clearReservation: Boolean = true, updateStatus: Boolean = true, + stopCamera: Boolean = false, + stopComponents: Boolean = false, ) { - // Fast, non-blocking section: flip flags on the caller thread so new - // connects/listens cannot start. Heavy teardown (native disconnect, - // thread joins, encoder stop) runs off the UI thread per AGENTS.md 6. + // Flip flags and update views on the caller thread. Socket, camera, + // codec, and thread teardown runs on the bounded session worker. callerGeneration += 1 - callerConnectThread?.interrupt() + callerConnecting = false pendingListenerStart = false listenerGeneration += 1 phoneServerRunning = false @@ -942,7 +1014,7 @@ class MainActivity : Activity() { // covers both pending (advertised) and confirmed selections. renderDisconnectVisibility() - val blockingWork = { + val blockingWork: () -> Unit = { streamClient.disconnect() if (thread != null && thread !== Thread.currentThread()) { runCatching { thread.join(LISTENER_STOP_TIMEOUT_MS) } @@ -961,6 +1033,12 @@ class MainActivity : Activity() { camera.stopStreaming() encoder.stop() audioEncoder.stop() + if (stopCamera) camera.stop() + if (stopComponents) { + obsDiscoveryClient.stop() + phoneAdvertiser.stop() + controlServer.stop() + } if (needsStatus) { mainHandler.post { statusText.text = getString(R.string.status_stopped) @@ -973,37 +1051,31 @@ class MainActivity : Activity() { if (pendingListenerStart) startPhoneServerIfAllowed() } } - if (Looper.myLooper() == Looper.getMainLooper()) { - Thread({ blockingWork() }, "shinStopServer").apply { isDaemon = true; start() } - } else { - blockingWork() - } + sessionWorker.submitCritical(blockingWork) } private fun stopStream(updateStatus: Boolean = true) { callerGeneration += 1 - callerConnectThread?.interrupt() + callerConnecting = false activeTargetName = null mainHandler.removeCallbacks(statsTicker) phoneConnected = false - streamClient.disconnect() - synchronized(callerLifecycleLock) { - stopActiveEncoding(updateStatus) + sessionWorker.submitCritical { + streamClient.disconnect() + synchronized(callerLifecycleLock) { + cancelLensRestart() + camera.stopStreaming() + encoder.stop() + audioEncoder.stop() + } + if (updateStatus) mainHandler.post { + statusText.text = getString(R.string.status_stopped) + statusDetail.text = "Camera preview remains active" + } } hideLiveState() } - private fun stopActiveEncoding(updateStatus: Boolean = true) { - cancelLensRestart() - camera.stopStreaming() - encoder.stop() - audioEncoder.stop() - if (updateStatus) { - statusText.text = getString(R.string.status_stopped) - statusDetail.text = "Camera preview remains active" - } - } - // ─────────────────────────── Live state UI ─────────────────────────── @Synchronized @@ -1266,7 +1338,12 @@ class MainActivity : Activity() { if (checkSelfPermission(Manifest.permission.CAMERA) == PackageManager.PERMISSION_GRANTED && cameraPreview.holder.surface.isValid ) { - camera.startPreview() + val generation = sessionWorkGeneration + sessionWorker.submitIfCurrent(generation, { sessionWorkGeneration }) { + if (checkSelfPermission(Manifest.permission.CAMERA) == PackageManager.PERMISSION_GRANTED && + cameraPreview.holder.surface.isValid + ) camera.startPreview() + } } } @@ -1342,8 +1419,8 @@ class MainActivity : Activity() { stopPhoneServer(clearReservation = false) } // Re-create the advertiser with new port - phoneAdvertiser.stop() - phoneAdvertiser = PhoneDiscoveryAdvertiser( + val previousAdvertiser = phoneAdvertiser + val replacementAdvertiser = PhoneDiscoveryAdvertiser( context = this, config = streamConfig, port = currentPort, @@ -1351,7 +1428,11 @@ class MainActivity : Activity() { reservedByProvider = { advertisedReservationId }, selectedObsHostProvider = { selectedObsHost }, ) - phoneAdvertiser.start() + phoneAdvertiser = replacementAdvertiser + sessionWorker.submit { + previousAdvertiser.stop() + replacementAdvertiser.start() + } if (wasListenerMode) { startPhoneServerIfAllowed() } diff --git a/android/app/src/main/java/dev/openstream/app/SessionWorker.kt b/android/app/src/main/java/dev/openstream/app/SessionWorker.kt new file mode 100644 index 00000000..d2a15d12 --- /dev/null +++ b/android/app/src/main/java/dev/openstream/app/SessionWorker.kt @@ -0,0 +1,109 @@ +package dev.openstream.app + +import android.util.Log +import java.util.concurrent.ArrayBlockingQueue +import java.util.concurrent.Callable +import java.util.concurrent.ExecutionException +import java.util.concurrent.FutureTask +import java.util.concurrent.RejectedExecutionException +import java.util.concurrent.ThreadPoolExecutor +import java.util.concurrent.TimeUnit + +/** + * Serializes camera, codec, and transport lifecycle work away from the UI thread. + * Normal work has [queueCapacity] waiting slots. One extra waiting slot is + * reserved for critical teardown, so normal saturation cannot reject Stop or + * disconnect cleanup. Work stays FIFO; normal overflow rejects the newest + * normal task. Critical submissions can still be rejected if the entire + * bounded queue, including its reserved slot, is already occupied. + */ +internal class SessionWorker( + queueCapacity: Int = 32, + private val onFailure: (Throwable) -> Unit = { error -> + Log.e("shinSession", "Session work failed", error) + }, +) : AutoCloseable { + private val normalQueueCapacity = queueCapacity.coerceAtLeast(1) + private val submissionLock = Any() + private val executor = ThreadPoolExecutor( + 1, + 1, + 0L, + TimeUnit.MILLISECONDS, + ArrayBlockingQueue(normalQueueCapacity + 1), + { task -> Thread(task, "shinSessionWorker").apply { isDaemon = true } }, + ) + + init { + executor.setRejectedExecutionHandler { _, _ -> + throw RejectedExecutionException("Session worker queue is full") + } + } + + fun submit(work: () -> Unit): Boolean = enqueue(work, critical = false) + + /** Queues teardown in the reserved slot when all normal waiting slots are occupied. */ + fun submitCritical(work: () -> Unit): Boolean = enqueue(work, critical = true) + + private fun enqueue(work: () -> Unit, critical: Boolean): Boolean = synchronized(submissionLock) { + val queued = executor.queue.size + val limit = if (critical) normalQueueCapacity + 1 else normalQueueCapacity + if (queued >= limit) { + onFailure(RejectedExecutionException( + if (critical) "Critical session worker queue is full" + else "Session worker queue is full", + )) + return@synchronized false + } + try { + executor.execute { + try { + work() + } catch (error: Throwable) { + onFailure(error) + } + } + true + } catch (error: RejectedExecutionException) { + onFailure(error) + false + } + } + + fun submitIfCurrent( + generation: Long, + currentGeneration: () -> Long, + work: () -> Unit, + ): Boolean = submit { + if (currentGeneration() == generation) work() + } + + /** Waits for a short worker operation; callers must be non-UI threads. */ + fun submitAndWait(work: () -> T): T { + val task = FutureTask(Callable { work() }) + try { + synchronized(submissionLock) { + if (executor.queue.size >= normalQueueCapacity) { + throw RejectedExecutionException("Session worker queue is full") + } + executor.execute(task) + } + } catch (error: RejectedExecutionException) { + onFailure(error) + throw error + } + return try { + task.get() + } catch (error: InterruptedException) { + task.cancel(false) + Thread.currentThread().interrupt() + throw error + } catch (error: ExecutionException) { + throw error.cause ?: error + } + } + + override fun close() { + executor.shutdown() + } +} diff --git a/android/app/src/main/java/dev/openstream/app/stream/ListenerAcceptCoordinator.kt b/android/app/src/main/java/dev/openstream/app/stream/ListenerAcceptCoordinator.kt new file mode 100644 index 00000000..48f6df61 --- /dev/null +++ b/android/app/src/main/java/dev/openstream/app/stream/ListenerAcceptCoordinator.kt @@ -0,0 +1,29 @@ +package dev.openstream.app.stream + +import dev.openstream.app.SessionWorker + +/** Coordinates short listener setup/accept operations through the session worker. */ +internal class ListenerAcceptCoordinator( + private val sessionWorker: SessionWorker, + private val pollIntervalMs: Long, + private val waitForNextPoll: (Long) -> Unit = { Thread.sleep(it) }, +) { + fun awaitAccepted( + expectedGeneration: Long, + currentGeneration: () -> Long, + startListening: () -> Unit, + acceptPending: () -> SrtAcceptResult, + ): SrtAcceptResult { + if (currentGeneration() != expectedGeneration) return SrtAcceptResult.Cancelled + sessionWorker.submitAndWait { + if (currentGeneration() == expectedGeneration) startListening() + } + while (currentGeneration() == expectedGeneration) { + val result = sessionWorker.submitAndWait(acceptPending) + if (currentGeneration() != expectedGeneration) return SrtAcceptResult.Cancelled + if (result != SrtAcceptResult.Pending) return result + waitForNextPoll(pollIntervalMs) + } + return SrtAcceptResult.Cancelled + } +} diff --git a/android/app/src/main/java/dev/openstream/app/stream/SrtStreamClient.kt b/android/app/src/main/java/dev/openstream/app/stream/SrtStreamClient.kt index c762f074..852ce9c1 100644 --- a/android/app/src/main/java/dev/openstream/app/stream/SrtStreamClient.kt +++ b/android/app/src/main/java/dev/openstream/app/stream/SrtStreamClient.kt @@ -22,6 +22,8 @@ data class SrtSendResult( val recoveryRequired: Boolean = false, ) +enum class SrtAcceptResult { Pending, Connected, Failed, Cancelled } + class SrtStreamClient { @Volatile private var connected = false private val sessionGeneration = AtomicLong() @@ -56,16 +58,34 @@ class SrtStreamClient { } } - fun listen(url: String, codecMime: String, width: Int, height: Int, fps: Int) { + fun startListening(url: String, codecMime: String, width: Int, height: Int, fps: Int) { require(url.startsWith("srt://")) { "shin expects an SRT URL" } synchronized(operationLock) { - // Caller and listener share one native SRT transport. establishSession() - // performs an explicit disconnect-before-relisten via - // beginSession(generation), so a stale listener socket can never - // survive into the new session. Callers must not start a second - // listen without going through establishSession/disconnect. - establishSession("listener") { generation -> - SrtNativeBridge.listen(url, codecMime, width, height, fps, generation) + establishSession("listener", connectedOnSuccess = false) { generation -> + SrtNativeBridge.startListen(url, codecMime, width, height, fps, generation) + } + } + } + + /** Polls the native listener once. The JNI operation returns immediately when no peer is ready. */ + fun acceptPending(): SrtAcceptResult { + val generation = synchronized(stateLock) { sessionGeneration.get() } + val result = SrtNativeBridge.accept(generation) + return when (result) { + ACCEPT_PENDING -> SrtAcceptResult.Pending + ACCEPT_CONNECTED -> { + synchronized(stateLock) { + if (sessionGeneration.get() != generation) return SrtAcceptResult.Cancelled + connected = true + } + SrtAcceptResult.Connected + } + ACCEPT_CANCELLED -> SrtAcceptResult.Cancelled + else -> { + synchronized(stateLock) { + if (sessionGeneration.get() == generation) connected = false + } + SrtAcceptResult.Failed } } } @@ -196,7 +216,11 @@ class SrtStreamClient { }.getOrElse { listOf(url) } } - private inline fun establishSession(operationName: String, nativeOperation: (Long) -> Boolean) { + private inline fun establishSession( + operationName: String, + connectedOnSuccess: Boolean = true, + nativeOperation: (Long) -> Boolean, + ) { check(SrtNativeBridge.isAvailable) { "Native SRT library unavailable: ${SrtNativeBridge.loadError?.message ?: "missing"}" } @@ -213,7 +237,7 @@ class SrtStreamClient { } else { check(didConnect) { "Native SRT bridge failed to $operationName" } resetStats() - connected = true + connected = connectedOnSuccess false } } @@ -255,6 +279,9 @@ class SrtStreamClient { } companion object { + private const val ACCEPT_PENDING = 0 + private const val ACCEPT_CONNECTED = 1 + private const val ACCEPT_CANCELLED = 2 private const val BUFFER_FLAG_KEY_FRAME = 1 private const val BUFFER_FLAG_CODEC_CONFIG = 2 } @@ -301,7 +328,7 @@ private object SrtNativeBridge { ): Boolean = guardOr(false) { nativeConnect(url, codecMime, width, height, fps, sessionGeneration) } - fun listen( + fun startListen( url: String, codecMime: String, width: Int, @@ -309,8 +336,9 @@ private object SrtNativeBridge { fps: Int, sessionGeneration: Long, ): Boolean = guardOr(false) { - nativeListen(url, codecMime, width, height, fps, sessionGeneration) + nativeStartListen(url, codecMime, width, height, fps, sessionGeneration) } + fun accept(sessionGeneration: Long): Int = guardOr(-1) { nativeAccept(sessionGeneration) } fun sendVideo( data: ByteArray, presentationTimeUs: Long, @@ -340,7 +368,7 @@ private object SrtNativeBridge { fps: Int, sessionGeneration: Long, ): Boolean - private external fun nativeListen( + private external fun nativeStartListen( url: String, codecMime: String, width: Int, @@ -348,6 +376,7 @@ private object SrtNativeBridge { fps: Int, sessionGeneration: Long, ): Boolean + private external fun nativeAccept(sessionGeneration: Long): Int private external fun nativeSendVideo( data: ByteArray, presentationTimeUs: Long, diff --git a/android/app/src/test/java/dev/openstream/app/SessionWorkerTest.kt b/android/app/src/test/java/dev/openstream/app/SessionWorkerTest.kt new file mode 100644 index 00000000..c1c531a0 --- /dev/null +++ b/android/app/src/test/java/dev/openstream/app/SessionWorkerTest.kt @@ -0,0 +1,187 @@ +package dev.openstream.app + +import dev.openstream.app.stream.SrtAcceptResult +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Test +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger + +class SessionWorkerTest { + @Test + fun rapidStartStopRunsAfterTheActiveOperationInOrder() { + val entered = CountDownLatch(1) + val release = CountDownLatch(1) + val completed = CountDownLatch(2) + val observed = mutableListOf() + val worker = SessionWorker() + try { + worker.submit { + entered.countDown() + release.await(2, TimeUnit.SECONDS) + synchronized(observed) { observed += "running-start" } + completed.countDown() + } + assertTrue(entered.await(2, TimeUnit.SECONDS)) + worker.submit { synchronized(observed) { observed += "latest-stop" }; completed.countDown() } + release.countDown() + + assertTrue(completed.await(2, TimeUnit.SECONDS)) + assertEquals(listOf("running-start", "latest-stop"), observed) + } finally { + release.countDown() + worker.close() + } + } + + @Test + fun staleSurfaceAndLensCallbacksCannotRunAfterGenerationChanges() { + var generation = 4L + val entered = CountDownLatch(1) + val release = CountDownLatch(1) + val secondEntered = CountDownLatch(1) + val secondRelease = CountDownLatch(1) + val completed = CountDownLatch(1) + val observed = mutableListOf() + val worker = SessionWorker() + try { + worker.submit { + entered.countDown() + release.await(2, TimeUnit.SECONDS) + } + assertTrue(entered.await(2, TimeUnit.SECONDS)) + worker.submit { synchronized(observed) { observed += "preview" } } + release.countDown() + worker.submit { + secondEntered.countDown() + secondRelease.await(2, TimeUnit.SECONDS) + } + assertTrue(secondEntered.await(2, TimeUnit.SECONDS)) + worker.submitIfCurrent(4L, { generation }) { + synchronized(observed) { observed += "stale-lens-restart" } + } + generation = 5L + worker.submit { completed.countDown() } + secondRelease.countDown() + + assertTrue(completed.await(2, TimeUnit.SECONDS)) + assertEquals(listOf("preview"), observed) + } finally { + release.countDown() + secondRelease.countDown() + worker.close() + } + } + + @Test + fun surfaceLensSettingsAndPreviewTransitionsRunSerially() { + val active = AtomicInteger() + val overlap = AtomicInteger() + val completed = CountDownLatch(4) + val worker = SessionWorker() + try { + listOf("surface-destroy", "lens-change", "settings-connect", "preview-recreate") + .forEach { + worker.submit { + if (active.incrementAndGet() != 1) overlap.incrementAndGet() + Thread.sleep(10) + active.decrementAndGet() + completed.countDown() + } + } + + assertTrue(completed.await(2, TimeUnit.SECONDS)) + assertEquals(0, overlap.get()) + } finally { + worker.close() + } + } + + @Test + fun fullBoundedQueueRejectsNewestWorkWithoutDroppingAcceptedStop() { + val entered = CountDownLatch(1) + val release = CountDownLatch(1) + val stopped = CountDownLatch(1) + val rejected = AtomicInteger() + val worker = SessionWorker(queueCapacity = 1, onFailure = { rejected.incrementAndGet() }) + try { + worker.submit { + entered.countDown() + release.await(2, TimeUnit.SECONDS) + } + assertTrue(entered.await(2, TimeUnit.SECONDS)) + assertTrue(worker.submit { stopped.countDown() }) + assertEquals(false, worker.submit { error("must be rejected") }) + release.countDown() + + assertTrue(stopped.await(2, TimeUnit.SECONDS)) + assertEquals(1, rejected.get()) + } finally { + release.countDown() + worker.close() + } + } + + @Test + fun criticalTeardownUsesReservedSlotWhenNormalQueueIsFull() { + val entered = CountDownLatch(1) + val release = CountDownLatch(1) + val completed = CountDownLatch(2) + val observed = mutableListOf() + val rejected = AtomicInteger() + val worker = SessionWorker(queueCapacity = 1, onFailure = { rejected.incrementAndGet() }) + try { + assertTrue(worker.submit { + entered.countDown() + release.await(2, TimeUnit.SECONDS) + }) + assertTrue(entered.await(2, TimeUnit.SECONDS)) + assertTrue(worker.submit { + synchronized(observed) { observed += "normal" } + completed.countDown() + }) + + // The normal waiting slot is occupied. A critical Stop still has + // its reserved slot and remains serialized behind accepted work. + assertTrue(worker.submitCritical { + synchronized(observed) { observed += "teardown" } + completed.countDown() + }) + assertEquals(false, worker.submit { error("normal overflow must reject newest") }) + release.countDown() + + assertTrue(completed.await(2, TimeUnit.SECONDS)) + assertEquals(listOf("normal", "teardown"), observed) + assertEquals(1, rejected.get()) + } finally { + release.countDown() + worker.close() + } + } + + @Test + fun nonBlockingAcceptPollYieldsTheWorkerForDisconnectBetweenPolls() { + var listenerActive = true + val worker = SessionWorker() + try { + val firstPoll = worker.submitAndWait { + if (listenerActive) SrtAcceptResult.Pending else SrtAcceptResult.Cancelled + } + assertEquals(SrtAcceptResult.Pending, firstPoll) + + val teardown = worker.submitAndWait { + listenerActive = false + "disconnected" + } + val nextPoll = worker.submitAndWait { + if (!listenerActive) SrtAcceptResult.Cancelled else SrtAcceptResult.Pending + } + + assertEquals("disconnected", teardown) + assertEquals(SrtAcceptResult.Cancelled, nextPoll) + } finally { + worker.close() + } + } +} diff --git a/android/app/src/test/java/dev/openstream/app/stream/ListenerAcceptCoordinatorTest.kt b/android/app/src/test/java/dev/openstream/app/stream/ListenerAcceptCoordinatorTest.kt new file mode 100644 index 00000000..d8468d61 --- /dev/null +++ b/android/app/src/test/java/dev/openstream/app/stream/ListenerAcceptCoordinatorTest.kt @@ -0,0 +1,70 @@ +package dev.openstream.app.stream + +import dev.openstream.app.SessionWorker +import org.junit.Assert.assertEquals +import org.junit.Test + +class ListenerAcceptCoordinatorTest { + @Test + fun pendingAcceptYieldsWorkerToDisconnectBeforeTheNextPoll() { + var generation = 7L + val events = mutableListOf() + val worker = SessionWorker() + val coordinator = ListenerAcceptCoordinator( + sessionWorker = worker, + pollIntervalMs = 250L, + waitForNextPoll = { delay -> + assertEquals(250L, delay) + worker.submitAndWait { + events += "disconnect" + generation = 8L + } + }, + ) + try { + val result = coordinator.awaitAccepted( + expectedGeneration = 7L, + currentGeneration = { generation }, + startListening = { events += "listen" }, + acceptPending = { + events += "poll" + SrtAcceptResult.Pending + }, + ) + + assertEquals(SrtAcceptResult.Cancelled, result) + assertEquals(listOf("listen", "poll", "disconnect"), events) + } finally { + worker.close() + } + } + + @Test + fun listenerReturnsOnlyAfterAWorkerPollAcceptsThePeer() { + val generation = 3L + var polls = 0 + var startThread = "" + var acceptThread = "" + val worker = SessionWorker() + val coordinator = ListenerAcceptCoordinator(worker, pollIntervalMs = 0L, waitForNextPoll = {}) + try { + val result = coordinator.awaitAccepted( + expectedGeneration = 3L, + currentGeneration = { generation }, + startListening = { startThread = Thread.currentThread().name }, + acceptPending = { + acceptThread = Thread.currentThread().name + polls += 1 + if (polls == 1) SrtAcceptResult.Pending else SrtAcceptResult.Connected + }, + ) + + assertEquals(SrtAcceptResult.Connected, result) + assertEquals(2, polls) + assertEquals("shinSessionWorker", startThread) + assertEquals("shinSessionWorker", acceptThread) + } finally { + worker.close() + } + } +} diff --git a/tests/test_prod_hardening_contract.py b/tests/test_prod_hardening_contract.py index b8c4ebdc..45b8d39a 100644 --- a/tests/test_prod_hardening_contract.py +++ b/tests/test_prod_hardening_contract.py @@ -45,9 +45,22 @@ def test_native_load_is_guarded(): def test_teardown_never_blocks_ui_thread(): app = read("android/app/src/main/java/dev/openstream/app/MainActivity.kt") - assert "shinActivityStop" in app or "shinStopServer" in app - assert "Thread(" in app - assert "Looper.getMainLooper" in app + worker = read("android/app/src/main/java/dev/openstream/app/SessionWorker.kt") + stop_server = app[app.index("private fun stopPhoneServer(") : app.index("private fun stopStream(")] + + # MainActivity owns a bounded serial worker, and teardown is submitted as a + # unit so native disconnect/camera/codec stop do not run in the UI callback. + assert "private val sessionWorker = SessionWorker" in app + assert "sessionWorker.submitCritical(blockingWork)" in stop_server + assert "streamClient.disconnect()" in stop_server + assert "camera.stopStreaming()" in stop_server + assert "encoder.stop()" in stop_server + assert "ArrayBlockingQueue(normalQueueCapacity + 1)" in worker + assert "queueCapacity: Int = 32" in worker + assert "Session worker queue is full" in worker + assert "fun submitCritical(work: () -> Unit)" in worker + assert "sessionWorker.submitCritical {" in app + assert "fun submitIfCurrent(" in worker def test_camera_manager_is_null_safe(): diff --git a/tests/test_srt_contract.py b/tests/test_srt_contract.py index 909f020e..b14eb7dc 100644 --- a/tests/test_srt_contract.py +++ b/tests/test_srt_contract.py @@ -44,15 +44,20 @@ def test_srt_runtime_lifetime_is_process_scoped_and_disconnect_safe(): source = SRT_NATIVE.read_text(encoding="utf-8") # NativeSender owns one libsrt runtime reference for its full lifetime. - # connect()/listen() must never acquire or release that global reference, - # so disconnect cannot tear the runtime down while setup is in progress. + # connect(), startListen(), and acceptPending() must never acquire or + # release that global reference, so disconnect cannot tear down runtime + # ownership while setup or an accept poll is in progress. constructor_start = source.index("NativeSender()") destructor_start = source.index("~NativeSender()", constructor_start) connect_start = source.index("bool connect(") constructor = source[constructor_start:destructor_start] destructor = source[destructor_start:connect_start] - connect = source[connect_start : source.index("bool listen(")] - listen = source[source.index("bool listen(") : source.index("bool sendNow(")] + start_listen_start = source.index("bool startListen(", connect_start) + accept_start = source.index("int acceptPending(", start_listen_start) + send_start = source.index("bool sendNow(", accept_start) + connect = source[connect_start:start_listen_start] + start_listen = source[start_listen_start:accept_start] + accept_pending = source[accept_start:send_start] disconnect = source[source.index("void disconnect()") : source.index("private:", source.index("void disconnect()"))] assert "srt_startup()" in constructor @@ -60,24 +65,29 @@ def test_srt_runtime_lifetime_is_process_scoped_and_disconnect_safe(): assert "srt_cleanup()" not in constructor assert "srt_cleanup();" in destructor assert "srt_startup()" not in connect - assert "srt_startup()" not in listen + assert "srt_startup()" not in start_listen + assert "srt_startup()" not in accept_pending assert "srt_cleanup()" not in connect - assert "srt_cleanup()" not in listen + assert "srt_cleanup()" not in start_listen + assert "srt_cleanup()" not in accept_pending assert "srt_cleanup()" not in disconnect # Connection attempts still fail cleanly if process-level startup failed. assert connect.count("if (!srtStarted_)") == 1 - assert listen.count("if (!srtStarted_)") == 1 + assert start_listen.count("if (!srtStarted_)") == 1 - # Post-startup failures close their sockets/session state via disconnect, - # without changing the process-scoped libsrt runtime ownership. + # Setup failures close their sockets/session state via disconnect, without + # changing process-scoped runtime ownership. Accept polling is separate: + # it reports errors to the coordinator, which schedules session teardown. assert "Could not create SRT socket\");\n disconnect();\n return false;" in connect assert "Could not resolve SRT host\");\n disconnect();\n return false;" in connect assert "SRT connect failed: %s\", srt_getlasterror_str());\n disconnect();\n return false;" in connect - assert "Could not create SRT listener socket\");\n disconnect();\n return false;" in listen - assert "SRT bind failed: %s\", srt_getlasterror_str());\n disconnect();\n return false;" in listen - assert "SRT listen failed: %s\", srt_getlasterror_str());\n disconnect();\n return false;" in listen - assert "SRT accept failed: %s\", srt_getlasterror_str());\n disconnect();\n return false;" in listen + assert "Could not create SRT listener socket\");\n disconnect();\n return false;" in start_listen + assert "SRT bind failed: %s\", srt_getlasterror_str());\n disconnect();\n return false;" in start_listen + assert "SRT listen failed: %s\", srt_getlasterror_str());\n disconnect();\n return false;" in start_listen + assert "SRT accept failed: %s\", srt_getlasterror_str());" in accept_pending + assert "srt_epoll_uwait(epoll, &event, 1, 0)" in accept_pending + assert accept_pending.index("event.events & SRT_EPOLL_IN") < accept_pending.index("srt_accept(") def test_native_connect_and_listen_publish_only_current_generation(): @@ -86,7 +96,7 @@ def test_native_connect_and_listen_publish_only_current_generation(): establish = block_after(kotlin, "private inline fun establishSession") assert "SrtNativeBridge.beginSession(generation)" in establish assert "SrtNativeBridge.connect(candidateUrl, codecMime, width, height, fps, generation)" in kotlin - assert "SrtNativeBridge.listen(url, codecMime, width, height, fps, generation)" in kotlin + assert "SrtNativeBridge.startListen(url, codecMime, width, height, fps, generation)" in kotlin publish = block_after(native, "bool setSocketForLifecycle") assert "std::lock_guard lock(socketMutex_)" in publish @@ -99,9 +109,20 @@ def test_native_connect_and_listen_publish_only_current_generation(): connect = block_after(native, "bool connect(const std::string &url, uint64_t expectedLifecycleGeneration)") assert connect.index("setSocketForLifecycle(socket, expectedLifecycleGeneration)") < connect.index("srt_connect(") - listen = block_after(native, "bool listen(const std::string &url, uint64_t expectedLifecycleGeneration)") - assert "setListenerSocketForLifecycle(listenerSocket, expectedLifecycleGeneration)" in listen - assert "setSocketForLifecycle(acceptedSocket, expectedLifecycleGeneration)" in listen + start_listen = block_after(native, "bool startListen(const std::string &url, uint64_t expectedLifecycleGeneration)") + accept_pending = block_after(native, "int acceptPending(uint64_t expectedLifecycleGeneration)") + assert "setListenerSocketForLifecycle(listenerSocket, expectedLifecycleGeneration)" in start_listen + assert "setSocketForLifecycle(acceptedSocket, expectedLifecycleGeneration)" in accept_pending + assert accept_pending.index("lifecycleGeneration_ != expectedLifecycleGeneration") < accept_pending.index("srt_epoll_uwait(") + assert accept_pending.index("srt_epoll_uwait(") < accept_pending.index("srt_accept(") + + jni_start_listen = block_after(native, "Java_dev_openstream_app_stream_SrtNativeBridge_startListen") + assert "g_state.sender.startListen(urlString, static_cast(session_generation))" in jni_start_listen + assert jni_start_listen.index("session_generation) != g_state.mediaSessionGeneration") < jni_start_listen.index("g_state.sender.startListen(") + jni_accept = block_after(native, "Java_dev_openstream_app_stream_SrtNativeBridge_accept") + assert "g_state.sender.acceptPending(generation)" in jni_accept + assert jni_accept.index("generation != g_state.mediaSessionGeneration") < jni_accept.index("g_state.sender.acceptPending(") + assert jni_accept.rindex("generation != g_state.mediaSessionGeneration") > jni_accept.index("g_state.sender.acceptPending(") def test_disconnect_invalidates_generation_before_native_teardown():