feat(streaming): 添加自编码串流模式切换功能

This commit is contained in:
2026-07-23 12:01:08 +08:00
parent a1c29246d7
commit e56723a010
20 changed files with 1684 additions and 16 deletions

View File

@@ -1,5 +1,12 @@
package com.ttstd.fluttercontroller
import io.flutter.embedding.android.FlutterActivity
import io.flutter.embedding.engine.FlutterEngine
class MainActivity : FlutterActivity()
class MainActivity : FlutterActivity() {
override fun configureFlutterEngine(flutterEngine: FlutterEngine) {
super.configureFlutterEngine(flutterEngine)
// 注册自编码解码器插件(仅 Android 平台提供硬解能力)。
flutterEngine.plugins.add(SelfCodecDecoderPlugin())
}
}

View File

@@ -0,0 +1,355 @@
package com.ttstd.fluttercontroller
import android.graphics.SurfaceTexture
import android.media.MediaCodec
import android.media.MediaFormat
import android.os.Handler
import android.os.Looper
import android.util.Log
import android.view.Surface
import io.flutter.embedding.engine.plugins.FlutterPlugin
import io.flutter.embedding.engine.plugins.activity.ActivityAware
import io.flutter.embedding.engine.plugins.activity.ActivityPluginBinding
import io.flutter.plugin.common.MethodCall
import io.flutter.plugin.common.MethodChannel
import io.flutter.view.TextureRegistry
import java.io.ByteArrayOutputStream
import java.nio.ByteBuffer
import java.nio.ByteOrder
private const val CHANNEL = "com.ttstd.fluttercontroller/self_codec"
private const val TAG = "SelfCodecDecoderPlugin"
private const val MAGIC: Byte = 0xAB.toByte()
private const val PROTOCOL_VERSION = 1
private const val HEADER_SIZE = 13
private const val UNIT_TYPE_CONFIG = 1
private const val UNIT_TYPE_FRAME = 2
private const val PLACEHOLDER_W = 1920
private const val PLACEHOLDER_H = 1080
/// 单个单元的分片重组缓冲。
private class ChunkBuffer(val seq: Int, val total: Int) {
private val chunks = arrayOfNulls<ByteArray>(total)
var received = 0
private set
fun add(idx: Int, data: ByteArray) {
if (idx in 0 until total && chunks[idx] == null) {
chunks[idx] = data
received++
}
}
fun isComplete(): Boolean = received == total
fun join(): ByteArray {
val out = ByteArrayOutputStream()
for (c in chunks) if (c != null) out.write(c)
return out.toByteArray()
}
}
private class Frame(val data: ByteArray, val pts: Long, val key: Boolean)
/// 自编码解码器(控制端原生实现)。
///
/// 对应 WebRTCController 的 SelfCodecDecoder.java从 `video` DataChannel 接收
/// 二进制分片,重组为完整单元,使用 Android MediaCodec 硬解 H.264
/// 并渲染到 Flutter 纹理TextureRegistry.SurfaceTextureEntry
class SelfCodecDecoderPlugin :
FlutterPlugin, ActivityAware, MethodChannel.MethodCallHandler {
private var methodChannel: MethodChannel? = null
private var textureRegistry: TextureRegistry? = null
private var textureEntry: TextureRegistry.SurfaceTextureEntry? = null
private var surfaceTexture: SurfaceTexture? = null
private var surface: Surface? = null
private var decoder: MediaCodec? = null
private var configured = false
private var enabled = false
private val configLock = Any()
private val chunkBuffers = HashMap<Int, ChunkBuffer>()
private val pendingConfigs = ArrayList<ByteArray>()
private val frameQueue = ArrayList<Frame>()
private val inputQueue = ArrayList<Int>()
private val mainHandler = Handler(Looper.getMainLooper())
override fun onAttachedToEngine(binding: FlutterPlugin.FlutterPluginBinding) {
textureRegistry = binding.textureRegistry
methodChannel = MethodChannel(binding.binaryMessenger, CHANNEL)
methodChannel?.setMethodCallHandler(this)
}
override fun onDetachedFromEngine(binding: FlutterPlugin.FlutterPluginBinding) {
methodChannel?.setMethodCallHandler(null)
methodChannel = null
textureRegistry = null
}
override fun onAttachedToActivity(binding: ActivityPluginBinding) {}
override fun onDetachedFromActivityForConfigChanges() {}
override fun onReattachedToActivityForConfigChanges(binding: ActivityPluginBinding) {}
override fun onDetachedFromActivity() {}
override fun onMethodCall(call: MethodCall, result: MethodChannel.Result) {
when (call.method) {
"create" -> {
try {
val id = createInternal()
if (id >= 0) result.success(id) else result.error("E_CREATE", "failed", null)
} catch (e: Exception) {
Log.e(TAG, "create error", e)
result.error("E_CREATE", e.message, null)
}
}
"feed" -> {
val bytes = call.arguments as? ByteArray
if (bytes != null) {
try {
onBinaryMessage(ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN))
} catch (e: Exception) {
Log.e(TAG, "feed error", e)
}
}
result.success(null)
}
"setEnabled" -> {
enabled = call.arguments as? Boolean ?: false
if (!enabled) releaseDecoder()
result.success(null)
}
"dispose" -> {
releaseAll()
result.success(null)
}
else -> result.notImplemented()
}
}
private fun createInternal(): Int {
if (textureEntry != null) return textureEntry!!.id().toInt()
val entry = textureRegistry?.createSurfaceTexture() ?: return -1
textureEntry = entry
val st = entry.surfaceTexture()
surfaceTexture = st
st.setDefaultBufferSize(PLACEHOLDER_W, PLACEHOLDER_H)
st.setOnFrameAvailableListener { texture: SurfaceTexture ->
// 解码器将帧渲染到 Surface 后触发,这里更新纹理内容供 Flutter 显示。
try {
texture.updateTexImage()
} catch (e: Exception) {
Log.w(TAG, "updateTexImage failed", e)
}
}
surface = Surface(st)
return entry.id().toInt()
}
private fun onBinaryMessage(buffer: ByteBuffer) {
if (!enabled) return
while (buffer.remaining() >= HEADER_SIZE) {
val start = buffer.position()
val magic = buffer.get()
val seq = buffer.int
val total = buffer.short.toInt()
val idx = buffer.short.toInt()
val len = buffer.int
if (magic != MAGIC) {
Log.w(TAG, "bad magic byte: $magic")
return
}
if (len < 0 || buffer.remaining() < len) {
buffer.position(start)
break
}
val chunkData = ByteArray(len)
buffer.get(chunkData)
handleChunk(seq, total, idx, chunkData)
}
}
private fun handleChunk(seq: Int, total: Int, idx: Int, data: ByteArray) {
var buf = chunkBuffers[seq]
if (buf == null) {
buf = ChunkBuffer(seq, total)
chunkBuffers[seq] = buf
}
buf.add(idx, data)
if (buf.isComplete()) {
chunkBuffers.remove(seq)
handleUnit(buf.join())
}
}
private fun handleUnit(unit: ByteArray) {
if (unit.size < 10) return
val bb = ByteBuffer.wrap(unit).order(ByteOrder.BIG_ENDIAN)
val type = bb.get().toInt() and 0xFF
val pts = bb.int.toLong() and 0xFFFFFFFFL
val isKey = bb.get().toInt() != 0
val len = bb.int
if (len < 0 || bb.remaining() < len) return
val data = ByteArray(len)
bb.get(data)
when (type) {
UNIT_TYPE_CONFIG -> handleConfig(data)
UNIT_TYPE_FRAME -> handleFrame(data, pts, isKey)
}
}
private fun handleConfig(config: ByteArray) {
synchronized(configLock) {
if (configured) return // 忽略重复参数集
pendingConfigs.add(config)
tryConfigure()
}
}
private fun handleFrame(frame: ByteArray, pts: Long, key: Boolean) {
synchronized(configLock) {
frameQueue.add(Frame(frame, pts, key))
tryConfigure()
feed()
}
}
private fun tryConfigure() {
synchronized(configLock) {
if (configured || surface == null || pendingConfigs.isEmpty()) return
val format = MediaFormat.createVideoFormat(
MediaFormat.MIMETYPE_VIDEO_AVC, PLACEHOLDER_W, PLACEHOLDER_H)
var codec: MediaCodec? = null
try {
codec = MediaCodec.createDecoderByType(MediaFormat.MIMETYPE_VIDEO_AVC)
codec.setCallback(object : MediaCodec.Callback() {
override fun onInputBufferAvailable(codec: MediaCodec, index: Int) {
synchronized(configLock) {
inputQueue.add(index)
feed()
}
}
override fun onOutputBufferAvailable(
codec: MediaCodec, index: Int, info: MediaCodec.BufferInfo) {
try {
codec.releaseOutputBuffer(index, true)
} catch (e: Exception) {
Log.w(TAG, "releaseOutputBuffer failed", e)
}
}
override fun onError(codec: MediaCodec, e: MediaCodec.CodecException) {
Log.e(TAG, "decoder error", e)
}
override fun onOutputFormatChanged(codec: MediaCodec, format: MediaFormat) {
val w = format.getInteger(MediaFormat.KEY_WIDTH, 0)
val h = format.getInteger(MediaFormat.KEY_HEIGHT, 0)
if (w > 0 && h > 0) {
surfaceTexture?.setDefaultBufferSize(w, h)
mainHandler.post {
methodChannel?.invokeMethod(
"onSize", mapOf("width" to w, "height" to h))
}
}
}
})
codec.configure(format, surface, null, 0)
codec.start()
decoder = codec
configured = true
feed()
} catch (e: Exception) {
Log.e(TAG, "configure failed", e)
try {
codec?.release()
} catch (_: Exception) {
}
decoder = null
configured = false
}
}
}
private fun feed() {
synchronized(configLock) {
if (!configured) return
// 先喂入参数集SPS / PPS
while (pendingConfigs.isNotEmpty() && inputQueue.isNotEmpty()) {
val config = pendingConfigs.removeAt(0)
val index = inputQueue.removeAt(0)
val inBuf = decoder?.getInputBuffer(index) ?: return
try {
inBuf.clear()
inBuf.put(config)
decoder?.queueInputBuffer(
index, 0, config.size, 0, MediaCodec.BUFFER_FLAG_CODEC_CONFIG)
} catch (e: Exception) {
Log.w(TAG, "queue config failed", e)
}
}
// 再喂入视频帧。
while (frameQueue.isNotEmpty() && inputQueue.isNotEmpty()) {
val frame = frameQueue.removeAt(0)
val index = inputQueue.removeAt(0)
val inBuf = decoder?.getInputBuffer(index) ?: return
try {
inBuf.clear()
inBuf.put(frame.data)
val flags = if (frame.key) MediaCodec.BUFFER_FLAG_KEY_FRAME else 0
decoder?.queueInputBuffer(
index, 0, frame.data.size, frame.pts, flags)
} catch (e: Exception) {
Log.w(TAG, "queue frame failed", e)
}
}
}
}
private fun releaseDecoder() {
synchronized(configLock) {
try {
decoder?.stop()
} catch (_: Exception) {
}
try {
decoder?.release()
} catch (_: Exception) {
}
decoder = null
configured = false
pendingConfigs.clear()
frameQueue.clear()
inputQueue.clear()
}
}
private fun releaseAll() {
releaseDecoder()
try {
surface?.release()
} catch (_: Exception) {
}
surface = null
try {
surfaceTexture?.release()
} catch (_: Exception) {
}
surfaceTexture = null
try {
textureEntry?.release()
} catch (_: Exception) {
}
textureEntry = null
chunkBuffers.clear()
enabled = false
}
}