diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt new file mode 100644 index 00000000..834e1cea --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt @@ -0,0 +1,226 @@ +package com.utun.chat.data + +import android.media.AudioAttributes +import android.media.AudioFormat +import android.media.AudioTimestamp +import android.media.AudioTrack +import android.media.MediaCodec +import android.media.MediaFormat +import android.view.Surface +import com.utun.chat.R +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import java.nio.ByteBuffer +import java.nio.ByteOrder +import java.util.ArrayDeque +import java.util.concurrent.locks.ReentrantLock +import kotlin.concurrent.withLock +import kotlin.math.roundToInt + +/** Один worker владеет декодерами и AudioTrack. Hardware audio clock управляет временем показа видео. + * Просроченные P-кадры декодируются без показа, чтобы сохранить references; при переполнении ждём KEY. + * viewerLock сериализует stop/open разных экранов до подтверждения освобождения native owner. */ +class StreamPlayerEngine(private val group: Long, private val source: Long) { + companion object { private val viewerLock = ReentrantLock() } + private data class Frame(val pts: Long, val key: Boolean, val width: Int, val height: Int, val rotation: Int, val data: ByteArray) + private val mutableAspect = MutableStateFlow(16f / 9f) + val aspect: StateFlow = mutableAspect + private val mutableStatus = MutableStateFlow(R.string.stream_connecting) + val status: StateFlow = mutableStatus + private val id = StreamController.nextId() + @Volatile private var running = false + private var worker: Thread? = null + + fun start(surface: Surface) { + if (worker != null) return + running = true + worker = Thread({ viewerLock.withLock { if (running) play(surface) } }, "stream-player").apply { start() } + } + + fun stop() { running = false; NativeLib.streamStop(id) } + + private fun play(surface: Surface) { + var decoder: MediaCodec? = null + var track: AudioTrack? = null + var opus = 0L + val videos = ArrayDeque(); val audios = ArrayDeque() + var videoBytes = 0 + var waitingKey = true + var audioBase = -1L + var written = 0L + var videoBase = -1L + var videoStartNs = 0L + var ended = false + val audioTimestamp = AudioTimestamp() + val bufferInfo = MediaCodec.BufferInfo() + var report = 0L + var decoded = 0L + var discarded = 0L + + fun resetVideo() { + try { decoder?.stop(); decoder?.release() } + catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder reset id=$id: ${e.message}") } + decoder = null; videos.clear(); videoBytes = 0; waitingKey = true + mutableStatus.value = R.string.stream_waiting_key + } + fun resetAudio() { + track?.pause(); track?.flush(); track?.release(); track = null + if (opus != 0L) NativeLib.streamOpusClose(opus) + opus = 0; audioBase = -1; written = 0; audios.clear() + } + fun currentPts(now: Long): Long { + val t = track + if (t != null && audioBase >= 0) { + val played = if (t.getTimestamp(audioTimestamp)) { + (audioTimestamp.framePosition + (now - audioTimestamp.nanoTime).coerceAtLeast(0) * 48000 / 1_000_000_000) + .coerceAtMost(written) + } else t.playbackHeadPosition.toLong() and 0xffffffffL + return audioBase + played * 1_000_000 / 48000 + } + return if (videoBase >= 0) videoBase + (now - videoStartNs) / 1000 else 0 + } + try { + check(NativeLib.streamOpen(id, group, source)) { "Viewer open refused" } + val openDeadline = System.nanoTime() + 10_000_000_000 + var opened = false + var prefillUntil = 0L + while (running && surface.isValid) { + if (!NativeLib.isReady()) error("Core stopped") + repeat(16) { + val raw = if (!ended) NativeLib.streamPull(id) else null + if (raw != null) { + check(raw.size >= 20) + val b = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN) + val kind = b.get().toInt(); val key = b.get().toInt() and 1 != 0 + val width = b.short.toInt() and 65535; val height = b.short.toInt() and 65535 + val rotation = b.short.toInt() and 65535; val pts = b.long; val size = b.int + check(size == b.remaining()) + when (kind) { + 1, 2 -> { + val frame = Frame(pts, key, width, height, rotation, ByteArray(size).also { b.get(it) }) + if (prefillUntil == 0L) prefillUntil = System.nanoTime() + 150_000_000 + if (kind == 1) { + if (videos.size >= 12 || videoBytes + size > H264AnnexB.MAX_FRAME) resetVideo() + if (!waitingKey || key) { + if (waitingKey) { + check(H264AnnexB.nals(frame.data).map { it.first }.toSet().containsAll(setOf(5, 7, 8))) + waitingKey = false + mutableAspect.value = if (rotation % 180 == 0) width.toFloat() / height else height.toFloat() / width + } + videos.addLast(frame); videoBytes += size + } + } else { + if (audios.size >= 16) { + LogManager.addLog("WARN", "stream", "audio queue overflow id=$id; reset timeline") + resetAudio(); resetVideo() + } + audios.addLast(frame) + } + } + 3 -> { LogManager.addLog("INFO", "stream", "viewer GAP id=$id"); resetVideo(); resetAudio(); videoBase = -1; prefillUntil = 0 } + 4 -> opened = true + 5 -> ended = true + 6 -> error("Native viewer closed") + else -> error("Unknown media event=$kind") + } + } + } + if (!opened && System.nanoTime() > openDeadline) error("Viewer open timeout") + val now = System.nanoTime() + if (prefillUntil != 0L && now >= prefillUntil) { + if (audios.isNotEmpty() && track == null) { + opus = NativeLib.streamOpusCreate(false); check(opus != 0L) + track = AudioTrack.Builder().setAudioAttributes(AudioAttributes.Builder().setUsage(AudioAttributes.USAGE_MEDIA) + .setContentType(AudioAttributes.CONTENT_TYPE_MOVIE).build()).setAudioFormat(AudioFormat.Builder() + .setSampleRate(48000).setChannelMask(AudioFormat.CHANNEL_OUT_MONO).setEncoding(AudioFormat.ENCODING_PCM_16BIT).build()) + .setBufferSizeInBytes(AudioTrack.getMinBufferSize(48000, AudioFormat.CHANNEL_OUT_MONO, AudioFormat.ENCODING_PCM_16BIT) + .coerceAtLeast(9600)).setTransferMode(AudioTrack.MODE_STREAM).build() + check(track!!.state == AudioTrack.STATE_INITIALIZED) + audioBase = audios.first.pts + track!!.play() + LogManager.addLog("INFO", "stream", "audio playback starts id=$id PTS=$audioBase") + } + val t = track + if (t != null && audios.isNotEmpty()) { + val head = t.playbackHeadPosition.toLong() and 0xffffffffL + if (written - head < 3840) { + val f = audios.removeFirst() + val missing = ((f.pts - audioBase - written * 1_000_000 / 48000).toDouble() / 20000).roundToInt() + if (missing > 5 || missing < -5) { + LogManager.addLog("WARN", "stream", "audio discontinuity id=$id missing=$missing packets") + resetAudio(); audios.addFirst(f); resetVideo() + } else if (missing >= 0) { + repeat(missing) { + val conceal = NativeLib.streamOpusDecode(opus, byteArrayOf()) ?: error("Opus PLC failed") + check(t.write(conceal, 0, conceal.size, AudioTrack.WRITE_BLOCKING) == conceal.size); written += conceal.size + } + val pcm = NativeLib.streamOpusDecode(opus, f.data) ?: error("Opus decode failed") + check(t.write(pcm, 0, pcm.size, AudioTrack.WRITE_BLOCKING) == pcm.size); written += pcm.size + } else LogManager.addLog("DEBUG", "stream", "late audio discarded id=$id pts=${f.pts}") + } + } + if (videoBase < 0 && videos.isNotEmpty()) { videoBase = videos.first.pts; videoStartNs = now } + val clock = currentPts(now) + if (videos.isNotEmpty() && videos.first.pts <= clock + 80000) { + val frame = videos.first + if (decoder == null) { + check(frame.key) + val nals = H264AnnexB.nals(frame.data) + val format = MediaFormat.createVideoFormat(MediaFormat.MIMETYPE_VIDEO_AVC, frame.width, frame.height).apply { + setByteBuffer("csd-0", ByteBuffer.wrap(nals.first { it.first == 7 }.second)) + setByteBuffer("csd-1", ByteBuffer.wrap(nals.first { it.first == 8 }.second)) + setInteger(MediaFormat.KEY_ROTATION, frame.rotation); setInteger(MediaFormat.KEY_MAX_INPUT_SIZE, H264AnnexB.MAX_FRAME) + } + decoder = MediaCodec.createDecoderByType(MediaFormat.MIMETYPE_VIDEO_AVC).also { it.configure(format, surface, null, 0); it.start() } + LogManager.addLog("INFO", "video", "decoder started id=$id ${frame.width}x${frame.height} rotation=${frame.rotation}") + } + val index = decoder!!.dequeueInputBuffer(0) + if (index >= 0) { + val input = decoder!!.getInputBuffer(index) ?: error("Missing decoder input") + check(frame.data.size <= input.capacity()) { "Decoder input too small" } + input.clear(); input.put(frame.data) + decoder!!.queueInputBuffer(index, 0, frame.data.size, frame.pts, 0) + videos.removeFirst(); videoBytes -= frame.data.size; decoded++ + } else if (clock - frame.pts > 250000) { + LogManager.addLog("WARN", "video", "decoder backpressure id=$id ageUs=${clock - frame.pts}; wait KEY") + resetVideo() + } + } + val codec = decoder + if (codec != null) repeat(8) { + val index = codec.dequeueOutputBuffer(bufferInfo, 0) + if (index >= 0) { + val clockPts = currentPts(System.nanoTime()) + if (bufferInfo.presentationTimeUs < clockPts - 100000) { + codec.releaseOutputBuffer(index, false); discarded++ + } else codec.releaseOutputBuffer(index, System.nanoTime() + (bufferInfo.presentationTimeUs - clockPts).coerceAtLeast(0) * 1000) + mutableStatus.value = R.string.stream_live + } + } + } + if (ended && videos.isEmpty() && audios.isEmpty()) { mutableStatus.value = R.string.stream_ended; break } + if (now - report >= 1_000_000_000) { + report = now + LogManager.addLog("DEBUG", "stream", "viewer id=$id decoded=$decoded discarded=$discarded videoQ=${videos.size}/$videoBytes " + + "audioQ=${audios.size} written=$written clock=${currentPts(now)} waitingKey=$waitingKey") + } + Thread.sleep(5) + } + } catch (e: Exception) { + if (running) { mutableStatus.value = R.string.stream_failed; LogManager.addLog("ERROR", "stream", "viewer failed id=$id: ${e.message}") } + } finally { + NativeLib.streamStop(id) + try { decoder?.stop(); decoder?.release() } catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") } + try { track?.stop(); track?.release() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") } + if (opus != 0L) NativeLib.streamOpusClose(opus) + val deadline = System.nanoTime() + 500_000_000 + while (NativeLib.isReady() && System.nanoTime() < deadline) { + val event = NativeLib.streamPull(id) + if (event != null && event[0].toInt() in 5..6) break + Thread.sleep(5) + } + running = false + LogManager.addLog("INFO", "stream", "viewer released id=$id decoded=$decoded discarded=$discarded") + } + } +}