From a3c5fd44ad9b2c03a2c3a3bc6ff97316f1072356 Mon Sep 17 00:00:00 2001 From: fdp <1286779656@qq.com> Date: Wed, 24 Sep 2025 18:16:41 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=962?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../yunqiinnovation/ble_service/BleService.kt | 322 ++++-------- .../ble_service/OpusAudioManager.kt | 491 ++++++++++++++++++ 2 files changed, 597 insertions(+), 216 deletions(-) create mode 100644 local_plugins/ble_service/android/src/main/kotlin/com/yunqiinnovation/ble_service/OpusAudioManager.kt diff --git a/local_plugins/ble_service/android/src/main/kotlin/com/yunqiinnovation/ble_service/BleService.kt b/local_plugins/ble_service/android/src/main/kotlin/com/yunqiinnovation/ble_service/BleService.kt index aad017767..c2a8d578a 100644 --- a/local_plugins/ble_service/android/src/main/kotlin/com/yunqiinnovation/ble_service/BleService.kt +++ b/local_plugins/ble_service/android/src/main/kotlin/com/yunqiinnovation/ble_service/BleService.kt @@ -9,12 +9,7 @@ import android.os.* import androidx.lifecycle.MutableLiveData import java.util.* import java.util.concurrent.CopyOnWriteArrayList -// Import Jieli Opus SDK classes -import com.jieli.jl_audio_decode.opus.model.OpusOption -import com.jieli.jl_audio_decode.callback.OnDecodeStreamCallback -import com.jieli.jl_audio_decode.callback.OnEncodeStreamCallback -import com.jieli.jl_audio_decode.opus.OpusManager -import com.jieli.jl_audio_decode.exceptions.OpusException + import java.io.File @@ -108,18 +103,16 @@ object BleService { // 回调集合 - 使用线程安全的集合存储所有回调 private val callbacks = CopyOnWriteArrayList() - // Opus解码器实例 - private var opusManager: OpusManager? = null - private var option: OpusOption? = null + // Opus音频管理器 + private val opusAudioManager = OpusAudioManager() private val mainHandler = Handler(Looper.getMainLooper()) - // 解码音频数据队列和处理线程 - private val audioDataQueue = LinkedBlockingQueue() - private var audioQueueProcessorThread: Thread? = null + // // 解码音频数据队列和处理线程 + // private val audioDataQueue = LinkedBlockingQueue() + // private var audioQueueProcessorThread: Thread? = null // 音频数据缓存 - private val audioDataBuffer = mutableListOf() - + //private val audioDataBuffer = mutableListOf() // 音频数据发送相关 private val audioSendQueue = LinkedBlockingQueue() private var audioSendThread: Thread? = null @@ -127,12 +120,10 @@ object BleService { private val isAudioSending = AtomicBoolean(false) // 音频数据分块发送的常量 - private val AUDIO_CHUNK_SIZE = 120 // 每次发送80字节 - private val AUDIO_SEND_INTERVAL = 60L // 发送间隔20ms - + private val AUDIO_CHUNK_SIZE = 120 // 每次发送120字节 + private val AUDIO_SEND_INTERVAL = 60L // 发送间隔60ms // 音频数据缓冲区，用于累积数据到80字节再发送 private val audioBuffer = mutableListOf() - // 重发机制相关常量 private val MAX_RETRY_COUNT = 1 // 最大重试次数 private val RETRY_DELAY = 30L // 重试延迟时间(毫秒) @@ -213,17 +204,15 @@ object BleService { recordfile1 = RecordingFile(this.context) recordfile1!!.fileName = "重新压缩" - // 启动队列处理 - startAudioQueueProcessing() - // 初始化OpusManager和OpusOption - try { - opusManager = OpusManager() - option = OpusOption() - Log.d(TAG, "OpusManager初始化成功") - } catch (e: OpusException) { - Log.e(TAG, "OpusManager初始化失败: ${e.message}", e) + // 初始化Opus音频管理器 + if (!opusAudioManager.initialize()) { + Log.e(TAG, "Opus音频管理器初始化失败") // 根据需要决定是否因为Opus初始化失败而返回false + } else { + // 设置Opus音频管理器回调 + setupOpusAudioManagerCallback() + Log.d(TAG, "Opus音频管理器初始化成功") } // 初始化指令发送器 @@ -237,6 +226,35 @@ object BleService { } } + /** + * 设置Opus音频管理器回调 + */ + private fun setupOpusAudioManagerCallback() { + opusAudioManager.setCallback(object : OpusAudioManager.OpusAudioCallback { + override fun onAudioDataDecoded(data: ByteArray, channel: Int) { + // 将解码后的音频数据通过BLE服务回调传递给上层 + notifyAudioDataReceived(data, channel) + } + + override fun onAudioDataEncoded(data: ByteArray) { + // 将编码后的音频数据加入发送队列 + addAudioDataToSendQueue(data) + } + + override fun onDecodeStreamStateChanged(isStarted: Boolean) { + Log.d(TAG, "Opus解码流状态变化: $isStarted") + } + + override fun onEncodeStreamStateChanged(isStarted: Boolean) { + Log.d(TAG, "Opus编码流状态变化: $isStarted") + } + + override fun onError(errorType: String, code: Int, message: String?) { + Log.e(TAG, "Opus音频管理器错误 [$errorType]: [$code] $message") + } + }) + } + /** * 添加回调监听器 * @param callback 要添加的回调监听器 @@ -313,7 +331,7 @@ object BleService { /** * 移除回调监听器 - * @param callback 要移除的回调监听器 + * @param callback 要移除回调监听器 * @return 是否移除成功 */ fun removeCallback(callback: Callback): Boolean { @@ -339,22 +357,7 @@ object BleService { * @param data 音频数据字节数组，通常为PCM格式 */ fun writeExternalAudioData(data: ByteArray) { - // 检查OpusManager是否已初始化 - if (opusManager == null) { - Log.e(TAG, "opusManager 未初始化") - return - } - - // 如果已经在编码流中，直接写入数据 - if (opusManager?.isEncodeStream == true) { - Log.d(TAG, "正在进行Opus编码流，写入音频数据") - // 将外部音频数据写入编码流，每次处理1280字节 - opusManager?.writeEncodeStream(data) - } else { - Log.w(TAG, "Opus编码流未启动，无法写入音频数据") - // 可选：自动启动编码流 - // startOpusEncodeStream() - } + opusAudioManager.writeExternalAudioData(data) } // ====================================================================================================== @@ -835,7 +838,7 @@ object BleService { return } Log.d(TAG, "找到通话音频服务所需特征") -startBytesStatistics() + startBytesStatistics() writeChar?.writeType = BluetoothGattCharacteristic.WRITE_TYPE_NO_RESPONSE callWriteChar?.writeType = BluetoothGattCharacteristic.WRITE_TYPE_NO_RESPONSE //要先设置音频服务的通知，否则接收不到 @@ -935,170 +938,65 @@ startBytesStatistics() } /** - * 处理音频数据 - * @param data 音频数据 (Opus) + * 处理接收到的音频数据 + * @param data 音频数据 (Opus编码) */ private fun processAudioData(data: ByteArray) { - try { - - - if (opusManager?.isDecodeStream == true) { - recordfile?.saveAudioDataToWav(data) - // 将完整的帧加入队列，由专门的线程处理 - audioDataQueue.offer(data) - - } else { - // Log.d(TAG, "Opus解码流未启动，忽略音频数据") - } - } catch (e: Exception) { - Log.e(TAG, "处理音频数据异常: ${e.message}", e) - } + // 委托给Opus音频管理器处理 + opusAudioManager.processAudioData(data) } // ====================================================================================================== - // Opus 解码相关 + // Opus 解码相关 - 委托给OpusAudioManager // ====================================================================================================== + /** + * 开始Opus数据流解码 + * @param hasHeader 是否包含文件头，通常BLE传输的Opus没有文件头 + * @param channel 音频通道数 + * @param sampleRate 采样率，确认设备端Opus编码采样率 + * @param packetSize 包大小，确认设备端Opus编码帧长 + * @return 是否成功开始解码 + */ public fun startOpusStreamDecoding( - hasHeader: Boolean = false, // 通常BLE传输的Opus没有文件头 + hasHeader: Boolean = false, channel: Int = 1, - sampleRate: Int = 16000, // 确认设备端Opus编码采样率 - packetSize: Int = 40 // 确认设备端Opus编码帧长，必须与iOS版本frameSize保持一致 + sampleRate: Int = 16000, + packetSize: Int = 40 ): Boolean { - if (opusManager == null) { - Log.e(TAG, "OpusManager未初始化，无法开始解码") - return false - } - // 如果已经在解码流，先停止 - if (opusManager?.isDecodeStream == true) { - opusManager?.stopDecodeStream() - Log.d(TAG, "已停止正在进行的Opus解码流") - } - - // 清理音频数据缓存，确保开始时是干净的状态 - synchronized(audioDataBuffer) { - audioDataBuffer.clear() - Log.d(TAG, "开始解码前已清理音频数据缓存") - } - - option!!.setHasHead(hasHeader) - option!!.setChannel(channel) - option!!.setSampleRate(sampleRate) - option!!.setPacketSize(packetSize) - - Log.d(TAG, "准备开始Opus数据流解码, 参数: $option") - - var streamStartedSuccessfully = false - opusManager?.startDecodeStream(option, object : OnDecodeStreamCallback { - override fun onDecodeStream(data: ByteArray?) { - if (data != null) { - if (option!!.getChannel() >= 1) { - notifyAudioDataReceived(data,option!!.getChannel()) - } else { - Log.e(TAG, "Opus解码数据错误: ${data.size} bytes") - } - } - } - - override fun onStart() { - streamStartedSuccessfully = true - Log.i(TAG, "Opus数据流解码已开始") - // 可以通过回调通知上层解码已开始 - } - - override fun onComplete(outPath: String?) { - Log.i(TAG, "Opus数据流解码完成: $outPath (通常流式解码不会调用此方法)") - } - - override fun onError(code: Int, message: String?) { - Log.e(TAG, "Opus数据流解码错误: [$code] $message") - // 可以通过回调通知上层解码错误 - } - }) - // 注意：startDecodeStream是异步启动的，这里的返回值可能不准确反映是否真的成功启动。 - // onStart回调才是真正启动成功的标志。为了简单，我们暂时认为调用即尝试启动。 - // 后续可以改造为返回Future或通过回调传递启动结果。 - return true // 暂定为调用即成功 + return opusAudioManager.startDecodeStream(hasHeader, channel, sampleRate, packetSize) } + /** + * 停止Opus数据流解码 + * @return 是否成功停止解码 + */ private fun stopOpusStreamDecoding(): Boolean { - if (opusManager?.isDecodeStream == true) { - opusManager?.stopDecodeStream() - - // 清理音频数据缓存 - synchronized(audioDataBuffer) { - audioDataBuffer.clear() - Log.d(TAG, "已清理音频数据缓存") - } - - Log.i(TAG, "已停止Opus数据流解码") - return true - } - Log.d(TAG, "Opus数据流未在解码或OpusManager未初始化") - return false + return opusAudioManager.stopDecodeStream() } -// ====================================================================================================== - // Opus 编码相关 + // ====================================================================================================== + // Opus 编码相关 - 委托给OpusAudioManager // ====================================================================================================== + /** + * 开始Opus数据流编码 + * @return 是否成功开始编码 + */ private fun startOpusEncodeStream(): Boolean { - if (opusManager == null) { - Log.e(TAG, "OpusManager未初始化，无法开始编码") - return false - } - - // 如果已经在编码流，先停止 - if (opusManager?.isEncodeStream == true) { - Log.d(TAG, "Opus编码流已在运行") - return true - } - - // 启动音频发送线程 - startAudioSendThread() - - var streamStartedSuccessfully = false - opusManager?.startEncodeStream(object : OnEncodeStreamCallback { - override fun onEncodeStream(data: ByteArray?) { - if (data != null) { - // 编码完成的数据处理： - // 1. 保存到WAV文件 - - // 2. 将编码后的数据加入发送队列进行分块发送 - addAudioDataToSendQueue(data) - } else { - Log.w(TAG, "编码回调收到空数据") - } - } - - override fun onStart() { - streamStartedSuccessfully = true - Log.i(TAG, "Opus数据流编码已开始") - } - - override fun onComplete(outPath: String?) { - Log.i(TAG, "Opus数据流编码完成: $outPath") - } - - override fun onError(code: Int, message: String?) { - Log.e(TAG, "Opus数据流编码错误: [$code] $message") - } - }) - - return true + startAudioSendThread() + return opusAudioManager.startEncodeStream() } + /** + * 停止Opus数据流编码 + * @return 是否成功停止编码 + */ private fun stopOpusEncodeStream(): Boolean { - if (opusManager?.isEncodeStream == true) { - opusManager?.stopEncodeStream() - stopAudioSendThread() // 停止音频发送线程 - Log.i(TAG, "已停止Opus数据流编码") - return true - } - Log.d(TAG, "Opus数据流未在编码或OpusManager未初始化") - return false + stopAudioSendThread() + return opusAudioManager.stopEncodeStream() } /** @@ -1132,10 +1030,11 @@ startBytesStatistics() } } + /** * 启动音频数据发送线程 */ - private fun startAudioSendThread() { + private fun startAudioSendThread() { if (audioSendThread?.isAlive == true) { Log.d(TAG, "音频发送线程已在运行") return @@ -1182,7 +1081,7 @@ startBytesStatistics() Log.i(TAG, "音频发送线程已停止，队列和缓冲区已清空") } - /** + /** * 发送音频数据块到设备（带重发机制） * @param chunk 要发送的音频数据块 * @param retryCount 当前重试次数 @@ -1322,6 +1221,7 @@ startBytesStatistics() BleConst.CODEC_CONTROL_CLOSE.toByte(), BleConst.AUDIO_CHANNEL_STEREO.toByte() ) + return commandSender.sendCommand(BleConst.CMD_CONTROL_CODEC.toByte(), cmdData) } @@ -1348,6 +1248,7 @@ startBytesStatistics() fun openA2DPDecoder(channelMode: Int = BleConst.AUDIO_CHANNEL_STEREO): Boolean { Log.i(TAG, "打开A2DP解码器...") startOpusEncodeStream() + startOpusStreamDecoding(false, 2, 16000, 80) val cmdData = byteArrayOf( BleConst.CODEC_CONTROL_A2DP_PLAY.toByte(), @@ -1473,7 +1374,7 @@ startBytesStatistics() * 清理资源 */ fun cleanup() { - Log.i(TAG, "清理BLE服务资源...") + Log.i(TAG, "清理BLE服务资源...") // 重置指令发送器 commandSender.reset() @@ -1484,19 +1385,33 @@ startBytesStatistics() // 断开连接 disconnect() + // 停止统计 + stopBytesStatistics() + + // 清理Opus音频管理器 + opusAudioManager.cleanup() + + // 停止音频发送线程 + stopAudioSendThread() + // 清理回调 clearCallbacks() - // 停止音频处理 + // 清理录音文件 + recordfile = null + recordfile1 = null + // 停止音频处理 stopAudioSendThread() stopOpusStreamDecoding() stopOpusEncodeStream() isInitialized = false - Log.i(TAG, "BLE服务资源清理完成") + Log.d(TAG, "BLE服务资源清理完成") } + + /** * 向所有回调监听器分发扫描结果 */ @@ -1585,31 +1500,6 @@ startBytesStatistics() } } - /** - * 启动音频队列处理 - */ - private fun startAudioQueueProcessing() { - audioQueueProcessorThread = Thread { - while (!Thread.currentThread().isInterrupted) { - try { - // 从队列中取出音频数据进行解码 - val audioData = audioDataQueue.take() // 阻塞等待数据 - - opusManager?.writeAudioStream(audioData) - } catch (e: InterruptedException) { - Log.d(TAG, "音频队列处理线程被中断") - break - } catch (e: Exception) { - Log.e(TAG, "音频队列处理异常: ${e.message}", e) - } - } - }.apply { - name = "AudioQueueProcessor" - start() - } - } - - } diff --git a/local_plugins/ble_service/android/src/main/kotlin/com/yunqiinnovation/ble_service/OpusAudioManager.kt b/local_plugins/ble_service/android/src/main/kotlin/com/yunqiinnovation/ble_service/OpusAudioManager.kt new file mode 100644 index 000000000..d3eb804b7 --- /dev/null +++ b/local_plugins/ble_service/android/src/main/kotlin/com/yunqiinnovation/ble_service/OpusAudioManager.kt @@ -0,0 +1,491 @@ +package com.yunqiinnovation.ble_service + +import android.os.Handler +import android.os.Looper +import android.util.Log +import com.jieli.jl_audio_decode.opus.model.OpusOption +import com.jieli.jl_audio_decode.callback.OnDecodeStreamCallback +import com.jieli.jl_audio_decode.callback.OnEncodeStreamCallback +import com.jieli.jl_audio_decode.opus.OpusManager +import com.jieli.jl_audio_decode.exceptions.OpusException +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.atomic.AtomicBoolean + +/** + * Opus音频编解码管理器 + * 负责处理音频数据的Opus编码和解码功能 + * + * 功能包括： + * 1. Opus解码流管理 + * 2. Opus编码流管理 + * 3. 音频数据缓冲和处理 + * 4. 编解码状态管理 + */ +class OpusAudioManager { + + companion object { + private const val TAG = "OpusAudioManager" + + // 音频数据分块发送的常量 + private const val AUDIO_CHUNK_SIZE = 120 // 每次发送120字节 + private const val AUDIO_SEND_INTERVAL = 60L // 发送间隔60ms + + // 重发机制相关常量 + private const val MAX_RETRY_COUNT = 1 // 最大重试次数 + private const val RETRY_DELAY = 30L // 重试延迟时间(毫秒) + } + + // ====================================================================================================== + // 回调接口定义 + // ====================================================================================================== + + /** + * Opus音频管理器回调接口 + */ + interface OpusAudioCallback { + /** + * 音频数据解码完成回调 + * @param data 解码后的音频数据 + * @param channel 音频通道数 + */ + fun onAudioDataDecoded(data: ByteArray, channel: Int) + + /** + * 音频数据编码完成回调 + * @param data 编码后的音频数据 + */ + fun onAudioDataEncoded(data: ByteArray) + + /** + * 解码流状态变化回调 + * @param isStarted 是否已启动 + */ + fun onDecodeStreamStateChanged(isStarted: Boolean) + + /** + * 编码流状态变化回调 + * @param isStarted 是否已启动 + */ + fun onEncodeStreamStateChanged(isStarted: Boolean) + + /** + * 错误回调 + * @param errorType 错误类型 (decode/encode) + * @param code 错误代码 + * @param message 错误信息 + */ + fun onError(errorType: String, code: Int, message: String?) + } + + // ====================================================================================================== + // 成员变量 + // ====================================================================================================== + + // Opus解码器实例 + private var opusManager: OpusManager? = null + private var option: OpusOption? = null + private val mainHandler = Handler(Looper.getMainLooper()) + + // 解码音频数据队列和处理线程 + private val audioDataQueue = LinkedBlockingQueue() + private var audioQueueProcessorThread: Thread? = null + + // 音频数据缓存 + private val audioDataBuffer = mutableListOf() + + // 音频数据发送相关 + private val audioSendQueue = LinkedBlockingQueue() + private var audioSendThread: Thread? = null + private val audioSendHandler = Handler(Looper.getMainLooper()) + private val isAudioSending = AtomicBoolean(false) + + // 音频数据缓冲区，用于累积数据到指定字节再发送 + private val audioBuffer = mutableListOf() + + // 回调接口 + private var callback: OpusAudioCallback? = null + + // 初始化状态 + private var isInitialized = false + + // ====================================================================================================== + // 初始化和配置方法 + // ====================================================================================================== + + /** + * 初始化Opus音频管理器 + * @return 是否初始化成功 + */ + fun initialize(): Boolean { + if (isInitialized) return true + + try { + // 初始化OpusManager和OpusOption + opusManager = OpusManager() + option = OpusOption() + + // 启动队列处理 + startAudioQueueProcessing() + + isInitialized = true + Log.d(TAG, "OpusAudioManager初始化成功") + return true + } catch (e: OpusException) { + Log.e(TAG, "OpusAudioManager初始化失败: ${e.message}", e) + return false + } + } + + /** + * 设置回调接口 + * @param callback 回调接口实例 + */ + fun setCallback(callback: OpusAudioCallback?) { + this.callback = callback + } + + /** + * 清理资源 + */ + fun cleanup() { + Log.d(TAG, "开始清理OpusAudioManager资源") + + // 停止编解码流 + stopDecodeStream() + stopEncodeStream() + + // 停止音频发送线程 + stopAudioSendThread() + + // 停止音频队列处理线程 + audioQueueProcessorThread?.interrupt() + audioQueueProcessorThread = null + + // 清理缓存 + synchronized(audioDataBuffer) { + audioDataBuffer.clear() + } + synchronized(audioBuffer) { + audioBuffer.clear() + } + audioDataQueue.clear() + audioSendQueue.clear() + + // 清理Opus相关资源 + opusManager = null + option = null + callback = null + + isInitialized = false + Log.d(TAG, "OpusAudioManager资源清理完成") + } + + // ====================================================================================================== + // Opus解码相关方法 + // ====================================================================================================== + + /** + * 开始Opus数据流解码 + * @param hasHeader 是否包含文件头，通常BLE传输的Opus没有文件头 + * @param channel 音频通道数 + * @param sampleRate 采样率，确认设备端Opus编码采样率 + * @param packetSize 包大小，确认设备端Opus编码帧长 + * @return 是否成功开始解码 + */ + fun startDecodeStream( + hasHeader: Boolean = false, + channel: Int = 1, + sampleRate: Int = 16000, + packetSize: Int = 40 + ): Boolean { + if (!isInitialized || opusManager == null) { + Log.e(TAG, "OpusManager未初始化，无法开始解码") + return false + } + + // 如果已经在解码流，先停止 + if (opusManager?.isDecodeStream == true) { + opusManager?.stopDecodeStream() + Log.d(TAG, "已停止正在进行的Opus解码流") + } + + // 清理音频数据缓存，确保开始时是干净的状态 + synchronized(audioDataBuffer) { + audioDataBuffer.clear() + Log.d(TAG, "开始解码前已清理音频数据缓存") + } + + // 配置解码参数 + option?.apply { + setHasHead(hasHeader) + setChannel(channel) + setSampleRate(sampleRate) + setPacketSize(packetSize) + } + + Log.d(TAG, "准备开始Opus数据流解码, 参数: hasHeader=$hasHeader, channel=$channel, sampleRate=$sampleRate, packetSize=$packetSize") + + try { + opusManager?.startDecodeStream(option, object : OnDecodeStreamCallback { + override fun onDecodeStream(data: ByteArray?) { + if (data != null) { + val channelCount = option?.getChannel() ?: 1 + if (channelCount >= 1) { + callback?.onAudioDataDecoded(data, channelCount) + } else { + Log.e(TAG, "Opus解码数据错误: ${data.size} bytes") + } + } + } + + override fun onStart() { + Log.i(TAG, "Opus数据流解码已开始") + callback?.onDecodeStreamStateChanged(true) + } + + override fun onComplete(outPath: String?) { + Log.i(TAG, "Opus数据流解码完成: $outPath") + callback?.onDecodeStreamStateChanged(false) + } + + override fun onError(code: Int, message: String?) { + Log.e(TAG, "Opus数据流解码错误: [$code] $message") + callback?.onError("decode", code, message) + callback?.onDecodeStreamStateChanged(false) + } + }) + + return true + } catch (e: Exception) { + Log.e(TAG, "启动Opus解码流失败: ${e.message}", e) + callback?.onError("decode", -1, e.message) + return false + } + } + + /** + * 停止Opus数据流解码 + * @return 是否成功停止解码 + */ + fun stopDecodeStream(): Boolean { + if (opusManager?.isDecodeStream == true) { + opusManager?.stopDecodeStream() + + // 清理音频数据缓存 + synchronized(audioDataBuffer) { + audioDataBuffer.clear() + Log.d(TAG, "已清理音频数据缓存") + } + + Log.i(TAG, "已停止Opus数据流解码") + callback?.onDecodeStreamStateChanged(false) + return true + } + Log.d(TAG, "Opus数据流未在解码或OpusManager未初始化") + return false + } + + /** + * 处理接收到的音频数据进行解码 + * @param data 音频数据 (Opus编码) + */ + fun processAudioData(data: ByteArray) { + if (!isInitialized) { + Log.w(TAG, "OpusAudioManager未初始化，忽略音频数据") + return + } + + if (opusManager?.isDecodeStream == true) { + // 将数据加入队列进行异步处理 + audioDataQueue.offer(data) + } else { + Log.d(TAG, "Opus解码流未启动，忽略音频数据") + } + } + + // ====================================================================================================== + // Opus编码相关方法 + // ====================================================================================================== + + /** + * 开始Opus数据流编码 + * @return 是否成功开始编码 + */ + fun startEncodeStream(): Boolean { + if (!isInitialized || opusManager == null) { + Log.e(TAG, "OpusManager未初始化，无法开始编码") + return false + } + + // 如果已经在编码流，先停止 + if (opusManager?.isEncodeStream == true) { + Log.d(TAG, "Opus编码流已在运行") + return true + } + + try { + Log.d(TAG, "准备开始Opus数据流编码") + opusManager?.startEncodeStream(object : OnEncodeStreamCallback { + override fun onEncodeStream(data: ByteArray?) { + Log.d(TAG, "Opus编码回调收到数据，大小: ${data?.size ?: 0} bytes") + if (data != null) { + + // 编码完成的数据处理： + // 1. 通知回调 + callback?.onAudioDataEncoded(data) + + // // 2. 将编码后的数据加入发送队列进行分块发送 + // addAudioDataToSendQueue(data) + } else { + Log.w(TAG, "编码回调收到空数据") + } + } + + override fun onStart() { + Log.i(TAG, "Opus数据流编码已开始") + callback?.onEncodeStreamStateChanged(true) + } + + override fun onComplete(outPath: String?) { + Log.i(TAG, "Opus数据流编码完成: $outPath") + callback?.onEncodeStreamStateChanged(false) + } + + override fun onError(code: Int, message: String?) { + Log.e(TAG, "Opus数据流编码错误: [$code] $message") + callback?.onError("encode", code, message) + callback?.onEncodeStreamStateChanged(false) + } + }) + + return true + } catch (e: Exception) { + Log.e(TAG, "启动Opus编码流失败: ${e.message}", e) + callback?.onError("encode", -1, e.message) + return false + } + } + + /** + * 停止Opus数据流编码 + * @return 是否成功停止编码 + */ + fun stopEncodeStream(): Boolean { + if (opusManager?.isEncodeStream == true) { + opusManager?.stopEncodeStream() + stopAudioSendThread() // 停止音频发送线程 + Log.i(TAG, "已停止Opus数据流编码") + callback?.onEncodeStreamStateChanged(false) + return true + } + Log.d(TAG, "Opus数据流未在编码或OpusManager未初始化") + return false + } + + /** + * 写入外部音频数据到Opus编码流 + * 该函数用于将外部音频数据（如麦克风录音）写入到Opus编码器中 + * + * @param data 音频数据字节数组，通常为PCM格式 + */ + fun writeExternalAudioData(data: ByteArray) { + if (!isInitialized || opusManager == null) { + Log.e(TAG, "OpusManager未初始化") + return + } + + // 如果已经在编码流中，直接写入数据 + if (opusManager?.isEncodeStream == true) { + Log.d(TAG, "正在进行Opus编码流，写入音频数据") + // 将外部音频数据写入编码流 + opusManager?.writeEncodeStream(data) + } else { + Log.w(TAG, "Opus编码流未启动，无法写入音频数据") + } + } + + // ====================================================================================================== + // 状态查询方法 + // ====================================================================================================== + + /** + * 检查是否正在解码 + * @return 是否正在解码 + */ + fun isDecoding(): Boolean { + return opusManager?.isDecodeStream == true + } + + /** + * 检查是否正在编码 + * @return 是否正在编码 + */ + fun isEncoding(): Boolean { + return opusManager?.isEncodeStream == true + } + + /** + * 检查是否已初始化 + * @return 是否已初始化 + */ + fun isInitialized(): Boolean { + return isInitialized + } + + // ====================================================================================================== + // 私有方法 - 音频数据处理 + // ====================================================================================================== + + + /** + * 停止音频发送线程 + */ + private fun stopAudioSendThread() { + isAudioSending.set(false) + audioSendThread?.interrupt() + + try { + audioSendThread?.join(1000) // 等待最多1秒 + } catch (e: InterruptedException) { + Log.w(TAG, "等待音频发送线程停止时被中断") + } + + audioSendThread = null + audioSendQueue.clear() + + synchronized(audioBuffer) { + audioBuffer.clear() + } + + Log.d(TAG, "音频发送线程已停止") + } + + /** + * 启动音频队列处理线程 + */ + private fun startAudioQueueProcessing() { + audioQueueProcessorThread = Thread { + Log.d(TAG, "音频队列处理线程已启动") + + while (!Thread.currentThread().isInterrupted) { + try { + // 从队列中取出音频数据进行处理 + val audioData = audioDataQueue.take() // 阻塞等待 + + // 将音频数据写入Opus解码流 + opusManager?.writeAudioStream(audioData) + + } catch (e: InterruptedException) { + Log.d(TAG, "音频队列处理线程被中断") + break + } catch (e: Exception) { + Log.e(TAG, "音频队列处理异常: ${e.message}", e) + } + } + + Log.d(TAG, "音频队列处理线程已停止") + } + + audioQueueProcessorThread?.start() + } +} \ No newline at end of file