jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Use adaptive ring backed playback and stereo tempo on Android

master
evgeny 6 days ago
parent
e6e276aed3
commit
acf374df4e
2 changed files with 227 additions and 262 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 25
      tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt
  2. 464
      tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt

25
tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt
Unescape View File

@ -207,7 +207,18 @@ object NativeLib {
width: Int = 0, height: Int = 0, rotation: Int = 0, audio: Boolean = true): Int =
nativeStreamFeed(id, video, pts, key, data, revision, width, height, rotation, audio)
fun streamStop(id: Long) { nativeStreamStop(id) }
fun streamPull(id: Long): ByteArray? = nativeStreamPull(id)
fun streamPull(id: Long, positionUs: Long = -1, videoAheadUs: Long = 80000, audioAheadUs: Long = 80000): ByteArray? =
nativeStreamPull(id, positionUs, videoAheadUs, audioAheadUs)
fun streamStatus(id: Long): LongArray? = nativeStreamStatus(id)
fun streamResync(id: Long): Boolean = nativeStreamResync(id)
fun streamPlaybackCreate(id: Long, pts: Long, audio: Boolean): Long = nativeStreamPlaybackCreate(id, pts, audio)
fun streamPlaybackFeed(handle: Long, pts: Long, packet: ByteArray): Int = nativeStreamPlaybackFeed(handle, pts, packet)
fun streamPlaybackPull(handle: Long, frames: Int, limit: Long): ShortArray? = nativeStreamPlaybackPull(handle, frames, limit)
fun streamPlaybackUpdate(handle: Long, now: Long, played: Long, ready: Long, guard: Long, starved: Boolean): LongArray? =
nativeStreamPlaybackUpdate(handle, now, played, ready, guard, starved)
fun streamPlaybackFinish(handle: Long) { nativeStreamPlaybackFinish(handle) }
fun streamPlaybackEmpty(handle: Long): Boolean = nativeStreamPlaybackEmpty(handle)
fun streamPlaybackClose(handle: Long) { nativeStreamPlaybackClose(handle) }
fun streamOpusCreate(encode: Boolean): Long = nativeStreamOpusCreate(encode)
fun streamOpusEncode(handle: Long, pcm: ShortArray, compressor: RadioCompressorSettings): ByteArray? =
nativeStreamOpusEncode(handle, pcm, compressor.enabled, compressor.maxGainDb, compressor.riseRate)
@ -219,7 +230,17 @@ object NativeLib {
private external fun nativeStreamFeed(id: Long, video: Boolean, pts: Long, key: Boolean, data: ByteArray,
revision: Long, width: Int, height: Int, rotation: Int, audio: Boolean): Int
private external fun nativeStreamStop(id: Long)
private external fun nativeStreamPull(id: Long): ByteArray?
private external fun nativeStreamPull(id: Long, position: Long, videoAhead: Long, audioAhead: Long): ByteArray?
private external fun nativeStreamStatus(id: Long): LongArray?
private external fun nativeStreamResync(id: Long): Boolean
private external fun nativeStreamPlaybackCreate(id: Long, pts: Long, audio: Boolean): Long
private external fun nativeStreamPlaybackFeed(handle: Long, pts: Long, packet: ByteArray): Int
private external fun nativeStreamPlaybackPull(handle: Long, frames: Int, limit: Long): ShortArray?
private external fun nativeStreamPlaybackUpdate(handle: Long, now: Long, played: Long, ready: Long,
guard: Long, starved: Boolean): LongArray?
private external fun nativeStreamPlaybackFinish(handle: Long)
private external fun nativeStreamPlaybackEmpty(handle: Long): Boolean
private external fun nativeStreamPlaybackClose(handle: Long)
private external fun nativeStreamOpusCreate(encode: Boolean): Long
private external fun nativeStreamOpusEncode(handle: Long, pcm: ShortArray, compressor: Boolean, maxGainDb: Int, riseRate: Int): ByteArray?
private external fun nativeStreamOpusDecode(handle: Long, packet: ByteArray): ShortArray?

464
tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt
Unescape View File

@ -15,15 +15,15 @@ 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. */
/** Запас кодированного потока остаётся в общем RS-кольце. Worker запрашивает ближайшие кадры,
* общий native controller ведёт минутную историю и темп, AudioTrack задаёт фактические часы.
* Временная нехватка сохраняет decoder references; GAP/вытеснение начинают новую точку KEY. */
class StreamPlayerEngine(private val group: Long, private val source: Long, private val epoch: Long,
private val localPreview: StreamPreview? = null, private val displayDegrees: Int = 0) {
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 top: Int?, val revision: Long, val data: ByteArray)
private data class Frame(val pts: Long, val key: Boolean, val width: Int, val height: Int,
val rotation: Int, val top: Int?, val revision: Long, val data: ByteArray)
private val mutableAspect = MutableStateFlow(16f / 9f)
val aspect: StateFlow<Float> = mutableAspect
private val mutableTop = MutableStateFlow<Int?>(null)
@ -39,294 +39,238 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
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 playback = 0L
val videos = ArrayDeque<Frame>()
val submitted = ArrayDeque<Long>()
var waitingKey = true
var audioBase = -1L
var written = 0L
var videoBase = -1L
var videoStartNs = 0L
var audioEnabled = false
var opened = false
var ended = false
val audioTimestamp = AudioTimestamp()
val bufferInfo = MediaCodec.BufferInfo()
var report = 0L
var pausedAudio = false
var videoRevision = 0L
var reportedDecoded = 0L
var position = -1L
var ready = -1L
var needsAudio = true
var written = 0L
var previousHead = 0L
var headEpoch = 0L
var generation = 0L
var decoded = 0L
var discarded = 0L
var lastHead = -1L
var headChangedNs = 0L
var endDrainUntil = 0L
var headEpoch = 0L
var previousRawHead = 0L
var prefillUntil = 0L
var firstAudioSkipFrames = 0
var prerollFrames = 0L
var warmAudioPts = -1L
var connectDeadline = System.nanoTime() + 10_000_000_000
var decodedThrough = -1L
var pendingOutput = -1
var pendingPts = 0L
var eosQueued = false
var decoderEnded = false
var endUntil = 0L
var reported = 0L
var deadline = System.nanoTime() + 10_000_000_000
val timestamp = AudioTimestamp()
val info = MediaCodec.BufferInfo()
var clockState: LongArray? = null
fun resetVideo() {
try { decoder?.stop() }
try { decoder?.stop(); decoder?.release() }
catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder reset id=$id: ${e.message}") }
try { decoder?.release() }
catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") }
decoder = null; videos.clear(); videoBytes = 0; waitingKey = true
decoder = null; videos.clear(); submitted.clear(); pendingOutput = -1; decodedThrough = -1
waitingKey = true; eosQueued = false; decoderEnded = false
mutableStatus.value = R.string.stream_waiting_key
}
fun resetAudio() {
try { track?.pause(); track?.flush() }
fun resetTimeline(request: Boolean) {
try { track?.pause(); track?.flush(); track?.release() }
catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio reset id=$id: ${e.message}") }
try { track?.release() }
catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") }
track = null
if (opus != 0L) NativeLib.streamOpusClose(opus)
opus = 0; audioBase = -1; written = 0; lastHead = -1; headEpoch = 0; previousRawHead = 0; audios.clear()
firstAudioSkipFrames = 0; prerollFrames = 0; warmAudioPts = -1
}
fun resetTimeline() {
resetAudio(); resetVideo(); videoBase = -1; prefillUntil = 0
connectDeadline = System.nanoTime() + 10_000_000_000
}
// Прогрев допускает джиттер PTS и короткий PLC; большой разрыв начинает новое состояние Opus.
fun prepareAudio(pts: Long) {
if (warmAudioPts >= 0 && (pts < warmAudioPts - 10000 || pts > warmAudioPts + 100000)) {
LogManager.addLog("WARN", "stream", "preroll audio discontinuity id=$id expected=$warmAudioPts pts=$pts")
if (opus != 0L) NativeLib.streamOpusClose(opus)
opus = 0; warmAudioPts = -1
}
if (opus == 0L) { opus = NativeLib.streamOpusCreate(false); check(opus != 0L) }
val missing = if (warmAudioPts < 0) 0 else ((pts - warmAudioPts + 10000) / 20000).toInt()
repeat(missing) {
check(NativeLib.streamOpusDecode(opus, byteArrayOf()) != null) { "Preroll Opus PLC failed" }
prerollFrames += 960
}
track = null; pausedAudio = false; written = 0; previousHead = 0; headEpoch = 0
if (playback != 0L) NativeLib.streamPlaybackClose(playback)
playback = 0; position = -1; generation = 0; needsAudio = true; clockState = null
resetVideo(); deadline = System.nanoTime() + 10_000_000_000
if (request) check(NativeLib.streamResync(id)) { "Media resync refused" }
}
// Отброшенный звук всё равно прогревает Opus; остающиеся пакеты декодируются один раз при воспроизведении.
fun warmAudio(frame: Frame) {
prepareAudio(frame.pts)
check(NativeLib.streamOpusDecode(opus, frame.data) != null) { "Preroll Opus decode failed" }
warmAudioPts = frame.pts + 20000; prerollFrames += 960
}
// PlaybackHeadPosition — unsigned 32 bit: при 48kHz оборачивается примерно через 25 часов.
fun playedFrames(t: AudioTrack): Long {
fun playedFrames(now: Long): Long {
val t = track ?: return 0
val raw = t.playbackHeadPosition.toLong() and 0xffffffffL
if (raw < previousRawHead) headEpoch += 0x100000000L
previousRawHead = raw
return headEpoch + raw
}
fun currentPts(now: Long): Long {
val t = track
if (t != null && audioBase >= 0) {
val head = playedFrames(t)
val played = if (t.getTimestamp(audioTimestamp)) {
var position = audioTimestamp.framePosition
if (position <= 0xffffffffL) {
position += headEpoch
if (position - head > 0x80000000L) position -= 0x100000000L
if (head - position > 0x80000000L) position += 0x100000000L
}
position + (now - audioTimestamp.nanoTime).coerceAtLeast(0) * 48000 / 1_000_000_000
} else {
if (head != lastHead) { lastHead = head; headChangedNs = now }
head + if (head >= written) (now - headChangedNs) * 48000 / 1_000_000_000 else 0
}
return audioBase + played * 1_000_000 / 48000
if (raw < previousHead) headEpoch += 0x100000000L
previousHead = raw
val head = headEpoch + raw
if (pausedAudio || !t.getTimestamp(timestamp)) return head.coerceAtMost(written)
var frames = timestamp.framePosition
if (frames <= 0xffffffffL) {
frames += headEpoch
if (frames - head > 0x80000000L) frames -= 0x100000000L
if (head - frames > 0x80000000L) frames += 0x100000000L
}
return if (videoBase >= 0) videoBase + (now - videoStartNs) / 1000 else 0
return (frames + (now - timestamp.nanoTime).coerceAtLeast(0) * 48000 / 1_000_000_000).coerceIn(0, written)
}
fun writePcm(t: AudioTrack, pcm: ShortArray, initialOffset: Int = 0) {
val deadline = System.nanoTime() + 100_000_000
var offset = initialOffset
fun writePcm(pcm: ShortArray) {
if (track == null) {
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_STEREO).setEncoding(AudioFormat.ENCODING_PCM_16BIT).build())
.setBufferSizeInBytes(AudioTrack.getMinBufferSize(48000, AudioFormat.CHANNEL_OUT_STEREO,
AudioFormat.ENCODING_PCM_16BIT).coerceAtLeast(3840)).setTransferMode(AudioTrack.MODE_STREAM).build()
check(track!!.state == AudioTrack.STATE_INITIALIZED); track!!.play()
LogManager.addLog("INFO", "stream", "audio output id=$id bufferFrames=${track!!.bufferSizeInFrames} stereo=1")
}
val until = System.nanoTime() + 100_000_000
var offset = 0
while (running && offset < pcm.size) {
val count = t.write(pcm, offset, pcm.size - offset, AudioTrack.WRITE_NON_BLOCKING)
check(count >= 0) { "AudioTrack write=$count" }
if (count > 0) {
check(count % 2 == 0) { "AudioTrack returned an incomplete stereo frame: samples=$count" }
offset += count; written += count / 2
}
else {
check(System.nanoTime() < deadline) { "AudioTrack stalled for 100ms" }
Thread.sleep(2)
}
val count = track!!.write(pcm, offset, pcm.size - offset, AudioTrack.WRITE_NON_BLOCKING)
check(count >= 0 && count % 2 == 0) { "AudioTrack write=$count" }
if (count > 0) { offset += count; written += count / 2 }
else { check(System.nanoTime() < until) { "AudioTrack stalled" }; Thread.sleep(2) }
}
}
try {
check(NativeLib.streamOpen(id, group, source, epoch = epoch)) { "Viewer open refused" }
var opened = false
check(NativeLib.streamOpen(id, group, source, audio = localPreview == null, epoch = epoch)) { "Viewer open refused" }
while (running && surface.isValid) {
if (!NativeLib.isReady()) error("Core stopped")
repeat(64) {
val raw = if (!ended) NativeLib.streamPull(id) else null
if (raw != null) {
check(raw.size >= 24)
val b = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN)
val kind = b.get().toInt(); val flags = b.get().toInt() and 255; val key = flags and 1 != 0
val width = b.short.toInt() and 65535; val height = b.short.toInt() and 65535
val broadcastRotation = b.short.toInt() and 65535; val pts = b.long; val size = b.int
val revision = b.int.toLong() and 0xffffffffL
check(size == b.remaining())
when (kind) {
1, 2 -> {
val camera = if (kind == 1) localPreview?.camera(revision) else null
if (kind == 1 && localPreview != null && camera == null) {
LogManager.addLog("WARN", "video", "preview camera unavailable id=$id revision=$revision; wait KEY")
resetVideo(); return@repeat
}
val display = StreamVideoDisplay.forFrame(broadcastRotation, camera, displayDegrees)
val rotation = display.rotation
val frame = Frame(pts, key, width, height, rotation, display.top, revision, ByteArray(size).also { b.get(it) })
if (kind == 1) {
if (revision != videoRevision) {
check(revision > videoRevision) { "Video revision regressed" }
LogManager.addLog("INFO", "video", "viewer configuration id=$id revision=$videoRevision->$revision ${width}x$height " +
"rotation=$rotation broadcastRotation=$broadcastRotation preview=${localPreview != null} top=${display.top}")
resetVideo(); videoRevision = revision
}
if (videos.size >= 128 || videoBytes + size > H264AnnexB.MAX_FRAME ||
(videos.isNotEmpty() && pts - videos.first.pts > 250000)) {
LogManager.addLog("WARN", "video", "viewer video overflow id=$id frames=${videos.size} bytes=$videoBytes revision=$revision")
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
mutableTop.value = frame.top
}
if (videoBase < 0) {
while (audios.isNotEmpty() && audios.first.pts + 20000 <= pts) warmAudio(audios.removeFirst())
if (audios.isNotEmpty() && audios.first.pts < pts)
firstAudioSkipFrames = ((pts - audios.first.pts) * 48000 + 999999).div(1000000).toInt()
prefillUntil = System.nanoTime() + 150_000_000
videoBase = pts; videoStartNs = prefillUntil
LogManager.addLog("INFO", "stream", "viewer playback start id=$id pts=$pts prefillMs=150 " +
"prerollFrames=$prerollFrames skipFrames=$firstAudioSkipFrames audioQ=${audios.size}")
}
videos.addLast(frame); videoBytes += size
}
} else {
if (localPreview != null) return@repeat
if (audios.size >= 16) {
if (videoBase < 0) warmAudio(audios.removeFirst())
else {
LogManager.addLog("WARN", "stream", "audio queue overflow id=$id; reset timeline")
resetTimeline()
}
}
audios.addLast(frame)
}
check(NativeLib.isReady()) { "Core stopped" }
val meta = NativeLib.streamStatus(id)
if (meta != null && meta[5] and 1 != 0L) {
ready = if (localPreview != null) meta[1] else meta[0]
if (playback == 0L) {
audioEnabled = localPreview == null && meta[5] and 2 != 0L
playback = NativeLib.streamPlaybackCreate(id, meta[3], audioEnabled)
check(playback != 0L); position = meta[3]; generation = meta[4]
LogManager.addLog("INFO", "stream", "playback starts id=$id PTS=$position generation=$generation fixedPrefill=0")
}
}
val now = System.nanoTime()
val played = playedFrames(now)
val deviceLead = (written - played).coerceAtLeast(0) * 1_000_000 / 48000
val videoAhead = if (videos.size < 2 && submitted.size < 3) 80000L else -1L
val audioAhead = if (audioEnabled && needsAudio) deviceLead + 80000 else -1L
repeat(4) {
val raw = if (ended) null else NativeLib.streamPull(id, position, videoAhead, audioAhead)
if (raw == null) return@repeat
check(raw.size >= 24)
val b = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN)
val kind = b.get().toInt(); val flags = b.get().toInt() and 255
val width = b.short.toInt() and 65535; val height = b.short.toInt() and 65535
val broadcastRotation = b.short.toInt() and 65535; val pts = b.long; val size = b.int
val revision = b.int.toLong() and 0xffffffffL
check(size == b.remaining())
when (kind) {
1 -> {
val camera = localPreview?.camera(revision)
if (localPreview != null && camera == null) {
LogManager.addLog("WARN", "video", "preview camera unavailable id=$id revision=$revision")
resetTimeline(true); return@repeat
}
val display = StreamVideoDisplay.forFrame(broadcastRotation, camera, displayDegrees)
if (revision != videoRevision) {
check(revision > videoRevision); resetVideo(); videoRevision = revision
deadline = System.nanoTime() + 10_000_000_000
}
3 -> { LogManager.addLog("INFO", "stream", "viewer GAP id=$id"); resetTimeline() }
4 -> opened = true
5 -> { ended = true; if (videoBase < 0) audios.clear() }
6 -> {
mutableStatus.value = when (flags) {
1 -> R.string.stream_group_closed
2 -> R.string.chat_service_stopped
3 -> R.string.stream_source_changed
4 -> R.string.stream_member_removed
else -> R.string.stream_failed
val key = flags and 1 != 0
if (!waitingKey || key) {
val bytes = ByteArray(size).also { b.get(it) }
if (waitingKey) {
check(H264AnnexB.nals(bytes).map { it.first }.toSet().containsAll(setOf(5, 7, 8)))
waitingKey = false; mutableTop.value = display.top
mutableAspect.value = if (display.rotation % 180 == 0) width.toFloat() / height else height.toFloat() / width
}
LogManager.addLog("WARN", "stream", "native viewer closed id=$id group=$group source=$source epoch=$epoch reason=$flags")
ended = true; running = false
check(videos.size < 4) { "Video preparation queue exceeded" }
videos.addLast(Frame(pts, key, width, height, display.rotation, display.top, revision, bytes))
}
}
2 -> if (audioEnabled && playback != 0L) {
val result = NativeLib.streamPlaybackFeed(playback, pts, ByteArray(size).also { b.get(it) })
if (result == -2) resetTimeline(true) else check(result >= 0) { "Opus feed failed=$result" }
}
3 -> { LogManager.addLog("INFO", "stream", "playback GAP id=$id"); resetTimeline(false) }
4 -> opened = true
5 -> { ended = true; if (playback != 0L) NativeLib.streamPlaybackFinish(playback) }
6 -> {
mutableStatus.value = when (flags) {
1 -> R.string.stream_group_closed; 2 -> R.string.chat_service_stopped
3 -> R.string.stream_source_changed; 4 -> R.string.stream_member_removed
else -> R.string.stream_failed
}
else -> error("Unknown media event=$kind")
LogManager.addLog("WARN", "stream", "native viewer closed id=$id reason=$flags"); running = false
}
else -> error("Unknown media event=$kind")
}
}
if (!ended && (!opened || videoBase < 0) && System.nanoTime() > connectDeadline) error("Viewer open timeout")
val now = System.nanoTime()
if (prefillUntil != 0L && now >= prefillUntil) {
if (audios.isNotEmpty() && track == null) {
prepareAudio(audios.first.pts)
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_STEREO).setEncoding(AudioFormat.ENCODING_PCM_16BIT).build())
.setBufferSizeInBytes(AudioTrack.getMinBufferSize(48000, AudioFormat.CHANNEL_OUT_STEREO, AudioFormat.ENCODING_PCM_16BIT)
.coerceAtLeast(19200)).setTransferMode(AudioTrack.MODE_STREAM).build()
check(track!!.state == AudioTrack.STATE_INITIALIZED)
audioBase = audios.first.pts + firstAudioSkipFrames * 1_000_000L / 48000
track!!.play()
LogManager.addLog("INFO", "stream", "audio playback starts id=$id PTS=$audioBase rate=48000 channels=${track!!.channelCount}")
}
if (!ended && (!opened || waitingKey) && System.nanoTime() > deadline) error("Viewer open timeout")
if (playback != 0L && audioEnabled && !pausedAudio) {
val head = playedFrames(System.nanoTime())
val t = track
if (t != null && audios.isNotEmpty()) {
val head = playedFrames(t)
if (written - head < 3840) {
val f = audios.removeFirst()
val pts = f.pts + firstAudioSkipFrames * 1_000_000L / 48000
val missing = ((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")
resetTimeline(); audios.addFirst(f)
} else if (missing >= 0) {
repeat(missing) {
val conceal = NativeLib.streamOpusDecode(opus, byteArrayOf()) ?: error("Opus PLC failed")
writePcm(t, conceal)
}
val pcm = NativeLib.streamOpusDecode(opus, f.data) ?: error("Opus decode failed")
writePcm(t, pcm, firstAudioSkipFrames * 2); firstAudioSkipFrames = 0
} else LogManager.addLog("DEBUG", "stream", "late audio discarded id=$id pts=${f.pts}")
val prime = t == null || (head == 0L && written < t.bufferSizeInFrames)
if (prime || written - head < 960) {
NativeLib.streamPlaybackPull(playback, 480, if (ended) Long.MAX_VALUE else ready)?.let { writePcm(it) }
}
}
if (playback != 0L && ready >= 0) {
val lateVideo = submitted.isNotEmpty() && submitted.first < position - 10000
if (lateVideo && track != null && !pausedAudio) { track!!.pause(); pausedAudio = true }
if (pausedAudio && !lateVideo) { track!!.play(); pausedAudio = false }
val head = playedFrames(System.nanoTime())
val audioEmpty = audioEnabled && written > 0 && head >= written && !ended
clockState = NativeLib.streamPlaybackUpdate(playback, System.nanoTime() / 1000, head,
if (ended) ready.coerceAtLeast(position) else ready,
(deviceLead + 10000).coerceAtLeast(33333), lateVideo || audioEmpty)
check(clockState != null) { "Playback timeline failed" }
position = clockState!![0]; needsAudio = clockState!![5] != 0L
}
repeat(4) {
if (videos.isEmpty()) return@repeat
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)
decoder!!.configure(format, surface, null, 0); decoder!!.start()
}
val index = decoder!!.dequeueInputBuffer(0)
if (index >= 0) {
val input = decoder!!.getInputBuffer(index) ?: error("Missing decoder input")
check(frame.data.size <= input.capacity()); input.clear(); input.put(frame.data)
decoder!!.queueInputBuffer(index, 0, frame.data.size, frame.pts, 0)
submitted.addLast(frame.pts); videos.removeFirst(); decoded++
}
}
val codec = decoder
if (codec != null) {
if (ended && videos.isEmpty() && !eosQueued) {
val index = codec.dequeueInputBuffer(0)
if (index >= 0) { codec.queueInputBuffer(index, 0, 0, decodedThrough.coerceAtLeast(0) + 1,
MediaCodec.BUFFER_FLAG_END_OF_STREAM); eosQueued = true }
}
val clock = currentPts(now)
repeat(8) {
if (videos.isEmpty() || videos.first.pts > clock + 80000) return@repeat
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)
if (pendingOutput < 0) {
val index = codec.dequeueOutputBuffer(info, 0)
if (index < 0) return@repeat
if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0 && info.size == 0) {
codec.releaseOutputBuffer(index, false); decoderEnded = true; return@repeat
}
decoder = MediaCodec.createDecoderByType(MediaFormat.MIMETYPE_VIDEO_AVC)
decoder!!.configure(format, surface, null, 0); decoder!!.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(16) {
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
pendingOutput = index; pendingPts = info.presentationTimeUs; decodedThrough = pendingPts
while (submitted.isNotEmpty() && submitted.first <= pendingPts) submitted.removeFirst()
}
if (pendingPts > position) return@repeat
val render = pendingPts >= position - 100000
codec.releaseOutputBuffer(pendingOutput, render); pendingOutput = -1
if (!render) discarded++ else mutableStatus.value = R.string.stream_live
}
}
if (ended && videos.isEmpty() && audios.isEmpty()) {
if (endDrainUntil == 0L) endDrainUntil = now + 250_000_000
if (now >= endDrainUntil) { mutableStatus.value = R.string.stream_ended; break }
if (ended && videos.isEmpty() && (decoder == null || decoderEnded) && NativeLib.streamPlaybackEmpty(playback)
&& playedFrames(System.nanoTime()) >= written) {
if (endUntil == 0L) endUntil = System.nanoTime() + 100_000_000
if (System.nanoTime() >= endUntil) { mutableStatus.value = R.string.stream_ended; break }
}
if (now - report >= 1_000_000_000) {
val seconds = if (report == 0L) 1.0 else (now - report) / 1_000_000_000.0
report = now
LogManager.addLog("DEBUG", "stream", "viewer id=$id revision=$videoRevision decodeFps=${"%.1f".format((decoded - reportedDecoded) / seconds)} decoded=$decoded discarded=$discarded videoQ=${videos.size}/$videoBytes " +
"audioQ=${audios.size} written=$written clock=${currentPts(now)} waitingKey=$waitingKey")
reportedDecoded = decoded
if (System.nanoTime() - reported >= 1_000_000_000) {
reported = System.nanoTime()
LogManager.addLog("DEBUG", "stream", "viewer id=$id revision=$videoRevision decoded=$decoded discarded=$discarded " +
"videoQ=${videos.size} submitted=${submitted.size} clock=$position reserve=${clockState?.get(1)} " +
"minimum=${clockState?.get(2)} tempo=${clockState?.get(3)} written=$written waitingKey=$waitingKey")
}
Thread.sleep(5)
}
@ -334,13 +278,13 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
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() } catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder stop id=$id: ${e.message}") }
try { decoder?.release() } catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") }
try { track?.stop() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio stop id=$id: ${e.message}") }
try { 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 (!ended && NativeLib.isReady() && System.nanoTime() < deadline) {
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 (playback != 0L) NativeLib.streamPlaybackClose(playback)
val until = System.nanoTime() + 500_000_000
while (NativeLib.isReady() && System.nanoTime() < until) {
val event = NativeLib.streamPull(id)
if (event != null && event[0].toInt() in 5..6) break
Thread.sleep(5)

Write Preview
Loading…
Cancel
Save