Acknowledge stream capture shutdown only after every owner is released
This commit is contained in:
evgeny committed
1 parent
5e9f7e01e0
commit
68d21fbffa
1 file changed
+119
-51
+119
-51
@@ -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
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user