Guard publisher handshake, error callbacks and native teardown

This commit is contained in:
evgeny committed 2026-10-29 13:34:24 +01:00
1 parent 2abdb229fd
commit e49ccfeb64
3 files changed
+44 -15

No files matched your search

@@ -75,6 +75,7 @@ object NativeLib {
StreamController.reset(); CallController.reset(); ChatApplication.instance.avatars.stop(); ChatApplication.instance.memberFiles.cancel()
CallService.requireReleased() // Не останавливать native при неподтверждённой очистке его media-владельца.
StreamPlayerEngine.stopForCore()
StreamController.requireReleased()
check(!CallController.blocksMedia()) {
LogManager.addLog("ERROR", "Call", "Core teardown blocked: call media preparation still running")
"Call media preparation still running"
@@ -35,7 +35,8 @@ object StreamController {
val txStats: StateFlow<StreamTxStats?> = mutableTx
fun updateTx(id: Long, data: LongArray) {
check(Looper.myLooper() == Looper.getMainLooper())
if (mutable.value?.id == id && mutable.value?.ongoing == true) mutableTx.value = StreamTxStats.fromNative(id, data)
if (mutable.value?.id == id && mutable.value?.ongoing == true) try { mutableTx.value = StreamTxStats.fromNative(id, data) }
catch (e: Exception) { LogManager.addLog("ERROR", "stream", "invalid publisher TX snapshot id=$id fields=${data.size}; previous retained: $e") }
}
@Volatile var capturing = false
private set
@@ -55,7 +56,7 @@ object StreamController {
check(Looper.myLooper() == Looper.getMainLooper())
if (mutable.value?.ongoing == true) { LogManager.addLog("WARN", "stream", "publisher already owned"); return }
val id = nextId()
if (!NativeLib.isReady() || group == 0L || CallController.state.value?.ongoing == true ||
if (!NativeLib.isReady() || group == 0L || CallController.blocksMedia() || CallController.state.value?.ongoing == true ||
NativeLib.voiceIsActive()) {
LogManager.addLog("WARN", "stream", "start rejected: core/call/voice busy group=$group")
mutable.value = StreamState(id, group, StreamPhase.FAILED, R.string.stream_resources_busy); return
@@ -65,8 +66,8 @@ object StreamController {
mutable.value = StreamState(id, group, StreamPhase.STARTING, pendingConfig = config, audioAvailable = audio, audioEnabled = audio)
mutableTx.value = null
try {
ContextCompat.startForegroundService(ChatApplication.instance, Intent(ChatApplication.instance, StreamService::class.java)
.putExtra("id", id).putExtra("group", group).putExtra("audio", audio))
checkNotNull(ContextCompat.startForegroundService(ChatApplication.instance, Intent(ChatApplication.instance, StreamService::class.java)
.putExtra("id", id).putExtra("group", group).putExtra("audio", audio))) { "Publisher foreground service not found" }
} catch (e: Exception) {
LogManager.addLog("ERROR", "stream", "service start failed id=$id: ${e.message}")
capturing = false; mutable.value = StreamState(id, group, StreamPhase.FAILED, R.string.stream_failed)
@@ -82,7 +83,7 @@ object StreamController {
}
if (st.config == config) return
mutable.value = st.copy(pendingConfig = config, notice = null)
owner.change(config)
try { owner.change(config) } catch (e: Exception) { failed(st.id, "capture change rejected: $e") }
}
fun setAudioEnabled(enabled: Boolean) {
@@ -92,25 +93,32 @@ object StreamController {
LogManager.addLog("WARN", "stream", "audio control unavailable id=${st.id} phase=${st.phase}"); return
}
if (st.audioEnabled == enabled) return
if (service!!.setAudioEnabled(enabled)) mutable.value = st.copy(audioEnabled = enabled)
else failed(st.id, "audio control failed enabled=$enabled")
try {
if (service!!.setAudioEnabled(enabled)) mutable.value = st.copy(audioEnabled = enabled)
else failed(st.id, "audio control failed enabled=$enabled")
} catch (e: Exception) { failed(st.id, "audio control failed enabled=$enabled: $e") }
}
fun captureReady(id: Long, config: StreamCaptureConfig, restored: Boolean) {
main.post {
if (!main.post {
val st = mutable.value ?: return@post
if (st.id != id || !st.ongoing || st.phase == StreamPhase.STOPPING) return@post
mutable.value = st.copy(config = config, pendingConfig = null, notice = if (restored) R.string.stream_mode_restored else null)
rememberSettings(config)
}
}) LogManager.addLog("ERROR", "stream", "capture ready post rejected id=$id; publisher owner retained")
}
fun serviceReady(id: Long, owner: StreamService): Boolean {
if (mutable.value?.id != id || mutable.value?.phase != StreamPhase.STARTING) return false
check(Looper.myLooper() == Looper.getMainLooper()) { "Publisher service handshake requires main" }
if (mutable.value?.id != id || mutable.value?.phase != StreamPhase.STARTING || (service != null && service !== owner)) {
LogManager.addLog("WARN", "stream", "stale publisher service rejected id=$id current=${mutable.value?.id} phase=${mutable.value?.phase}")
return false
}
service = owner; return true
}
fun opened(id: Long) {
check(Looper.myLooper() == Looper.getMainLooper()) { "Publisher open callback requires main" }
val st = mutable.value ?: return
if (st.id == id && st.phase == StreamPhase.STARTING) mutable.value = st.copy(phase = StreamPhase.LIVE)
}
@@ -122,13 +130,15 @@ object StreamController {
LogManager.addLog("ERROR", "stream", "publisher failed id=$id $detail")
mutable.value = st.copy(error = R.string.stream_failed); stop()
}
if (Looper.myLooper() == Looper.getMainLooper()) update.run() else main.post(update)
if (Looper.myLooper() == Looper.getMainLooper()) update.run()
else if (!main.post(update)) LogManager.addLog("ERROR", "stream", "publisher failure post rejected id=$id detail=$detail; owner retained")
}
fun stop() {
check(Looper.myLooper() == Looper.getMainLooper()) { "Publisher stop requires main" }
val st = mutable.value ?: return
if (!st.ongoing || st.phase == StreamPhase.STOPPING) return
mutable.value = st.copy(phase = StreamPhase.STOPPING)
if (!st.ongoing) return
if (st.phase != StreamPhase.STOPPING) mutable.value = st.copy(phase = StreamPhase.STOPPING)
service?.shutdown() // До onStartCommand отменённый start завершит foreground-handshake и остановится.
}
@@ -143,9 +153,12 @@ object StreamController {
}
fun ended(id: Long, owner: StreamService) {
if (service === owner) service = null
check(Looper.myLooper() == Looper.getMainLooper()) { "Publisher completion requires main" }
if (service != null && service !== owner) { LogManager.addLog("WARN", "stream", "foreign publisher completion ignored id=$id"); return }
val st = mutable.value ?: return
if (st.id != id || !st.ongoing) return
if (service === owner) service = null
st.preview.close()
capturing = false
mutableTx.value = null
mutable.value = st.copy(phase = if (st.error != null) StreamPhase.FAILED else StreamPhase.ENDED, pendingConfig = null)
@@ -153,6 +166,7 @@ object StreamController {
}
fun reset() {
check(Looper.myLooper() == Looper.getMainLooper()) { "Publisher reset requires main" }
val st = mutable.value
if (st?.ongoing == true) {
mutable.value = st.copy(phase = StreamPhase.STOPPING, error = R.string.stream_failed)
@@ -160,11 +174,20 @@ object StreamController {
}
}
/** Незавершённый foreground handshake и SDK cleanup запрещают освобождать native publisher. */
fun requireReleased() {
check(!capturing && service == null) {
LogManager.addLog("ERROR", "stream", "core teardown blocked by publisher id=${mutable.value?.id} phase=${mutable.value?.phase}")
"Stream publisher cleanup pending"
}
}
/** Звонок/голосовое не отбирают микрофон у foreground-трансляции; рация использует общий захват. */
fun blocksCapture(): Boolean {
if (!capturing) return false
LogManager.addLog("WARN", "stream", "microphone/camera owned by stream id=${mutable.value?.id}")
main.post { Toast.makeText(AppLanguage.context(), R.string.stream_resources_busy, Toast.LENGTH_SHORT).show() }
if (!main.post { Toast.makeText(AppLanguage.context(), R.string.stream_resources_busy, Toast.LENGTH_SHORT).show() })
LogManager.addLog("ERROR", "stream", "capture busy notice post rejected")
return true
}
}
@@ -59,6 +59,11 @@ class StreamService : Service() {
override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
lastStartId = startId
if (controllerEnded && !destroyed) {
if (intent?.action == "stop") { stopSelf(startId); return START_NOT_STICKY }
id = 0; stopping = false; aborting = false; nativeStopped = false; drainUntil = 0
stopRequested = false; controllerEnded = false
}
if (intent?.action == "stop") {
val requested = intent.getLongExtra("id", 0)
if (requested != 0L && requested == StreamController.state.value?.id) StreamController.stop()