Play live streams with bounded queues and audio clock synchronization
This commit is contained in:
evgeny committed
1 parent
f2a5dde761
commit
ee851b4aa6
1 file changed
+226
@@ -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<Float> = mutableAspect
|
||||
private val mutableStatus = MutableStateFlow(R.string.stream_connecting)
|
||||
val status: StateFlow<Int> = 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<Frame>(); val audios = ArrayDeque<Frame>()
|
||||
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")
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user