From 2abdb229fdf3292c2fe7c9b87273866dcb258252 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 29 Oct 2026 01:52:03 +0100 Subject: [PATCH] Keep publisher service owners until capture and foreground cleanup are acknowledged --- .../com/utun/chat/data/StreamCaptureEngine.kt | 7 +- .../java/com/utun/chat/data/StreamService.kt | 137 +++++++++++++----- 2 files changed, 103 insertions(+), 41 deletions(-) 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 d4387528..3166148a 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 @@ -73,6 +73,7 @@ class StreamCaptureEngine( private var videoReleasing = false private var videoReleaseSince = 0L private var slowReleaseReported = false + private var stopVideoRequested = false private var generation = 0L private var videoRevision = 0L private var videoRotation = 0 @@ -502,7 +503,7 @@ class StreamCaptureEngine( /** Только capture worker: независимый cleanup, refs снимаются после возврата SDK либо callback закрытия. */ private fun cleanupResources() { if (stopping) { - if (!videoReleasing) releaseVideo {} + if (!stopVideoRequested) { stopVideoRequested = true; releaseVideo {} } releaseDone = null; audioRunning = false try { observeOrientation(false) } catch (e: Exception) { LogManager.addLog("ERROR", "video", "capture orientation release failed id=$id; retained: $e") @@ -542,7 +543,9 @@ class StreamCaptureEngine( if (!stopping) try { done?.invoke() } catch (e: Exception) { failed("capture continuation: $e") } } } - if (stopping && !videoReleasing && cameraClosed && audioCapture == null && audioThread == null && !orientationEnabled) { + if (stopping && !videoReleasing && !opening && camera == null && closingCameras.isEmpty() && + ownedSessions.isEmpty() && cameraSessions.isEmpty() && encoderResources == null && input == null && + audioCapture == null && audioThread == null && !orientationEnabled) { stopped = true; retained.remove(this) synchronized(this) { handler.removeCallbacks(cleanupTask); cleanupQueued = false } thread.quitSafely() diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamService.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamService.kt index 76f33d84..2c9e3c4f 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamService.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamService.kt @@ -11,6 +11,8 @@ import android.os.IBinder import android.os.Handler import android.os.Looper import android.os.PowerManager +import android.os.SystemClock +import android.net.Uri import androidx.core.app.NotificationCompat import androidx.core.app.ServiceCompat import com.utun.chat.MainActivity @@ -27,9 +29,14 @@ import kotlinx.coroutines.runBlocking import kotlinx.coroutines.CancellationException import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.collect +import java.util.concurrent.ConcurrentHashMap -/** Foreground-владелец камеры, микрофона и CPU lock; не восстанавливает live session после смерти процесса. */ +/** Main-владелец capture/native publisher/CPU lock/notification. Сохраняет refs до ACK, также после onDestroy. + * Handler повторяет независимые ошибки; FIN следует после capture.stop=true, доставка ждёт до 4 секунд. + * Live session не восстанавливается после смерти процесса. + */ class StreamService : Service() { + companion object { private val retained = ConcurrentHashMap.newKeySet() } private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate) private var capture: StreamCaptureEngine? = null private var lock: PowerManager.WakeLock? = null @@ -37,27 +44,50 @@ class StreamService : Service() { private var stopping = false private var aborting = false private var poll: Job? = null + private val main = Handler(Looper.getMainLooper()) + private val retry = Runnable { cleanup() } + private var retryDelay = 250L + private var nativeStopped = false + private var drainUntil = 0L + private var destroyed = false + private var foregroundOwned = false + private var stopRequested = false + private var controllerEnded = false + private var lastStartId = 0 + private var openIntent: PendingIntent? = null + private var stopIntent: PendingIntent? = null override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int { - if (intent?.action == "stop") { StreamController.stop(); return START_NOT_STICKY } - if (id != 0L) return START_NOT_STICKY + lastStartId = startId + if (intent?.action == "stop") { + val requested = intent.getLongExtra("id", 0) + if (requested != 0L && requested == StreamController.state.value?.id) StreamController.stop() + else LogManager.addLog("WARN", "stream", "stale notification stop ignored requested=$requested current=${StreamController.state.value?.id}") + if (id == 0L) finish() + return START_NOT_STICKY + } + if (id != 0L || stopping) { + LogManager.addLog("WARN", "stream", "duplicate service start ignored id=$id stopping=$stopping") + return START_NOT_STICKY + } id = intent?.getLongExtra("id", 0) ?: 0 val audio = intent?.getBooleanExtra("audio", true) ?: true try { val notifications = getSystemService(NotificationManager::class.java) notifications.createNotificationChannel(NotificationChannel("live_stream", getString(R.string.stream_title), NotificationManager.IMPORTANCE_LOW)) - val open = PendingIntent.getActivity(this, 0, Intent(this, MainActivity::class.java), PendingIntent.FLAG_IMMUTABLE) - val stop = PendingIntent.getService(this, 0, Intent(this, StreamService::class.java).setAction("stop"), PendingIntent.FLAG_IMMUTABLE) + val open = PendingIntent.getActivity(this, 0, Intent(this, MainActivity::class.java).setData(Uri.parse("utun-stream:open:$id")), + PendingIntent.FLAG_IMMUTABLE).also { openIntent = it } + val stop = PendingIntent.getService(this, 0, Intent(this, StreamService::class.java).setAction("stop") + .setData(Uri.parse("utun-stream:stop:$id")).putExtra("id", id), PendingIntent.FLAG_IMMUTABLE).also { stopIntent = it } val notification = NotificationCompat.Builder(this, "live_stream").setSmallIcon(android.R.drawable.ic_menu_camera) .setContentTitle(getString(R.string.stream_live)).setContentIntent(open).setOngoing(true) .addAction(android.R.drawable.ic_media_pause, getString(R.string.stream_stop), stop).build() val type = if (Build.VERSION.SDK_INT >= 30) ServiceInfo.FOREGROUND_SERVICE_TYPE_CAMERA or (if (audio) ServiceInfo.FOREGROUND_SERVICE_TYPE_MICROPHONE else 0) else 0 - ServiceCompat.startForeground(this, 1004, notification, type) + foregroundOwned = true; ServiceCompat.startForeground(this, 1004, notification, type) if (id == 0L || !StreamController.serviceReady(id, this)) { finish(); return START_NOT_STICKY } - lock = getSystemService(PowerManager::class.java).newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "utun:live_stream").apply { - setReferenceCounted(false); acquire() - } + lock = getSystemService(PowerManager::class.java).newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "utun:live_stream") + lock!!.setReferenceCounted(false); lock!!.acquire() val group = intent!!.getLongExtra("group", 0) val publisher = StreamController.state.value ?: error("Missing publisher") val config = publisher.pendingConfig ?: error("Missing capture configuration") @@ -65,7 +95,8 @@ class StreamService : Service() { val audioSettings = if (audio) runBlocking { provider.radioCompressorSettings.first() } else RadioCompressorSettings() capture = StreamCaptureEngine(this, id, group, config, audio, publisher.preview, audioSettings, { active, restored -> StreamController.captureReady(id, active, restored) }, - { detail -> StreamController.failed(id, detail) }).also { it.start() } + { detail -> StreamController.failed(id, detail) }) + capture!!.start() if (audio) scope.launch { try { provider.radioCompressorSettings.collect { capture?.audioCompressorSettings = it } } catch (e: CancellationException) { throw e } @@ -74,6 +105,7 @@ class StreamService : Service() { } } poll = scope.launch { + try { val deadline = System.nanoTime() + 10_000_000_000 var opened = false var nextStats = 0L @@ -94,6 +126,11 @@ class StreamService : Service() { } delay(20) } + } catch (e: CancellationException) { throw e } + catch (e: Exception) { + LogManager.addLog("ERROR", "stream", "publisher polling failed id=$id: $e") + StreamController.failed(id, "publisher polling: $e"); shutdown(abort = true) + } } LogManager.addLog("INFO", "stream", "foreground started id=$id group=$group microphone=$audio CPU lock=${lock?.isHeld}") } catch (e: Exception) { @@ -117,42 +154,64 @@ class StreamService : Service() { /** Producers завершатся до FIN; при уничтожении ядра очередь не ждём. */ fun shutdown(abort: Boolean = false) { + check(Looper.myLooper() == Looper.getMainLooper()) { "Stream service shutdown requires main" } aborting = aborting || abort - if (stopping) return - stopping = true - val engine = capture - if (engine == null) { NativeLib.streamStop(id); finish(); return } - engine.stop { - scope.launch { - capture = null; NativeLib.streamStop(id) - if (aborting) finish() - else scope.launch { delay(4000); if (stopping) finish() } - } - } + stopping = true; retained.add(this); main.removeCallbacks(retry); cleanup() } private fun finish() { - if (capture != null) { shutdown(abort = true); return } - poll?.cancel(); poll = null - lock?.let { if (it.isHeld) it.release() }; lock = null - StreamController.ended(id, this) - stopForeground(STOP_FOREGROUND_REMOVE); stopSelf() + shutdown(abort = true) + } + + /** Main остаётся доступен для SDK callbacks; false от engine сохраняет владельца до следующей попытки. */ + private fun cleanup() { + main.removeCallbacks(retry) + capture?.let { owned -> try { + if (owned.stop()) capture = null + else LogManager.addLog("DEBUG", "stream", "publisher capture cleanup pending id=$id") + } catch (e: Exception) { LogManager.addLog("ERROR", "stream", "publisher capture stop failed id=$id; retained: $e") } } + if (capture == null && !nativeStopped) try { + if (id != 0L) NativeLib.streamStop(id) + nativeStopped = true; drainUntil = SystemClock.elapsedRealtime() + 4000 + LogManager.addLog("INFO", "stream", "publisher producers released id=$id; FIN requested abort=$aborting") + } catch (e: Exception) { LogManager.addLog("ERROR", "stream", "publisher native stop failed id=$id; retained: $e") } + val draining = !aborting && !destroyed && nativeStopped && SystemClock.elapsedRealtime() < drainUntil + if (capture == null && nativeStopped && !draining) { + poll?.cancel(); poll = null + lock?.let { owned -> try { if (owned.isHeld) owned.release(); lock = null } catch (e: Exception) { + LogManager.addLog("ERROR", "stream", "publisher CPU lock release failed id=$id; retained: $e") + } } + openIntent?.let { owned -> try { owned.cancel(); openIntent = null } catch (e: Exception) { + LogManager.addLog("ERROR", "stream", "publisher open token cancellation failed id=$id; retained: $e") + } } + stopIntent?.let { owned -> try { owned.cancel(); stopIntent = null } catch (e: Exception) { + LogManager.addLog("ERROR", "stream", "publisher stop token cancellation failed id=$id; retained: $e") + } } + if (foregroundOwned) try { stopForeground(STOP_FOREGROUND_REMOVE); foregroundOwned = false } catch (e: Exception) { + LogManager.addLog("ERROR", "stream", "publisher foreground removal failed id=$id; retained: $e") + } + if (!stopRequested) try { if (!destroyed) stopSelf(lastStartId); stopRequested = true } catch (e: Exception) { + LogManager.addLog("ERROR", "stream", "publisher service stop failed id=$id; retained: $e") + } + if (lock == null && openIntent == null && stopIntent == null && !foregroundOwned && stopRequested && !controllerEnded) try { + StreamController.ended(id, this); controllerEnded = true + } catch (e: Exception) { LogManager.addLog("ERROR", "stream", "publisher completion failed id=$id; retained: $e") } + } + if (controllerEnded) { + retained.remove(this); retryDelay = 250 + LogManager.addLog("INFO", "stream", "publisher service cleanup confirmed id=$id") + } else { + val delay = if (draining) minOf(250, drainUntil - SystemClock.elapsedRealtime()).coerceAtLeast(1) else retryDelay + if (!draining) retryDelay = (retryDelay * 2).coerceAtMost(30_000) + if (!main.postDelayed(retry, delay)) LogManager.addLog("ERROR", "stream", "publisher cleanup retry rejected id=$id; owner retained") + else LogManager.addLog("DEBUG", "stream", "publisher cleanup scheduled id=$id delayMs=$delay capture=${capture != null} drain=$draining") + } } override fun onDestroy() { - poll?.cancel() - val engine = capture; capture = null - val ownedLock = lock; lock = null - val release = { - NativeLib.streamStop(id) - ownedLock?.let { if (it.isHeld) it.release() } - Handler(Looper.getMainLooper()).post { StreamController.ended(id, this) } - Unit - } - if (engine != null) engine.stop(release) else release() - scope.cancel() - LogManager.addLog("INFO", "stream", "foreground destroyed id=$id") - super.onDestroy() + destroyed = true; scope.cancel() + try { shutdown(abort = true); LogManager.addLog("INFO", "stream", "foreground destroyed id=$id pending=${!controllerEnded}") } + finally { super.onDestroy() } } override fun onBind(intent: Intent?): IBinder? = null }