From f72e2fca94bdedcf72184b36d1f5286d3cef2812 Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 14 Oct 2026 13:36:38 +0200 Subject: [PATCH] Drive Android decoding by codec readiness and bounded working frames --- .../com/utun/chat/data/StreamDecodePump.kt | 94 ++++++++++ .../com/utun/chat/data/StreamPlayerEngine.kt | 161 ++++++++++-------- .../utun/chat/data/StreamDecodePumpTest.kt | 79 +++++++++ 3 files changed, 262 insertions(+), 72 deletions(-) create mode 100644 tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamDecodePump.kt create mode 100644 tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamDecodePumpTest.kt diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamDecodePump.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamDecodePump.kt new file mode 100644 index 00000000..237721c7 --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamDecodePump.kt @@ -0,0 +1,94 @@ +package com.utun.chat.data + +import java.util.ArrayDeque + +/** Один рабочий вход и один готовый выход. Codec владеет внутренним конвейером; + * число поданных кадров и часы показа не ограничивают заполнение этого конвейера. + * Все вызовы принадлежат одному player worker. Ошибки логирует владелец вместе с session ID. */ +internal class StreamDecodePump( + private val submit: (F) -> Boolean, + private val receive: () -> Output?, + private val release: (Output, Boolean) -> Unit +) { + interface Frame { val pts: Long } + data class Output(val index: Int, val pts: Long, val eos: Boolean = false, val empty: Boolean = false) + private val submitted = ArrayDeque() + var input: F? = null + private set + var output: Output? = null + private set + var firstOutputPts: Long? = null + private set + var lastInputPts = -1L + private set + var inputCount = 0L + private set + var outputCount = 0L + private set + var shown = 0L + private set + var discarded = 0L + private set + var decoderLeadUs = 0L + private set + var framePeriodUs = 33333L + private set + var ended = false + private set + private var waitingSince = -1L + private var lastInputAt = -1L + val wantsInput: Boolean get() = input == null && output == null && !ended + val oldestSubmittedPts: Long get() = submitted.firstOrNull() ?: -1L + val inflight: Int get() = submitted.size + + fun offer(frame: F, now: Long) { + // Уже запрошенный ответ может прийти после появления будущего output. + check(input == null && !ended) { "Decoder already owns a working frame" } + // Новые данные после сетевой паузы получают собственное окно ожидания decoder progress. + if (lastInputAt < 0 || now - lastInputAt > 1_000_000) waitingSince = now + input = frame + } + + /** Выход проверяется перед каждым входом. Свободный input buffer берут лишь для готового кадра. */ + fun step(position: Long, now: Long) { + drain(position, now) + if (output != null || ended) return + val frame = input + if (frame != null && submit(frame)) { + check(lastInputPts < frame.pts) { "Decoder input PTS did not advance" } + if (lastInputPts >= 0) framePeriodUs = (frame.pts - lastInputPts).coerceIn(1, 1_000_000) + lastInputPts = frame.pts; lastInputAt = now; inputCount++; submitted.addLast(frame.pts); input = null + drain(position, now) + } + } + + private fun drain(position: Long, now: Long) { + repeat(8) { + if (ended) return + if (output == null) { + val next = receive() ?: return + output = next + if (!next.empty) { + if (firstOutputPts == null) firstOutputPts = next.pts + outputCount++; decoderLeadUs = (lastInputPts - next.pts).coerceAtLeast(0) + while (submitted.isNotEmpty() && submitted.first <= next.pts) submitted.removeFirst() + } + waitingSince = now + } + val next = output!! + if (!next.empty && next.pts > position) return + val render = !next.empty && next.pts >= position - 100000 + release(next, render); output = null + if (!next.empty) { if (render) shown++ else discarded++ } + if (next.eos) ended = true + } + } + + /** Нехватка сети и ожидание PTS допустимы; зависший decoder при наличии данных — ошибка. */ + fun checkProgress(now: Long, finishing: Boolean) { + val hasWork = input != null || (lastInputAt >= 0 && now - lastInputAt < 1_000_000) || finishing + check(ended || output != null || !hasWork || waitingSince < 0 || now - waitingSince <= 10_000_000) { + "Decoder made no output progress: inputs=$inputCount outputs=$outputCount inflight=$inflight lastPTS=$lastInputPts" + } + } +} 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 index ec039c0d..9f3822cf 100644 --- 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 @@ -5,14 +5,16 @@ import android.media.AudioFormat import android.media.AudioTimestamp import android.media.AudioTrack import android.media.MediaCodec +import android.media.MediaCodecInfo +import android.media.MediaCodecList import android.media.MediaFormat +import android.os.Build 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 @@ -23,9 +25,9 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv private val localPreview: StreamPreview? = null, private val displayDegrees: Int = 0) { companion object { private val viewerLock = ReentrantLock() } private class UnsupportedCodec(message: String) : IllegalStateException(message) - private data class Frame(val pts: Long, val key: Boolean, val width: Int, val height: Int, + private data class Frame(override val pts: Long, val key: Boolean, val width: Int, val height: Int, val rotation: Int, val top: Int?, val revision: Long, val codec: StreamVideoCodec, - val config: ByteArray, val data: ByteArray) + val config: ByteArray, val data: ByteArray) : StreamDecodePump.Frame private val mutableAspect = MutableStateFlow(16f / 9f) val aspect: StateFlow = mutableAspect private val mutableTop = MutableStateFlow(null) @@ -47,8 +49,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv var decoder: MediaCodec? = null var track: AudioTrack? = null var playback = 0L - val videos = ArrayDeque() - val submitted = ArrayDeque() + lateinit var video: StreamDecodePump var waitingKey = true var audioEnabled = false var opened = false @@ -62,14 +63,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv var previousHead = 0L var headEpoch = 0L var generation = 0L - var decoded = 0L - var discarded = 0L - var lastSubmittedPts = -1L - var pendingOutput = -1 - var pendingPts = 0L - var pendingEos = false + var playbackStarted = false var eosQueued = false - var decoderEnded = false var endUntil = 0L var reported = 0L var deadline = System.nanoTime() + 10_000_000_000 @@ -77,6 +72,53 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv val info = MediaCodec.BufferInfo() var clockState: LongArray? = null + fun createVideoPump() = StreamDecodePump( + submit = { frame -> + if (decoder == null) { + check(frame.key) + val format = frame.codec.decoderFormat(frame.width, frame.height, frame.rotation, frame.config) + val name = MediaCodecList(MediaCodecList.REGULAR_CODECS).findDecoderForFormat(format) + ?: throw UnsupportedCodec("Unsupported decoder ${frame.codec.label} ${frame.width}x${frame.height}") + decoder = MediaCodec.createByCodecName(name) + val lowLatency = Build.VERSION.SDK_INT >= 30 && decoder!!.codecInfo.getCapabilitiesForType(frame.codec.mime) + .isFeatureSupported(MediaCodecInfo.CodecCapabilities.FEATURE_LowLatency) + if (lowLatency) format.setInteger(MediaFormat.KEY_LOW_LATENCY, 1) + LogManager.addLog("INFO", "video", "viewer codec id=$id type=${frame.codec.label} decoder=$name " + + "config=${frame.config.size} lowLatency=$lowLatency revision=$videoRevision") + decoder!!.configure(format, surface, null, 0); decoder!!.start() + } + val index = decoder!!.dequeueInputBuffer(0) + if (index < 0) false else { + val input = decoder!!.getInputBuffer(index) ?: error("Missing decoder input") + check(frame.data.size <= input.capacity()) { "Decoder input is too small: ${input.capacity()}/${frame.data.size}" } + input.clear(); input.put(frame.data); decoder!!.queueInputBuffer(index, 0, frame.data.size, frame.pts, 0); true + } + }, + receive = { + var output: StreamDecodePump.Output? = null + val codec = decoder + if (codec != null) { + for (attempt in 0 until 3) { + val index = codec.dequeueOutputBuffer(info, 0) + if (index == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED) { + LogManager.addLog("INFO", "video", "viewer output format id=$id revision=$videoRevision ${codec.outputFormat}") + continue + } + if (index >= 0) { + val eos = info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0 + output = StreamDecodePump.Output(index, info.presentationTimeUs, eos, eos && info.size == 0) + } + break + } + } + output + }, + release = { output, render -> + decoder!!.releaseOutputBuffer(output.index, render) + if (render) mutableStatus.value = R.string.stream_live + } + ) + fun observeStatus(meta: LongArray?) { if (meta == null || meta[5] and 1 == 0L) return ready = if (localPreview != null) meta[1] else meta[0] @@ -90,8 +132,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv fun resetVideo() { try { decoder?.stop(); decoder?.release() } catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder reset id=$id: ${e.message}") } - decoder = null; videos.clear(); submitted.clear(); pendingOutput = -1; lastSubmittedPts = -1; pendingEos = false - waitingKey = true; eosQueued = false; decoderEnded = false + decoder = null; video = createVideoPump() + waitingKey = true; eosQueued = false mutableStatus.value = R.string.stream_waiting_key } fun resetTimeline(request: Boolean) { @@ -99,7 +141,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio reset id=$id: ${e.message}") } 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 + playback = 0; position = -1; generation = 0; needsAudio = true; clockState = null; playbackStarted = false resetVideo(); deadline = System.nanoTime() + 10_000_000_000 if (request) check(NativeLib.streamResync(id)) { "Media resync refused" } } @@ -137,6 +179,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv else { check(System.nanoTime() < until) { "AudioTrack stalled" }; Thread.sleep(2) } } } + video = createVideoPump() try { check(NativeLib.streamOpen(id, group, source, audio = localPreview == null, epoch = epoch)) { "Viewer open refused" } while (running && surface.isValid) { @@ -145,11 +188,11 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv 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 + for (work in 0 until 4) { + video.step(position, System.nanoTime() / 1000) + val audioAhead = if (audioEnabled && needsAudio) deviceLead + 80000 else -1L + val raw = if (ended) null else NativeLib.streamPull(id, position, video.wantsInput, audioAhead) + if (raw == null) break check(raw.size >= 28) val b = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN) val kind = b.get().toInt(); val flags = b.get().toInt() and 255 @@ -165,7 +208,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv 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 + resetTimeline(true); continue } val display = StreamVideoDisplay.forFrame(broadcastRotation, camera, displayDegrees) if (revision != videoRevision) { @@ -180,9 +223,11 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv if (waitingKey) { waitingKey = false; mutableTop.value = display.top mutableAspect.value = if (display.rotation % 180 == 0) width.toFloat() / height else height.toFloat() / width + mutableStatus.value = R.string.stream_waiting_decoder + LogManager.addLog("INFO", "video", "viewer KEY accepted id=$id revision=$revision PTS=$pts bytes=$size") } - check(videos.size < 4) { "Video preparation queue exceeded" } - videos.addLast(Frame(pts, key, width, height, display.rotation, display.top, revision, codec, access.first, access.second)) + video.offer(Frame(pts, key, width, height, display.rotation, display.top, revision, codec, access.first, access.second), + System.nanoTime() / 1000) } } 2 -> if (audioEnabled && playback != 0L) { @@ -204,7 +249,12 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv } } if (!ended && (!opened || waitingKey) && System.nanoTime() > deadline) error("Viewer open timeout") - if (playback != 0L && audioEnabled && !pausedAudio) { + if (!playbackStarted && video.firstOutputPts != null) { + playbackStarted = true + LogManager.addLog("INFO", "video", "viewer first decoded frame id=$id revision=$videoRevision " + + "PTS=${video.firstOutputPts} inputs=${video.inputCount} decoderLeadUs=${video.decoderLeadUs}") + } + if (playbackStarted && playback != 0L && audioEnabled && !pausedAudio) { val head = playedFrames(System.nanoTime()) val t = track val prime = t == null || (head == 0L && written < t.bufferSizeInFrames) @@ -212,73 +262,39 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv 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 (playbackStarted && playback != 0L && ready >= 0) { + val lateVideo = video.firstOutputPts == null || (video.oldestSubmittedPts >= 0 && video.oldestSubmittedPts < 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) + maxOf(deviceLead + 10000, video.decoderLeadUs + video.framePeriodUs), 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 format = frame.codec.decoderFormat(frame.width, frame.height, frame.rotation, frame.config) - val name = android.media.MediaCodecList(android.media.MediaCodecList.REGULAR_CODECS).findDecoderForFormat(format) - ?: throw UnsupportedCodec("Unsupported decoder ${frame.codec.label} ${frame.width}x${frame.height}") - decoder = MediaCodec.createByCodecName(name) - LogManager.addLog("INFO", "video", "viewer codec id=$id type=${frame.codec.label} decoder=$name config=${frame.config.size}") - 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); lastSubmittedPts = frame.pts; videos.removeFirst(); decoded++ - } - } val codec = decoder if (codec != null) { - if (ended && videos.isEmpty() && !eosQueued) { + if (ended && video.input == null && video.output == null && !eosQueued) { val index = codec.dequeueInputBuffer(0) - if (index >= 0) { codec.queueInputBuffer(index, 0, 0, lastSubmittedPts.coerceAtLeast(0), + if (index >= 0) { codec.queueInputBuffer(index, 0, 0, video.lastInputPts.coerceAtLeast(0), MediaCodec.BUFFER_FLAG_END_OF_STREAM); eosQueued = true } } - repeat(8) { - 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 - } - pendingOutput = index; pendingPts = info.presentationTimeUs - pendingEos = info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0 - 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 (pendingEos) { decoderEnded = true; pendingEos = false } - if (!render) discarded++ else mutableStatus.value = R.string.stream_live - } + video.step(position, System.nanoTime() / 1000); video.checkProgress(System.nanoTime() / 1000, ended) } - if (ended && videos.isEmpty() && (decoder == null || decoderEnded) && NativeLib.streamPlaybackEmpty(playback) + if (ended && video.input == null && (decoder == null || video.ended) && 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 (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") + LogManager.addLog("DEBUG", "stream", "viewer id=$id revision=$videoRevision inputs=${video.inputCount} " + + "decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded} input=${video.input != null} " + + "output=${video.output?.pts} inflight=${video.inflight} decoderLeadUs=${video.decoderLeadUs} " + + "clock=$position reserve=${clockState?.get(1)} minimum=${clockState?.get(2)} " + + "tempo=${clockState?.get(3)} written=$written waitingKey=$waitingKey") } Thread.sleep(5) } @@ -296,12 +312,13 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv if (playback != 0L) NativeLib.streamPlaybackClose(playback) val until = System.nanoTime() + 500_000_000 while (NativeLib.isReady() && System.nanoTime() < until) { - val event = NativeLib.streamPull(id) + val event = NativeLib.streamPull(id, wantVideo = false, audioAheadUs = -1) 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") + LogManager.addLog("INFO", "stream", "viewer released id=$id inputs=${video.inputCount} " + + "decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded}") } } } diff --git a/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamDecodePumpTest.kt b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamDecodePumpTest.kt new file mode 100644 index 00000000..e966cd8e --- /dev/null +++ b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamDecodePumpTest.kt @@ -0,0 +1,79 @@ +package com.utun.chat.data + +import org.junit.Assert.* +import org.junit.Test +import java.util.ArrayDeque + +class StreamDecodePumpTest { + private data class Frame(override val pts: Long) : StreamDecodePump.Frame + private class Codec(val depth: Int) { + val queued = ArrayDeque() + var released = 0 + var blocked = false + var finishing = false + var held = false + var nextIndex = 0 + val pump = StreamDecodePump( + submit = { if (blocked) false else { queued.addLast(it); true } }, + receive = { + if (held || (queued.size < depth && !finishing) || queued.isEmpty()) null + else { held = true; StreamDecodePump.Output(nextIndex++, queued.removeFirst().pts) } + }, + release = { _, _ -> check(held); held = false; released++ } + ) + } + + @Test fun fourAndEightFramePipelinesStartWithFrozenPresentationClock() { + for (depth in listOf(4, 8)) { + val codec = Codec(depth) + repeat(depth) { i -> + assertTrue(codec.pump.wantsInput) + codec.pump.offer(Frame(i * 66666L), i * 66666L) + codec.pump.step(0, i * 66666L) + } + assertEquals(depth.toLong(), codec.pump.inputCount) + assertEquals(1L, codec.pump.outputCount) + assertEquals(1, codec.released) + assertEquals(0L, codec.pump.firstOutputPts) + assertEquals((depth - 1) * 66666L, codec.pump.decoderLeadUs) + } + } + + @Test fun futureOutputStopsReadsAndNeverReleasesEarly() { + val codec = Codec(1) + codec.pump.offer(Frame(100000), 0); codec.pump.step(0, 0) + assertFalse(codec.pump.wantsInput); assertNotNull(codec.pump.output); assertEquals(0, codec.released) + codec.pump.step(99999, 15_000_000); codec.pump.checkProgress(15_000_000, false) + assertEquals(0, codec.released) + codec.pump.step(100000, 15_000_001) + assertEquals(1, codec.released); assertTrue(codec.pump.wantsInput) + } + + @Test fun blockedCodecKeepsOneInputAndReportsStall() { + val codec = Codec(4); codec.blocked = true + codec.pump.offer(Frame(0), 0); codec.pump.step(0, 0) + assertNotNull(codec.pump.input); assertFalse(codec.pump.wantsInput) + try { codec.pump.checkProgress(10_000_001, false); fail("Stalled decoder must fail") } + catch (expected: IllegalStateException) { assertTrue(expected.message!!.contains("no output progress")) } + } + + @Test fun networkPauseKeepsReferencesAndAllowsIncompletePipelineToResume() { + val codec = Codec(4) + repeat(2) { i -> codec.pump.offer(Frame(i * 40000L), i * 40000L); codec.pump.step(0, i * 40000L) } + codec.pump.checkProgress(20_000_000, false) + repeat(2) { i -> + val now = 20_000_000 + i * 40000L + codec.pump.offer(Frame((i + 2) * 40000L), now); codec.pump.step(0, now); codec.pump.checkProgress(now, false) + } + assertEquals(4L, codec.pump.inputCount); assertEquals(1, codec.released) + } + + @Test fun finishDrainsShortStreamAndPreservesTheLastPicture() { + val codec = Codec(8) + repeat(3) { i -> codec.pump.offer(Frame(i * 40000L), i * 40000L); codec.pump.step(0, i * 40000L) } + assertEquals(0, codec.released); codec.finishing = true + codec.pump.step(0, 120000); assertEquals(1, codec.released); assertFalse(codec.pump.wantsInput) + codec.pump.step(40000, 160000); assertEquals(2, codec.released) + codec.pump.step(80000, 200000); assertEquals(3, codec.released) + } +}