diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamCaptureEngine.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamCaptureEngine.kt index 5cf5cb29..d4387528 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamCaptureEngine.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamCaptureEngine.kt @@ -70,6 +70,9 @@ class StreamCaptureEngine( private var opening = false private var openingGeneration = 0L private var releaseDone: (() -> Unit)? = null + private var videoReleasing = false + private var videoReleaseSince = 0L + private var slowReleaseReported = false private var generation = 0L private var videoRevision = 0L private var videoRotation = 0 @@ -286,7 +289,7 @@ class StreamCaptureEngine( } catch (e: Exception) { captureError(token, "video setup: ${e.message}") } } - private fun current(token: Long) = token == generation && !stopping && !stopped && !terminalFailure && releaseDone == null + private fun current(token: Long) = token == generation && !stopping && !stopped && !terminalFailure && !videoReleasing private fun cameraFailure(device: CameraDevice, token: Long, detail: String) { if (device in closedCameras) { LogManager.addLog("WARN", "video", "closed capture camera error ignored id=$id token=$token: $detail"); return } @@ -487,39 +490,99 @@ class StreamCaptureEngine( } /** Новая камера открывается только после onClosed; старые codec callbacks уже отсечены. */ - private fun releaseVideo(done: (Boolean) -> Unit) { - if (releaseDone != null) { releaseDone = done; return } - val token = ++generation - sessionGeneration++ - preview.update(preview.state.value.format) - releaseDone = done - try { session?.close() } catch (e: Exception) { LogManager.addLog("WARN", "video", "camera session close id=$id: ${e.message}") } - session = null - try { camera?.close() } catch (e: Exception) { LogManager.addLog("WARN", "video", "camera close id=$id: ${e.message}") } - if (camera == null && !opening) finishRelease(true) - else handler.postDelayed({ - if (releaseDone != null && generation == token) { - LogManager.addLog("ERROR", "video", "camera close timeout id=$id opening=$opening"); finishRelease(false) - } - }, 3000) + private fun releaseVideo(done: () -> Unit) { + if (!videoReleasing) { + ++generation; sessionGeneration++; videoReleasing = true + videoReleaseSince = System.nanoTime(); slowReleaseReported = false + preview.update(preview.state.value.format) + } + releaseDone = done; retained.add(this); queueCleanup(0) } - private fun finishRelease(closed: Boolean) { - val done = releaseDone ?: return - try { encoder?.stop() } catch (e: Exception) { LogManager.addLog("WARN", "video", "encoder stop id=$id: ${e.message}") } - try { encoder?.release() } catch (e: Exception) { LogManager.addLog("WARN", "video", "encoder release id=$id: ${e.message}") } - encoder = null; input?.release(); input = null; partial.reset(); partialConfig.reset(); csd = byteArrayOf(); releaseDone = null - LogManager.videoTrace("video resources released id=$id closed=$closed generation=$generation") - done(closed) + /** Только capture worker: независимый cleanup, refs снимаются после возврата SDK либо callback закрытия. */ + private fun cleanupResources() { + if (stopping) { + if (!videoReleasing) releaseVideo {} + releaseDone = null; audioRunning = false + try { observeOrientation(false) } catch (e: Exception) { + LogManager.addLog("ERROR", "video", "capture orientation release failed id=$id; retained: $e") + } + preview.close(); previewTarget?.release(); previewTarget = null + try { audioCapture?.close(); audioCapture = null } catch (e: Exception) { + LogManager.addLog("ERROR", "stream", "capture microphone subscription close failed id=$id; retained: $e") + } + val worker = audioThread + if (worker?.isAlive == true) try { worker.join(250) } catch (e: InterruptedException) { + Thread.currentThread().interrupt() + LogManager.addLog("ERROR", "stream", "capture audio join interrupted id=$id; worker retained: $e") + } + if (worker?.isAlive != true) audioThread = null + } + if (videoReleasing) { + closingSessions.addAll(ownedSessions.keys); camera?.let { closingCameras.add(it) } + } + for (owned in closingSessions.toList()) try { owned.close() } catch (e: Exception) { + LogManager.addLog("ERROR", "video", "capture session close failed id=$id; retained: $e") + } + for (owned in closingCameras.toList()) try { owned.close() } catch (e: Exception) { + LogManager.addLog("ERROR", "video", "capture camera close failed id=$id device=${owned.id}; retained: $e") + } + val cameraClosed = !opening && camera == null && closingCameras.isEmpty() && ownedSessions.isEmpty() && cameraSessions.isEmpty() + if (videoReleasing && cameraClosed) { + try { encoderResources?.close(); encoderResources = null } catch (e: Exception) { + LogManager.addLog("ERROR", "video", "capture encoder close failed id=$id; input Surface retained: $e") + } + if (encoderResources == null) try { input?.release(); input = null } catch (e: Exception) { + LogManager.addLog("ERROR", "video", "capture input Surface close failed id=$id; retained: $e") + } + if (encoderResources == null && input == null) { + videoReleasing = false; partial.reset(); partialConfig.reset(); csd = byteArrayOf() + val done = releaseDone; releaseDone = null + LogManager.addLog("INFO", "video", "capture video cleanup confirmed id=$id generation=$generation") + if (!stopping) try { done?.invoke() } catch (e: Exception) { failed("capture continuation: $e") } + } + } + if (stopping && !videoReleasing && cameraClosed && audioCapture == null && audioThread == null && !orientationEnabled) { + stopped = true; retained.remove(this) + synchronized(this) { handler.removeCallbacks(cleanupTask); cleanupQueued = false } + thread.quitSafely() + LogManager.addLog("INFO", "stream", "capture cleanup confirmed id=$id frames=$frames dropped=$dropped bytes=$bytesSent") + return + } + val pending = videoReleasing || closingCameras.isNotEmpty() || closingSessions.isNotEmpty() || stopping + if (pending) { + val detail = "id=$id stopping=$stopping opening=$opening devices=${closingCameras.size} " + + "sessions=${ownedSessions.size}/${cameraSessions.size} codec=${encoderResources != null} input=${input != null} " + + "audio=${audioThread?.state} subscription=${audioCapture != null} sensor=$orientationEnabled" + if (videoReleasing && !slowReleaseReported && System.nanoTime() - videoReleaseSince >= 3_000_000_000) { + slowReleaseReported = true + LogManager.addLog("ERROR", "video", "capture close awaiting SDK acknowledgement; owners retained $detail") + } + LogManager.addLog("DEBUG", "stream", "capture cleanup pending $detail delayMs=$cleanupDelay") + queueCleanup(cleanupDelay); cleanupDelay = (cleanupDelay * 2).coerceAtMost(30_000) + } else { retained.remove(this); cleanupDelay = 250 } + } + + @Synchronized private fun queueCleanup(delay: Long): Boolean { + if (stopped) return true + val due = System.nanoTime() + delay * 1_000_000 + if (cleanupQueued && cleanupDue <= due) return true + handler.removeCallbacks(cleanupTask); cleanupDue = due + cleanupQueued = if (delay == 0L) handler.post(cleanupTask) else handler.postDelayed(cleanupTask, delay) + if (!cleanupQueued) LogManager.addLog("ERROR", "stream", "capture cleanup post rejected id=$id; owner retained") + return cleanupQueued } /** Меняем только свою копию PCM; общий микрофон и рация продолжают работать. */ fun setAudioEnabled(enabled: Boolean): Boolean { - if (!microphone || stopped) { + if (!microphone || stopping || stopped) { LogManager.addLog("WARN", "stream", "audio change without capture id=$id enabled=$enabled"); return false } val previous = audioMuted audioMuted = !enabled - if (!NativeLib.streamSetAudioEnabled(id, enabled)) { + val accepted = try { NativeLib.streamSetAudioEnabled(id, enabled) } catch (e: Exception) { + LogManager.addLog("ERROR", "stream", "audio change failed id=$id enabled=$enabled: $e"); false + } + if (!accepted) { audioMuted = previous LogManager.addLog("ERROR", "stream", "audio change refused id=$id enabled=$enabled"); return false } @@ -548,7 +611,8 @@ class StreamCaptureEngine( waitingStamp = false if (expected >= 0 && frame.sequence != expected) { LogManager.addLog("WARN", "stream", "audio queue discontinuity id=$id expected=$expected sequence=${frame.sequence}") - NativeLib.streamOpusClose(opus); opus = NativeLib.streamOpusCreate(true); check(opus != 0L) + NativeLib.streamOpusClose(opus); opus = 0 + opus = NativeLib.streamOpusCreate(true); check(opus != 0L) } expected = frame.sequence + 960 val pts = (frame.timeNs - originNs) / 1000 @@ -558,40 +622,44 @@ class StreamCaptureEngine( val packet = NativeLib.streamOpusEncode(opus, frame.pcm, audioCompressorSettings) ?: error("Stream compressor/Opus encode failed") check(NativeLib.streamFeed(id, false, pts, false, packet) >= 0) { "Stream audio publisher detached" } } - } catch (e: Exception) { if (audioRunning) handler.post { failed("audio capture: ${e.message}") } } - finally { if (opus != 0L) NativeLib.streamOpusClose(opus) } - }, "stream-audio-capture").apply { start() } + } catch (e: Exception) { + LogManager.addLog(if (audioRunning) "ERROR" else "WARN", "stream", "capture audio worker ended id=$id running=$audioRunning: $e") + if (audioRunning && !handler.post { failed("audio capture: ${e.message}") }) + LogManager.addLog("ERROR", "stream", "capture audio failure notification rejected id=$id") + } finally { + var delay = 250L + if (Thread.interrupted()) LogManager.addLog("WARN", "stream", "capture audio interrupted before Opus release id=$id") + while (opus != 0L) try { NativeLib.streamOpusClose(opus); opus = 0 } catch (e: Exception) { + LogManager.addLog("ERROR", "stream", "capture Opus close failed id=$id handle=$opus; retained: $e") + try { Thread.sleep(delay) } catch (interrupted: InterruptedException) { + LogManager.addLog("WARN", "stream", "capture Opus cleanup wait interrupted id=$id; retrying") + } + delay = (delay * 2).coerceAtMost(30_000) + } + } + }, "stream-audio-capture") + audioThread!!.start() } private fun failed(detail: String) { if (stopping || stopped || terminalFailure) return terminalFailure = true - observeOrientation(false) LogManager.addLog("ERROR", "stream", "capture failed id=$id $detail") - failure(detail) + stopping = true; audioRunning = false; retained.add(this); queueCleanup(0) + try { failure(detail) } catch (e: Exception) { LogManager.addLog("ERROR", "stream", "capture failure callback failed id=$id: $e") } } /** Producers закрываются до native FIN; уже принятые кадры дренирует native publisher. */ - fun stop(done: () -> Unit) { - if (stopped) { done(); return } - if (!handler.post({ - if (stopped) { done(); return@post } - stopCallbacks.add(done) - if (stopping) return@post - stopping = true; audioRunning = false - observeOrientation(false) - preview.close(); previewTarget?.release(); previewTarget = null - audioCapture?.close(); audioCapture = null - audioThread?.join(); audioThread = null - releaseVideo { - stopped = true - LogManager.addLog("INFO", "stream", "capture released id=$id frames=$frames dropped=$dropped bytes=$bytesSent") - stopCallbacks.toList().also { stopCallbacks.clear() }.forEach { it() } - maybeQuit() - } - })) { - LogManager.addLog("WARN", "stream", "capture handler already closed id=$id; complete stop") - done() + fun stop(): Boolean { + check(Looper.myLooper() == Looper.getMainLooper()) { "Capture stop requires main" } + if (stopped) return true + stopping = true; audioRunning = false; retained.add(this) + if (!::handler.isInitialized) { + preview.close(); stopped = true; retained.remove(this); thread.quitSafely() + LogManager.addLog("INFO", "stream", "passive capture released before worker startup id=$id") + return true } + check(queueCleanup(0)) { "Capture cleanup worker rejected stop id=$id" } + return stopped } }