From 3ae1d8a76d0c6f540a7dcaa13a0cf10424abae05 Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Thu, 2 Jul 2026 02:22:46 +0200 Subject: [PATCH 01/10] fixing common and rtsp errors --- .../main/java/com/pedro/common/Extensions.kt | 19 +++++++++++++------ .../com/pedro/common/StreamBlockingQueue.kt | 3 ++- .../java/com/pedro/common/base/BaseSender.kt | 7 +++---- .../encoder/video/VideoEncoderHelper.java | 3 ++- .../util/AndroidMuxerRecordController.kt | 4 ++-- .../library/util/FlvMuxerRecordController.kt | 3 ++- .../util/Mpeg2TsMuxerRecordController.kt | 3 ++- .../com/pedro/rtsp/rtcp/SenderReportTcp.kt | 10 +++++----- .../java/com/pedro/rtsp/rtsp/RtspClient.kt | 4 ++-- .../pedro/rtsp/rtsp/commands/CommandParser.kt | 16 ++++++---------- .../rtsp/rtsp/commands/CommandsManager.kt | 5 +++-- .../com/pedro/rtsp/rtsp/CommandParserTest.kt | 4 ++-- 12 files changed, 44 insertions(+), 37 deletions(-) diff --git a/common/src/main/java/com/pedro/common/Extensions.kt b/common/src/main/java/com/pedro/common/Extensions.kt index b3ce33630e..499b97260b 100644 --- a/common/src/main/java/com/pedro/common/Extensions.kt +++ b/common/src/main/java/com/pedro/common/Extensions.kt @@ -31,6 +31,7 @@ import androidx.annotation.RequiresApi import com.pedro.common.frame.MediaFrame import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.withContext +import kotlinx.io.IOException import java.io.InputStream import java.io.OutputStream import java.io.UnsupportedEncodingException @@ -53,9 +54,9 @@ import kotlin.coroutines.Continuation @Suppress("DEPRECATION") fun MediaCodec.BufferInfo.isKeyframe(): Boolean { return if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.LOLLIPOP) { - this.flags == MediaCodec.BUFFER_FLAG_KEY_FRAME + this.flags and MediaCodec.BUFFER_FLAG_KEY_FRAME != 0 } else { - this.flags == MediaCodec.BUFFER_FLAG_SYNC_FRAME + this.flags and MediaCodec.BUFFER_FLAG_SYNC_FRAME != 0 } } @@ -74,6 +75,7 @@ fun ByteBuffer.toByteArray( } fun ByteBuffer.getStartCodeSize(): Int { + if (this.remaining() < 4) return 0 var startCodeSize = 0 if (this.get(0).toInt() == 0x00 && this.get(1).toInt() == 0x00 && this.get(2).toInt() == 0x00 && this.get(3).toInt() == 0x01) { @@ -236,29 +238,34 @@ fun BigInteger.toByteArray(length: Int): ByteArray { } } +@Throws(IOException::class) fun InputStream.readUntil(byteArray: ByteArray) { var bytesRead = 0 while (bytesRead < byteArray.size) { val result = read(byteArray, bytesRead, byteArray.size - bytesRead) - if (result != -1) bytesRead += result + if (result == -1) throw IOException("End of stream") + bytesRead += result } } +@Throws(IOException::class) fun InputStream.readUInt32(): Int { val data = ByteArray(4) - read(data) + readUntil(data) return data.toUInt32() } +@Throws(IOException::class) fun InputStream.readUInt24(): Int { val data = ByteArray(3) - read(data) + readUntil(data) return data.toUInt24() } +@Throws(IOException::class) fun InputStream.readUInt16(): Int { val data = ByteArray(2) - read(data) + readUntil(data) return data.toUInt16() } diff --git a/common/src/main/java/com/pedro/common/StreamBlockingQueue.kt b/common/src/main/java/com/pedro/common/StreamBlockingQueue.kt index 3bbbab6b01..adc8a083ff 100644 --- a/common/src/main/java/com/pedro/common/StreamBlockingQueue.kt +++ b/common/src/main/java/com/pedro/common/StreamBlockingQueue.kt @@ -55,7 +55,8 @@ class StreamBlockingQueue(var capacity: Int) { fun setCacheTime(cache: Long) { cacheTime = cache - cacheQueue = PriorityBlockingQueue((cache / 5).toInt()) { p0, p1 -> + if (cacheTime == 0L) return + cacheQueue = PriorityBlockingQueue(maxOf(1, (cache / 5).toInt())) { p0, p1 -> p0.info.timestamp.compare(p1.info.timestamp) } } diff --git a/common/src/main/java/com/pedro/common/base/BaseSender.kt b/common/src/main/java/com/pedro/common/base/BaseSender.kt index d85d743a30..c505b90fff 100644 --- a/common/src/main/java/com/pedro/common/base/BaseSender.kt +++ b/common/src/main/java/com/pedro/common/base/BaseSender.kt @@ -23,9 +23,8 @@ abstract class BaseSender( @Volatile protected var running = false - private var cacheSize = 400 - protected val queue = StreamBlockingQueue(cacheSize) + protected val queue = StreamBlockingQueue(400) protected val audioFramesSent = AtomicLong(0) protected val videoFramesSent = AtomicLong(0) @@ -92,7 +91,7 @@ abstract class BaseSender( @Throws(IllegalArgumentException::class) fun hasCongestion(percentUsed: Float = 20f): Boolean { - if (percentUsed < 0 || percentUsed > 100) throw IllegalArgumentException("the value must be in range 0 to 100") + if (percentUsed !in 0.0..100.0) throw IllegalArgumentException("the value must be in range 0 to 100") val size = queue.getSize().toFloat() val remaining = queue.remainingCapacity().toFloat() val capacity = size + remaining @@ -106,7 +105,7 @@ abstract class BaseSender( queue.capacity = newSize } - fun getCacheSize(): Int = cacheSize + fun getCacheSize(): Int = queue.capacity fun getItemsInCache(): Int = queue.getSize() diff --git a/encoder/src/main/java/com/pedro/encoder/video/VideoEncoderHelper.java b/encoder/src/main/java/com/pedro/encoder/video/VideoEncoderHelper.java index 9aeb23163d..b1c18441c4 100644 --- a/encoder/src/main/java/com/pedro/encoder/video/VideoEncoderHelper.java +++ b/encoder/src/main/java/com/pedro/encoder/video/VideoEncoderHelper.java @@ -21,6 +21,7 @@ import android.media.MediaCodec; import android.util.Pair; +import com.pedro.common.ExtensionsKt; import com.pedro.common.av1.Av1Parser; import com.pedro.common.av1.Obu; import com.pedro.common.av1.ObuType; @@ -123,7 +124,7 @@ public static List extractVpsSpsPpsFromH265(ByteBuffer csd0byteBuffe */ public static ByteBuffer extractObuSequence(ByteBuffer buffer, MediaCodec.BufferInfo bufferInfo) { //we can only extract info from keyframes - if (bufferInfo.flags != MediaCodec.BUFFER_FLAG_KEY_FRAME) return null; + if (!ExtensionsKt.isKeyframe(bufferInfo)) return null; byte[] av1Data = new byte[buffer.remaining()]; buffer.get(av1Data); Av1Parser av1Parser = new Av1Parser(); diff --git a/library/src/main/java/com/pedro/library/util/AndroidMuxerRecordController.kt b/library/src/main/java/com/pedro/library/util/AndroidMuxerRecordController.kt index 29fd858fe4..efdbaecde5 100644 --- a/library/src/main/java/com/pedro/library/util/AndroidMuxerRecordController.kt +++ b/library/src/main/java/com/pedro/library/util/AndroidMuxerRecordController.kt @@ -116,7 +116,7 @@ class AndroidMuxerRecordController : AsyncBaseRecordController() { when (frame.type) { MediaFrame.Type.VIDEO -> { if (recordStatus == RecordController.Status.STARTED && videoFormat != null && (audioFormat != null || tracks == RecordTracks.VIDEO)) { - if (frame.info.flags == MediaCodec.BUFFER_FLAG_KEY_FRAME || isKeyFrame(frame.data)) { + if (frame.info.isKeyFrame || isKeyFrame(frame.data)) { myRequestKeyFrame = null videoTrack = mediaMuxer?.addTrack(videoFormat!!) ?: -1 init() @@ -124,7 +124,7 @@ class AndroidMuxerRecordController : AsyncBaseRecordController() { myRequestKeyFrame?.onRequestKeyFrame() myRequestKeyFrame = null } - } else if (recordStatus == RecordController.Status.RESUMED && (frame.info.flags == MediaCodec.BUFFER_FLAG_KEY_FRAME + } else if (recordStatus == RecordController.Status.RESUMED && (frame.info.isKeyFrame || isKeyFrame(frame.data)) ) { recordStatus = RecordController.Status.RECORDING diff --git a/library/src/main/java/com/pedro/library/util/FlvMuxerRecordController.kt b/library/src/main/java/com/pedro/library/util/FlvMuxerRecordController.kt index d2c65049a8..319fd95227 100644 --- a/library/src/main/java/com/pedro/library/util/FlvMuxerRecordController.kt +++ b/library/src/main/java/com/pedro/library/util/FlvMuxerRecordController.kt @@ -6,6 +6,7 @@ import android.util.Log import com.pedro.common.AudioCodec import com.pedro.common.VideoCodec import com.pedro.common.frame.MediaFrame +import com.pedro.common.isKeyframe import com.pedro.common.toMediaCodecBufferInfo import com.pedro.common.toUInt24 import com.pedro.common.toUInt32 @@ -111,7 +112,7 @@ class FlvMuxerRecordController: AsyncBaseRecordController() { } private fun getVideoInfo(buffer: ByteBuffer, info: MediaCodec.BufferInfo) { - if (info.flags == MediaCodec.BUFFER_FLAG_KEY_FRAME || isKeyFrame(buffer)) { + if (info.isKeyframe() || isKeyFrame(buffer)) { if (!sendInfo) { when (videoPacket) { is H264Packet -> { diff --git a/library/src/main/java/com/pedro/library/util/Mpeg2TsMuxerRecordController.kt b/library/src/main/java/com/pedro/library/util/Mpeg2TsMuxerRecordController.kt index 075a5b23bc..4efda320e7 100644 --- a/library/src/main/java/com/pedro/library/util/Mpeg2TsMuxerRecordController.kt +++ b/library/src/main/java/com/pedro/library/util/Mpeg2TsMuxerRecordController.kt @@ -6,6 +6,7 @@ import android.util.Log import com.pedro.common.AudioCodec import com.pedro.common.VideoCodec import com.pedro.common.frame.MediaFrame +import com.pedro.common.isKeyframe import com.pedro.common.toMediaCodecBufferInfo import com.pedro.encoder.video.VideoEncoderHelper import com.pedro.library.base.recording.AsyncBaseRecordController @@ -184,7 +185,7 @@ class Mpeg2TsMuxerRecordController : AsyncBaseRecordController() { } private fun getVideoInfo(buffer: ByteBuffer, info: MediaCodec.BufferInfo) { - if (info.flags == MediaCodec.BUFFER_FLAG_KEY_FRAME || isKeyFrame(buffer)) { + if (info.isKeyframe() || isKeyFrame(buffer)) { if (!sendInfo) { when (getVideoCodec()) { VideoCodec.H264 -> { diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtcp/SenderReportTcp.kt b/rtsp/src/main/java/com/pedro/rtsp/rtcp/SenderReportTcp.kt index fb243cbd81..165b1ddb40 100644 --- a/rtsp/src/main/java/com/pedro/rtsp/rtcp/SenderReportTcp.kt +++ b/rtsp/src/main/java/com/pedro/rtsp/rtcp/SenderReportTcp.kt @@ -37,16 +37,16 @@ class SenderReportTcp(rtpTracks: RtpTracks): BaseSenderReport(rtpTracks) { @Throws(IOException::class) override suspend fun sendReport(buffer: ByteArray, rtpFrame: RtpFrame) { - sendReportTCP(buffer, rtpFrame.channelIdentifier) + tcpHeader[1] = (2 * rtpFrame.channelIdentifier + 1).toByte() + socket?.write(tcpHeader) + socket?.write(buffer, 0, RtpConstants.REPORT_PACKET_LENGTH) + socket?.flush() } override suspend fun close() {} @Throws(IOException::class) private suspend fun sendReportTCP(buffer: ByteArray, channelIdentifier: Int) { - tcpHeader[1] = (2 * channelIdentifier + 1).toByte() - socket?.write(tcpHeader) - socket?.write(buffer, 0, RtpConstants.REPORT_PACKET_LENGTH) - socket?.flush() + } } \ No newline at end of file diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt b/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt index 7467f48f06..8e1a1c4636 100644 --- a/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt +++ b/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt @@ -309,7 +309,7 @@ class RtspClient(private val connectChecker: ConnectChecker) { if (!commandsManager.videoDisabled) { socket.write(commandsManager.createSetup(commandsManager.rtpTracks.trackVideo)) socket.flush() - val setupVideoStatus = commandsManager.getResponse(socket, Method.SETUP).status + val setupVideoStatus = commandsManager.getResponse(socket, Method.SETUP, commandsManager.rtpTracks.trackVideo).status if (setupVideoStatus != 200) { onMainThread { connectChecker.onConnectionFailed("Error configure stream, setup video $setupVideoStatus") @@ -320,7 +320,7 @@ class RtspClient(private val connectChecker: ConnectChecker) { if (!commandsManager.audioDisabled) { socket.write(commandsManager.createSetup(commandsManager.rtpTracks.trackAudio)) socket.flush() - val setupAudioStatus = commandsManager.getResponse(socket, Method.SETUP).status + val setupAudioStatus = commandsManager.getResponse(socket, Method.SETUP, commandsManager.rtpTracks.trackAudio).status if (setupAudioStatus != 200) { onMainThread { connectChecker.onConnectionFailed("Error configure stream, setup audio $setupAudioStatus") diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandParser.kt b/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandParser.kt index 43586f6bdc..b7d0527347 100644 --- a/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandParser.kt +++ b/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandParser.kt @@ -29,17 +29,13 @@ class CommandParser { private const val TAG = "CommandParser" } - fun loadServerPorts(command: Command, protocol: Protocol, audioClientPorts: Array, - videoClientPorts: Array, audioServerPorts: Array, videoServerPorts: Array): Boolean { - var isAudio = true + fun loadServerPorts( + command: Command, protocol: Protocol, isAudio: Boolean, + audioClientPorts: Array, + videoClientPorts: Array, + audioServerPorts: Array, videoServerPorts: Array + ): Boolean { if (command.method == Method.SETUP && protocol == Protocol.UDP) { - val clientPattern = Pattern.compile("client_port=([0-9]+)-([0-9]+)") - val clientMatcher = clientPattern.matcher(command.text) - if (clientMatcher.find()) { - val port = (clientMatcher.group(1) ?: "-1").toInt() - isAudio = port == audioClientPorts[0] - } - val rtspPattern = Pattern.compile("server_port=([0-9]+)-([0-9]+)") val matcher = rtspPattern.matcher(command.text) if (matcher.find()) { diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandsManager.kt b/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandsManager.kt index 762147adb1..d6b2d66dac 100644 --- a/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandsManager.kt +++ b/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandsManager.kt @@ -261,7 +261,7 @@ open class CommandsManager { } @Throws(IOException::class) - suspend fun getResponse(socket: TcpStreamSocket, method: Method = Method.UNKNOWN): Command { + suspend fun getResponse(socket: TcpStreamSocket, method: Method = Method.UNKNOWN, track: Int? = null): Command { var response = "" var line: String? while (socket.readLine().also { line = it } != null) { @@ -276,7 +276,8 @@ open class CommandsManager { val command = commandParser.parseResponse(method, response) sessionId = commandParser.getSessionId(command) if (command.method == Method.SETUP && protocol == Protocol.UDP) { - commandParser.loadServerPorts(command, protocol, audioClientPorts, videoClientPorts, + val isAudio = track == rtpTracks.trackAudio + commandParser.loadServerPorts(command, protocol, isAudio, audioClientPorts, videoClientPorts, audioServerPorts, videoServerPorts) } command diff --git a/rtsp/src/test/java/com/pedro/rtsp/rtsp/CommandParserTest.kt b/rtsp/src/test/java/com/pedro/rtsp/rtsp/CommandParserTest.kt index c39f418fd9..25027889ce 100644 --- a/rtsp/src/test/java/com/pedro/rtsp/rtsp/CommandParserTest.kt +++ b/rtsp/src/test/java/com/pedro/rtsp/rtsp/CommandParserTest.kt @@ -56,8 +56,8 @@ class CommandParserTest { val commandAudio = Command(Method.SETUP, 1, 200, serverCommandAudio) val commandVideo = Command(Method.SETUP, 1, 200, serverCommandVideo) - commandParser.loadServerPorts(commandAudio, Protocol.UDP, audioClientPorts, videoClientPorts, audioServerPorts, videoServerPorts) - commandParser.loadServerPorts(commandVideo, Protocol.UDP, audioClientPorts, videoClientPorts, audioServerPorts, videoServerPorts) + commandParser.loadServerPorts(commandAudio, Protocol.UDP, true, audioClientPorts, videoClientPorts, audioServerPorts, videoServerPorts) + commandParser.loadServerPorts(commandVideo, Protocol.UDP, false, audioClientPorts, videoClientPorts, audioServerPorts, videoServerPorts) assertArrayEquals(audioServerPorts, expectedAudioServerPorts) assertArrayEquals(videoServerPorts, expectedVideoServerPorts) } From bcb3f6b18b3ad34e90f3cd9814b3e8601eed33fb Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Thu, 2 Jul 2026 15:25:48 +0200 Subject: [PATCH 02/10] fixing common errors --- .../main/java/com/pedro/common/BitBuffer.kt | 29 +++---------------- .../main/java/com/pedro/common/FpsUtils.kt | 2 -- .../java/com/pedro/common/base/BaseSender.kt | 2 +- .../java/com/pedro/common/nal/NalReader.kt | 5 +++- .../socket/java/TcpStreamSocketJavaBase.kt | 19 ++++++++---- .../common/socket/java/UdpStreamSocketJava.kt | 2 +- .../common/socket/ktor/TcpStreamSocketKtor.kt | 1 - .../socket/ktor/TcpStreamSocketKtorBase.kt | 3 +- .../common/socket/ktor/UdpStreamSocketKtor.kt | 1 - .../java/com/pedro/common/BitBufferTest.kt | 12 ++++---- .../java/com/pedro/common/NalReaderTest.kt | 18 ++++++++++++ .../pedro/rtmp/flv/audio/packet/AacPacket.kt | 16 +++++----- .../pedro/rtmp/flv/audio/packet/OpusPacket.kt | 12 ++++---- 13 files changed, 65 insertions(+), 57 deletions(-) diff --git a/common/src/main/java/com/pedro/common/BitBuffer.kt b/common/src/main/java/com/pedro/common/BitBuffer.kt index 217ffbe344..f779aca56a 100644 --- a/common/src/main/java/com/pedro/common/BitBuffer.kt +++ b/common/src/main/java/com/pedro/common/BitBuffer.kt @@ -71,34 +71,13 @@ class BitBuffer(val buffer: ByteBuffer) { } } - fun readUVLC(): Int { + fun readUVLC(): Long { var leadingZeros = 0 - var value = 0 - var currentIndex = bufferPosition / 8 - var currentBit = 7 - bufferPosition % 8 - - while (buffer[currentIndex].toInt() and (1 shl currentBit) == 0) { + while (!getBool()) { leadingZeros++ - if (currentBit == 0) { - currentIndex++ - currentBit = 7 - } else { - currentBit-- - } - } - - repeat((0 until leadingZeros + 1).count()) { - if (currentBit == 0) { - currentIndex++ - currentBit = 7 - } else { - currentBit-- - } - - value = (value shl 1) or ((buffer[currentIndex].toInt() ushr currentBit) and 1) + if (leadingZeros >= 32) return (1L shl 32) - 1 } - bufferPosition += 2 * leadingZeros + 1 - return value + return getLong(leadingZeros) + (1L shl leadingZeros) - 1 } fun resetPosition() { diff --git a/common/src/main/java/com/pedro/common/FpsUtils.kt b/common/src/main/java/com/pedro/common/FpsUtils.kt index deb7204de1..c4085ec289 100644 --- a/common/src/main/java/com/pedro/common/FpsUtils.kt +++ b/common/src/main/java/com/pedro/common/FpsUtils.kt @@ -7,9 +7,7 @@ import kotlin.math.abs object FpsUtils { fun adaptFpsRange(expectedFps: Int, fpsRanges: List): IntArray { - val expectedRange = intArrayOf(expectedFps, expectedFps) if (fpsRanges.isEmpty()) throw IllegalArgumentException("fpsRanges is empty") - if (fpsRanges.contains(expectedRange)) return expectedRange val exactFps = fpsRanges.filter { it[1] == expectedFps } if (exactFps.isNotEmpty()) return exactFps.sortedBy { abs(expectedFps - it[0]) }[0] val higherFps = fpsRanges.filter { it[1] > expectedFps } diff --git a/common/src/main/java/com/pedro/common/base/BaseSender.kt b/common/src/main/java/com/pedro/common/base/BaseSender.kt index c505b90fff..99d3b1a955 100644 --- a/common/src/main/java/com/pedro/common/base/BaseSender.kt +++ b/common/src/main/java/com/pedro/common/base/BaseSender.kt @@ -99,7 +99,7 @@ abstract class BaseSender( } fun resizeCache(newSize: Int) { - if (newSize < queue.getSize() - queue.remainingCapacity()) { + if (newSize < queue.getSize()) { throw RuntimeException("Can't fit current cache inside new cache size") } queue.capacity = newSize diff --git a/common/src/main/java/com/pedro/common/nal/NalReader.kt b/common/src/main/java/com/pedro/common/nal/NalReader.kt index c318d01e1f..2722291e90 100644 --- a/common/src/main/java/com/pedro/common/nal/NalReader.kt +++ b/common/src/main/java/com/pedro/common/nal/NalReader.kt @@ -30,6 +30,7 @@ object NalReader { val start = offset + buffer.position() val limit = offset + buffer.limit() var payloadStart = -1 + var nalFound = false var i = start while (i < limit - 2) { @@ -40,6 +41,7 @@ object NalReader { duplicate.position(payloadStart - offset) duplicate.limit(previousPayloadEnd - offset) val nal = duplicate.slice() + nalFound = true if (shouldKeepNal(nal, codec, shouldDiscardVideoInfo)) units.add(nal) } payloadStart = i + 3 @@ -53,9 +55,10 @@ object NalReader { duplicate.position(payloadStart - offset) duplicate.limit(limit - offset) val nal = duplicate.slice() + nalFound = true if (shouldKeepNal(nal, codec, shouldDiscardVideoInfo)) units.add(nal) } - if (units.isEmpty()) units.add(buffer) + if (!nalFound) units.add(buffer) return units } diff --git a/common/src/main/java/com/pedro/common/socket/java/TcpStreamSocketJavaBase.kt b/common/src/main/java/com/pedro/common/socket/java/TcpStreamSocketJavaBase.kt index 0c82434e2a..10bff3975e 100644 --- a/common/src/main/java/com/pedro/common/socket/java/TcpStreamSocketJavaBase.kt +++ b/common/src/main/java/com/pedro/common/socket/java/TcpStreamSocketJavaBase.kt @@ -2,10 +2,8 @@ package com.pedro.common.socket.java import com.pedro.common.readUntil import com.pedro.common.socket.base.TcpStreamSocket -import java.io.BufferedReader import java.io.ByteArrayInputStream import java.io.ByteArrayOutputStream -import java.io.InputStreamReader import java.net.Socket abstract class TcpStreamSocketJavaBase: TcpStreamSocket() { @@ -13,13 +11,11 @@ abstract class TcpStreamSocketJavaBase: TcpStreamSocket() { private var socket = Socket() private var input = ByteArrayInputStream(byteArrayOf()).buffered() private var output = ByteArrayOutputStream().buffered() - private var reader = InputStreamReader(input).buffered() abstract fun onConnectSocket(timeout: Long): Socket override suspend fun connect() { socket = onConnectSocket(timeout) - reader = BufferedReader(InputStreamReader(socket.getInputStream())) output = socket.getOutputStream().buffered() input = socket.getInputStream().buffered() } @@ -62,7 +58,20 @@ abstract class TcpStreamSocketJavaBase: TcpStreamSocket() { return data } - override suspend fun readLine(): String? = reader.readLine() + override suspend fun readLine(): String? { + var value = input.read() + if (value == -1) return null + val line = ByteArrayOutputStream() + while (value != -1 && value != '\n'.code) { + line.write(value) + value = input.read() + } + var bytes = line.toByteArray() + if (bytes.isNotEmpty() && bytes.last() == '\r'.code.toByte()) { + bytes = bytes.copyOf(bytes.size - 1) + } + return String(bytes) + } override fun isConnected(): Boolean = socket.isConnected diff --git a/common/src/main/java/com/pedro/common/socket/java/UdpStreamSocketJava.kt b/common/src/main/java/com/pedro/common/socket/java/UdpStreamSocketJava.kt index 152225a015..17c97743d0 100644 --- a/common/src/main/java/com/pedro/common/socket/java/UdpStreamSocketJava.kt +++ b/common/src/main/java/com/pedro/common/socket/java/UdpStreamSocketJava.kt @@ -19,7 +19,7 @@ class UdpStreamSocketJava( ): UdpStreamSocket() { private var socket: DatagramSocket? = null - private val packetSize = receiveSize ?: SocketOptions.SO_RCVBUF + private val packetSize = receiveSize ?: 65536 private var remoteHost: String? = null private var remotePort: Int? = null diff --git a/common/src/main/java/com/pedro/common/socket/ktor/TcpStreamSocketKtor.kt b/common/src/main/java/com/pedro/common/socket/ktor/TcpStreamSocketKtor.kt index 1504129077..1a55a2071a 100644 --- a/common/src/main/java/com/pedro/common/socket/ktor/TcpStreamSocketKtor.kt +++ b/common/src/main/java/com/pedro/common/socket/ktor/TcpStreamSocketKtor.kt @@ -18,7 +18,6 @@ class TcpStreamSocketKtor( ): TcpStreamSocketKtorBase(host, port) { override suspend fun onConnectSocket(timeout: Long, error: (Throwable) -> Unit): ReadWriteSocket { - selectorManager = SelectorManager(Dispatchers.IO) val builder = aSocket(selectorManager).tcp().connect( remoteAddress = InetSocketAddress(host, port), configure = { if (!secured) socketTimeout = timeout } diff --git a/common/src/main/java/com/pedro/common/socket/ktor/TcpStreamSocketKtorBase.kt b/common/src/main/java/com/pedro/common/socket/ktor/TcpStreamSocketKtorBase.kt index c1b8583bd5..50c82dc339 100644 --- a/common/src/main/java/com/pedro/common/socket/ktor/TcpStreamSocketKtorBase.kt +++ b/common/src/main/java/com/pedro/common/socket/ktor/TcpStreamSocketKtorBase.kt @@ -31,6 +31,7 @@ abstract class TcpStreamSocketKtorBase( abstract suspend fun onConnectSocket(timeout: Long, error: (Throwable) -> Unit): ReadWriteSocket override suspend fun connect() { + selectorManager = SelectorManager(Dispatchers.IO) val socket = onConnectSocket(timeout) { runBlocking { close() } } @@ -83,7 +84,7 @@ abstract class TcpStreamSocketKtorBase( override suspend fun readLine(): String? = input?.readLine() - override fun isConnected(): Boolean = socket?.isClosed != true + override fun isConnected(): Boolean = socket?.let { !it.isClosed } ?: false override fun isReachable(): Boolean = address?.isReachable(timeout.toInt()) ?: false } \ No newline at end of file diff --git a/common/src/main/java/com/pedro/common/socket/ktor/UdpStreamSocketKtor.kt b/common/src/main/java/com/pedro/common/socket/ktor/UdpStreamSocketKtor.kt index 0c07478999..b056615db2 100644 --- a/common/src/main/java/com/pedro/common/socket/ktor/UdpStreamSocketKtor.kt +++ b/common/src/main/java/com/pedro/common/socket/ktor/UdpStreamSocketKtor.kt @@ -35,7 +35,6 @@ class UdpStreamSocketKtor( private var myAddress: InetAddress? = null override suspend fun connect() { - selectorManager = SelectorManager(Dispatchers.IO) val builder = aSocket(selectorManager).udp() val localAddress = if (sourcePort == null) null else { val localAddress = InetSocketAddress(sourceHost ?: "0.0.0.0", sourcePort) diff --git a/common/src/test/java/com/pedro/common/BitBufferTest.kt b/common/src/test/java/com/pedro/common/BitBufferTest.kt index 5d8533697f..4c52b84347 100644 --- a/common/src/test/java/com/pedro/common/BitBufferTest.kt +++ b/common/src/test/java/com/pedro/common/BitBufferTest.kt @@ -182,14 +182,14 @@ class BitBufferTest { @Test fun `GIVEN a uvlc with no leading zeros WHEN readUVLC THEN return 0 and consume one bit`() { val bitBuffer = bitBufferOf(0x80) // 1000 0000 - assertEquals(0, bitBuffer.readUVLC()) + assertEquals(0L, bitBuffer.readUVLC()) assertEquals(1, bitBuffer.bufferPosition) } @Test fun `GIVEN a uvlc with leading zeros WHEN readUVLC THEN decode value and consume 2n+1 bits`() { val bitBuffer = bitBufferOf(0x60) // 0110 0000 - assertEquals(2, bitBuffer.readUVLC()) + assertEquals(2L, bitBuffer.readUVLC()) assertEquals(3, bitBuffer.bufferPosition) } @@ -197,7 +197,7 @@ class BitBufferTest { fun `GIVEN a uvlc whose value crosses a byte boundary WHEN readUVLC THEN decode it`() { // 0000 1 101 | 10... -> 4 leading zeros, terminator, value bits span into the second byte val bitBuffer = bitBufferOf(0x0D, 0x80) - assertEquals(22, bitBuffer.readUVLC()) + assertEquals(26L, bitBuffer.readUVLC()) assertEquals(9, bitBuffer.bufferPosition) } @@ -205,16 +205,16 @@ class BitBufferTest { fun `GIVEN a uvlc whose leading zeros cross a byte boundary WHEN readUVLC THEN decode it`() { // 8 leading zeros (whole first byte), terminator on the second byte, value spans into the third val bitBuffer = bitBufferOf(0x00, 0xD5, 0x40) - assertEquals(341, bitBuffer.readUVLC()) + assertEquals(425L, bitBuffer.readUVLC()) assertEquals(17, bitBuffer.bufferPosition) } @Test fun `GIVEN an unaligned position WHEN readUVLC THEN decode from that bit offset`() { - // skip 4 bits, then 0 1 11 -> 1 leading zero, terminator, value = 3 + // skip 4 bits, then 0 1 1 -> 1 leading zero, terminator, value bit 1 -> 1 + (1 shl 1) - 1 = 2 val bitBuffer = bitBufferOf(0x07) bitBuffer.skip(4) - assertEquals(3, bitBuffer.readUVLC()) + assertEquals(2L, bitBuffer.readUVLC()) assertEquals(7, bitBuffer.bufferPosition) } diff --git a/common/src/test/java/com/pedro/common/NalReaderTest.kt b/common/src/test/java/com/pedro/common/NalReaderTest.kt index c2af56c934..f68953b258 100644 --- a/common/src/test/java/com/pedro/common/NalReaderTest.kt +++ b/common/src/test/java/com/pedro/common/NalReaderTest.kt @@ -45,4 +45,22 @@ class NalReaderTest { assertEquals(10_000, it.capacity()) } } + + @Test + fun testReturnEmptyWhenEveryNalIsDiscarded() { + val sei = header.plus(0x06).plus(ByteArray(10) { 0x0f }) + val aud = header.plus(0x09).plus(ByteArray(10) { 0x0f }) + val buffer = ByteBuffer.wrap(sei.plus(aud)) + val nals = NalReader.extractNals(buffer, VideoCodec.H264, true) + assertEquals(0, nals.size) + } + + @Test + fun testFallbackToWholeBufferWhenNoStartCodeFound() { + val raw = ByteArray(10_000) { 0x0f } + val buffer = ByteBuffer.wrap(raw) + val nals = NalReader.extractNals(buffer, VideoCodec.H264, true) + assertEquals(1, nals.size) + assertEquals(10_000, nals[0].capacity()) + } } \ No newline at end of file diff --git a/rtmp/src/main/java/com/pedro/rtmp/flv/audio/packet/AacPacket.kt b/rtmp/src/main/java/com/pedro/rtmp/flv/audio/packet/AacPacket.kt index 25be65006f..140c9b1e94 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/flv/audio/packet/AacPacket.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/flv/audio/packet/AacPacket.kt @@ -60,6 +60,8 @@ class AacPacket: BasePacket() { callback: suspend (FlvPacket) -> Unit ) { val fixedBuffer = mediaFrame.data.removeInfo(mediaFrame.info) + val ts = mediaFrame.info.timestamp / 1000 + //header is 2 bytes length //4 bits sound format, 2 bits sound rate, 1 bit sound size, 1 bit sound type //8 bits sound data (always 10 because we are using aac) @@ -71,20 +73,20 @@ class AacPacket: BasePacket() { 5500 -> AudioSoundRate.SR_5_5K else -> AudioSoundRate.SR_44_1K } + header[0] = type or (audioSize.value shl 1).toByte() or (soundRate.value shl 2).toByte() or (AudioFormat.AAC.value shl 4).toByte() - val buffer: ByteArray if (!configSend) { header[1] = Type.SEQUENCE.mark val config = AacAudioSpecificConfig(objectType, sampleRate, if (isStereo) 2 else 1) - buffer = ByteArray(header.size).plus(config.calculate()) + val buffer = ByteArray(header.size).plus(config.calculate()) configSend = true - } else { - header[1] = Type.RAW.mark - buffer = ByteArray(fixedBuffer.remaining() + header.size) - fixedBuffer.get(buffer, header.size, fixedBuffer.remaining()) + System.arraycopy(header, 0, buffer, 0, header.size) + callback(FlvPacket(buffer, ts, buffer.size, FlvType.AUDIO)) } + header[1] = Type.RAW.mark + val buffer = ByteArray(fixedBuffer.remaining() + header.size) + fixedBuffer.get(buffer, header.size, fixedBuffer.remaining()) System.arraycopy(header, 0, buffer, 0, header.size) - val ts = mediaFrame.info.timestamp / 1000 callback(FlvPacket(buffer, ts, buffer.size, FlvType.AUDIO)) } diff --git a/rtmp/src/main/java/com/pedro/rtmp/flv/audio/packet/OpusPacket.kt b/rtmp/src/main/java/com/pedro/rtmp/flv/audio/packet/OpusPacket.kt index 04fce2f949..3f3099b391 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/flv/audio/packet/OpusPacket.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/flv/audio/packet/OpusPacket.kt @@ -58,18 +58,18 @@ class OpusPacket: BasePacket() { header[3] = (codec shr 8).toByte() header[4] = codec.toByte() - val buffer: ByteArray if (!configSend) { header[0] = ((AudioFormat.EX_HEADER.value shl 4) or (AudioFourCCPacketType.SEQUENCE_START.value and 0x0F)).toByte() val config = OpusAudioSpecificConfig(sampleRate, if (isStereo) 2 else 1) - buffer = ByteArray(config.size + header.size) + val buffer = ByteArray(config.size + header.size) config.write(buffer, header.size) configSend = true - } else { - header[0] = ((AudioFormat.EX_HEADER.value shl 4) or (AudioFourCCPacketType.CODED_FRAMES.value and 0x0F)).toByte() - buffer = ByteArray(fixedBuffer.remaining() + header.size) - fixedBuffer.get(buffer, header.size, fixedBuffer.remaining()) + System.arraycopy(header, 0, buffer, 0, header.size) + callback(FlvPacket(buffer, ts, buffer.size, FlvType.AUDIO)) } + header[0] = ((AudioFormat.EX_HEADER.value shl 4) or (AudioFourCCPacketType.CODED_FRAMES.value and 0x0F)).toByte() + val buffer = ByteArray(fixedBuffer.remaining() + header.size) + fixedBuffer.get(buffer, header.size, fixedBuffer.remaining()) System.arraycopy(header, 0, buffer, 0, header.size) callback(FlvPacket(buffer, ts, buffer.size, FlvType.AUDIO)) } From c9fefe3fe816ff23cb27f60285c202da3d271fef Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Thu, 2 Jul 2026 17:32:53 +0200 Subject: [PATCH 03/10] fixing encoder errors --- .../java/com/pedro/encoder/BaseEncoder.java | 4 +-- .../com/pedro/encoder/audio/AudioEncoder.java | 2 +- .../input/audio/MicrophoneManager.java | 35 +++++++++++++++---- .../encoder/input/decoder/AudioDecoder.java | 24 ++----------- .../encoder/input/decoder/BaseDecoder.java | 9 ++++- .../encoder/input/decoder/VideoDecoder.java | 1 + .../input/gl/render/BaseRenderOffScreen.java | 3 -- .../encoder/input/gl/render/CameraRender.java | 2 ++ .../filters/AndroidViewFilterRender.java | 2 +- .../java/com/pedro/encoder/utils/PCMUtil.java | 19 +++++----- .../java/com/pedro/encoder/PCMUtilTest.kt | 29 +++++++++++++++ .../pedro/extrasources/Media3AudioSource.kt | 10 +++--- .../pedro/extrasources/Media3VideoSource.kt | 10 +++--- .../pedro/library/view/GlStreamInterface.kt | 12 +++++-- .../java/com/pedro/library/view/OpenGlView.kt | 11 ++++-- udp/src/main/java/com/pedro/udp/UdpClient.kt | 1 + 16 files changed, 115 insertions(+), 59 deletions(-) create mode 100644 encoder/src/test/java/com/pedro/encoder/PCMUtilTest.kt diff --git a/encoder/src/main/java/com/pedro/encoder/BaseEncoder.java b/encoder/src/main/java/com/pedro/encoder/BaseEncoder.java index 49f9500450..c4d4ebbfc5 100644 --- a/encoder/src/main/java/com/pedro/encoder/BaseEncoder.java +++ b/encoder/src/main/java/com/pedro/encoder/BaseEncoder.java @@ -107,6 +107,7 @@ protected void setCallback() { } private void initCodec() { + running = true; if (!type.equals(CodecUtil.G711_MIME)) codec.start(); if (Build.VERSION.SDK_INT < Build.VERSION_CODES.M || type.equals(CodecUtil.G711_MIME)) { executorService = Executors.newSingleThreadExecutor(); @@ -121,7 +122,6 @@ private void initCodec() { } }); } - running = true; } public abstract boolean reset(); @@ -237,7 +237,7 @@ private void processInput(@NonNull ByteBuffer byteBuffer, @NonNull MediaCodec me Frame frame = getInputFrame(); while (frame == null) frame = getInputFrame(); byteBuffer.clear(); - int size = Math.max(0, Math.min(frame.getSize(), byteBuffer.remaining()) - frame.getOffset()); + int size = Math.max(0, Math.min(frame.getSize(), byteBuffer.remaining())); byteBuffer.put(frame.getBuffer(), frame.getOffset(), size); long pts = calculatePts(frame, presentTimeUs); mediaCodec.queueInputBuffer(inBufferIndex, 0, size, pts, 0); diff --git a/encoder/src/main/java/com/pedro/encoder/audio/AudioEncoder.java b/encoder/src/main/java/com/pedro/encoder/audio/AudioEncoder.java index a3ea7e30dd..54ac6186b5 100644 --- a/encoder/src/main/java/com/pedro/encoder/audio/AudioEncoder.java +++ b/encoder/src/main/java/com/pedro/encoder/audio/AudioEncoder.java @@ -150,7 +150,7 @@ protected long calculatePts(Frame frame, long presentTimeUs) { } else { if (tsBuffer == 0) tsBuffer = clockPts; int channels = isStereo ? 2 : 1; - tsBuffer += (long)((double)frame.getSize() / (sampleRate * channels * 2L)) * 1_000_000L; + tsBuffer += (long)((double)frame.getSize() / (sampleRate * channels * 2L) * 1_000_000L); if (clockPts - tsBuffer > 500_000) tsBuffer = clockPts; pts = tsBuffer; } diff --git a/encoder/src/main/java/com/pedro/encoder/input/audio/MicrophoneManager.java b/encoder/src/main/java/com/pedro/encoder/input/audio/MicrophoneManager.java index df15adb4a2..a3b5739db8 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/audio/MicrophoneManager.java +++ b/encoder/src/main/java/com/pedro/encoder/input/audio/MicrophoneManager.java @@ -34,6 +34,8 @@ import com.pedro.encoder.Frame; import com.pedro.encoder.audio.AudioEncoder; +import java.util.Arrays; + /** * Created by pedro on 19/01/17. */ @@ -49,7 +51,7 @@ public class MicrophoneManager { protected byte[] pcmBufferDevice = new byte[AudioEncoder.inputSize]; protected byte[] pcmBufferMix = new byte[AudioEncoder.inputSize]; protected byte[] pcmBufferMuted = new byte[AudioEncoder.inputSize]; - protected boolean running = false; + protected volatile boolean running = false; private boolean created = false; //default parameters for microphone private int sampleRate = 32000; //hz @@ -57,6 +59,7 @@ public class MicrophoneManager { private int channel = AudioFormat.CHANNEL_IN_STEREO; protected boolean muted = false; private AudioPostProcessEffect audioPostProcessEffect; + private AudioPostProcessEffect audioPostProcessEffectDevice; protected HandlerThread handlerThread; protected CustomAudioEffect customAudioEffect = new NoAudioEffect(); private Mode mode = Mode.MICROPHONE; @@ -149,9 +152,9 @@ public boolean createInternalMicrophone(AudioPlaybackCaptureConfiguration config .build()) .setBufferSizeInBytes(AudioEncoder.inputSize * 5) .build(); - audioPostProcessEffect = new AudioPostProcessEffect(audioRecordDevice.getAudioSessionId()); - if (echoCanceler) audioPostProcessEffect.enableEchoCanceler(); - if (noiseSuppressor) audioPostProcessEffect.enableNoiseSuppressor(); + audioPostProcessEffectDevice = new AudioPostProcessEffect(audioRecordDevice.getAudioSessionId()); + if (echoCanceler) audioPostProcessEffectDevice.enableEchoCanceler(); + if (noiseSuppressor) audioPostProcessEffectDevice.enableNoiseSuppressor(); String chl = (isStereo) ? "Stereo" : "Mono"; if (audioRecordDevice.getState() != AudioRecord.STATE_INITIALIZED) { throw new IllegalArgumentException("Some parameters specified are not valid"); @@ -258,24 +261,33 @@ protected Frame read() { long timeStamp = TimeUtils.getCurrentTimeMicro(); switch (mode) { case MICROPHONE -> { + AudioRecord audioRecord = this.audioRecord; + if (audioRecord == null) return null; int size = audioRecord.read(pcmBuffer, 0, pcmBuffer.length); if (size < 0) { Log.e(TAG, "read error: " + size); return null; } audioUtils.applyVolume(pcmBuffer, microphoneVolume); - return new Frame(muted ? pcmBufferMuted : customAudioEffect.process(pcmBuffer), 0, size, timeStamp); + byte[] data = muted ? pcmBufferMuted : Arrays.copyOf(customAudioEffect.process(pcmBuffer), size); + return new Frame(data, 0, size, timeStamp); } case INTERNAL -> { + AudioRecord audioRecordDevice = this.audioRecordDevice; + if (audioRecordDevice == null) return null; int size = audioRecordDevice.read(pcmBufferDevice, 0, pcmBufferDevice.length); if (size < 0) { Log.e(TAG, "read error: " + size); return null; } audioUtils.applyVolume(pcmBufferDevice, internalVolume); - return new Frame(muted ? pcmBufferMuted : customAudioEffect.process(pcmBufferDevice), 0, size, timeStamp); + byte[] data = muted ? pcmBufferMuted : Arrays.copyOf(customAudioEffect.process(pcmBufferDevice), size); + return new Frame(data, 0, size, timeStamp); } case MIX -> { + AudioRecord audioRecord = this.audioRecord; + AudioRecord audioRecordDevice = this.audioRecordDevice; + if (audioRecord == null || audioRecordDevice == null) return null; int size = audioRecord.read(pcmBuffer, 0, pcmBuffer.length); if (size < 0) { Log.e(TAG, "read error: " + size); @@ -287,7 +299,8 @@ protected Frame read() { return null; } audioUtils.applyVolumeAndMix(pcmBuffer, microphoneVolume, pcmBufferDevice, internalVolume, pcmBufferMix); - return new Frame(muted ? pcmBufferMuted : customAudioEffect.process(pcmBufferMix), 0, size, timeStamp); + byte[] data = muted ? pcmBufferMuted : Arrays.copyOf(customAudioEffect.process(pcmBufferMix), size); + return new Frame(data, 0, size, timeStamp); } default -> { return null; } } @@ -305,6 +318,11 @@ public synchronized void stop() { } else { handlerThread.quit(); } + try { + handlerThread.join(200); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } } if (audioRecord != null) { audioRecord.setRecordPositionUpdateListener(null); @@ -321,6 +339,9 @@ public synchronized void stop() { if (audioPostProcessEffect != null) { audioPostProcessEffect.release(); } + if (audioPostProcessEffectDevice != null) { + audioPostProcessEffectDevice.release(); + } Log.i(TAG, "Microphone stopped"); } diff --git a/encoder/src/main/java/com/pedro/encoder/input/decoder/AudioDecoder.java b/encoder/src/main/java/com/pedro/encoder/input/decoder/AudioDecoder.java index b1988e6c50..c1c47eba06 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/decoder/AudioDecoder.java +++ b/encoder/src/main/java/com/pedro/encoder/input/decoder/AudioDecoder.java @@ -34,9 +34,6 @@ public class AudioDecoder extends BaseDecoder { private int sampleRate; private boolean isStereo; private int channels = 1; - private int size = 2048; - private byte[] pcmBuffer = new byte[size]; - private byte[] pcmBufferMuted = new byte[11]; private boolean muted = false; public AudioDecoder(GetMicrophoneData getMicrophoneData, @@ -51,7 +48,6 @@ public AudioDecoder(GetMicrophoneData getMicrophoneData, @Override protected boolean extract(Extractor extractor) { try { - size = 2048; mime = extractor.selectTrack(MediaFrame.Type.AUDIO); AudioInfo audioInfo = extractor.getAudioInfo(); mediaFormat = extractor.getFormat(); @@ -59,7 +55,6 @@ protected boolean extract(Extractor extractor) { isStereo = channels >= 2; this.sampleRate = audioInfo.getSampleRate(); this.duration = audioInfo.getDuration(); - fixBuffer(); return true; } catch (Exception e) { mime = ""; @@ -67,11 +62,6 @@ protected boolean extract(Extractor extractor) { } } - private void fixBuffer() { - if (channels >= 2) size *= channels; - pcmBuffer = new byte[size]; - } - public boolean prepareAudio() { return prepare(null); } @@ -81,15 +71,11 @@ protected boolean decodeOutput(ByteBuffer outputBuffer, long timeStamp) { //This buffer is PCM data GetMicrophoneData getMicrophoneData = this.getMicrophoneData; if (getMicrophoneData == null) return false; + byte[] pcmBuffer = new byte[outputBuffer.remaining()]; if (muted) { - outputBuffer.get(pcmBufferMuted, 0, - Math.min(outputBuffer.remaining(), pcmBufferMuted.length)); - getMicrophoneData.inputPCMData(new Frame(pcmBufferMuted, 0, pcmBufferMuted.length, timeStamp)); + getMicrophoneData.inputPCMData(new Frame(pcmBuffer, 0, pcmBuffer.length, timeStamp)); } else { - if (pcmBuffer.length < outputBuffer.remaining()) { - pcmBuffer = new byte[outputBuffer.remaining()]; - } - outputBuffer.get(pcmBuffer, 0, Math.min(outputBuffer.remaining(), pcmBuffer.length)); + outputBuffer.get(pcmBuffer, 0, outputBuffer.remaining()); if (channels > 2) { //downgrade to stereo byte[] bufferStereo = PCMUtil.pcmToStereo(pcmBuffer, channels); getMicrophoneData.inputPCMData(new Frame(bufferStereo, 0, bufferStereo.length, timeStamp)); @@ -125,10 +111,6 @@ public boolean isStereo() { return isStereo; } - public int getSize() { - return size; - } - public void setGetMicrophoneData(GetMicrophoneData getMicrophoneData) { if (running) return; this.getMicrophoneData = getMicrophoneData; diff --git a/encoder/src/main/java/com/pedro/encoder/input/decoder/BaseDecoder.java b/encoder/src/main/java/com/pedro/encoder/input/decoder/BaseDecoder.java index cb2d4cfc46..dd6c405b17 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/decoder/BaseDecoder.java +++ b/encoder/src/main/java/com/pedro/encoder/input/decoder/BaseDecoder.java @@ -33,6 +33,7 @@ import java.nio.ByteBuffer; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Semaphore; import java.util.concurrent.atomic.AtomicBoolean; public abstract class BaseDecoder { @@ -49,6 +50,7 @@ public abstract class BaseDecoder { protected long duration; protected final Object sync = new Object(); protected AtomicBoolean pause = new AtomicBoolean(false); + protected final Semaphore semaphore = new Semaphore(0); protected volatile boolean looped = false; private final DecoderInterface decoderInterface; private Extractor extractor = new AndroidExtractor(); @@ -217,7 +219,12 @@ private void decode() { boolean shouldFinish = false; long sleepTime = 0; while (running) { - if (pause.get()) continue; + if (pause.get()) { + try { + semaphore.acquire(); + } catch (InterruptedException ignored) {} + continue; + } if (shouldFinish) { finished(); break; diff --git a/encoder/src/main/java/com/pedro/encoder/input/decoder/VideoDecoder.java b/encoder/src/main/java/com/pedro/encoder/input/decoder/VideoDecoder.java index f64e9de068..b83d9b90dc 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/decoder/VideoDecoder.java +++ b/encoder/src/main/java/com/pedro/encoder/input/decoder/VideoDecoder.java @@ -101,6 +101,7 @@ public void pauseRender() { public void resumeRender() { synchronized (sync) { pause.set(false); + semaphore.release(); } } } diff --git a/encoder/src/main/java/com/pedro/encoder/input/gl/render/BaseRenderOffScreen.java b/encoder/src/main/java/com/pedro/encoder/input/gl/render/BaseRenderOffScreen.java index 92e53afec5..be84bc906d 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/gl/render/BaseRenderOffScreen.java +++ b/encoder/src/main/java/com/pedro/encoder/input/gl/render/BaseRenderOffScreen.java @@ -45,9 +45,6 @@ public abstract class BaseRenderOffScreen { protected RenderHandler renderHandler = new RenderHandler(); - protected int width; - protected int height; - public abstract void initGl(int width, int height, Context context, int previewWidth, int previewHeight); diff --git a/encoder/src/main/java/com/pedro/encoder/input/gl/render/CameraRender.java b/encoder/src/main/java/com/pedro/encoder/input/gl/render/CameraRender.java index a5b8cebba3..226cf41659 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/gl/render/CameraRender.java +++ b/encoder/src/main/java/com/pedro/encoder/input/gl/render/CameraRender.java @@ -52,6 +52,8 @@ public class CameraRender extends BaseRenderOffScreen { private SurfaceTexture surfaceTexture = new SurfaceTexture(textureID[0]); private Surface surface = new Surface(surfaceTexture); + private int width; + private int height; public CameraRender() { Matrix.setIdentityM(MVPMatrix, 0); diff --git a/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/AndroidViewFilterRender.java b/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/AndroidViewFilterRender.java index 4bfaa07928..83e73bca22 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/AndroidViewFilterRender.java +++ b/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/AndroidViewFilterRender.java @@ -73,7 +73,7 @@ public class AndroidViewFilterRender extends BaseFilterRender { private SurfaceTexture surfaceTexture, surfaceTexture2; private Surface surface, surface2; private final Handler mainHandler; - private boolean running = false; + private volatile boolean running = false; private ExecutorService thread = null; private boolean hardwareMode = true; private final AndroidViewSprite sprite; diff --git a/encoder/src/main/java/com/pedro/encoder/utils/PCMUtil.java b/encoder/src/main/java/com/pedro/encoder/utils/PCMUtil.java index ed0c84cf3f..f4e281bb84 100644 --- a/encoder/src/main/java/com/pedro/encoder/utils/PCMUtil.java +++ b/encoder/src/main/java/com/pedro/encoder/utils/PCMUtil.java @@ -56,22 +56,23 @@ public static byte[] mixPCM(byte[] pcm1, byte[] pcm2) { return pcmL; } - private static final byte[] pcmBufferStereo = new byte[4096]; - /** * Experimental method to downgrade pcm with 3 channels or more to stereo. + * Keeps the first two channels (16 bits little endian per sample) and drops the rest. * * @return pcm buffer in stereo (2 channels) */ public static byte[] pcmToStereo(byte[] pcm, int channels) { + int frameSize = channels * 2; + int frames = pcm.length / frameSize; + byte[] pcmBufferStereo = new byte[frames * 4]; int cont = 0; - for (int i = 0; i < pcm.length; i += channels) { - byte channel1 = pcm[i]; - byte channel2 = pcm[i + 1]; - - pcmBufferStereo[cont] = channel1; - pcmBufferStereo[cont + 1] = channel2; - cont += 2; + for (int i = 0; i + frameSize <= pcm.length; i += frameSize) { + pcmBufferStereo[cont] = pcm[i]; + pcmBufferStereo[cont + 1] = pcm[i + 1]; + pcmBufferStereo[cont + 2] = pcm[i + 2]; + pcmBufferStereo[cont + 3] = pcm[i + 3]; + cont += 4; } return pcmBufferStereo; } diff --git a/encoder/src/test/java/com/pedro/encoder/PCMUtilTest.kt b/encoder/src/test/java/com/pedro/encoder/PCMUtilTest.kt new file mode 100644 index 0000000000..b43d721bb3 --- /dev/null +++ b/encoder/src/test/java/com/pedro/encoder/PCMUtilTest.kt @@ -0,0 +1,29 @@ +package com.pedro.encoder + +import com.pedro.encoder.utils.PCMUtil +import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertEquals +import org.junit.Test + +class PCMUtilTest { + + @Test + fun `GIVEN a 3 channel pcm WHEN pcmToStereo THEN keep 16 bit samples of channels 0 and 1`() { + // 2 frames, 3 channels, 16 bits per sample -> 6 bytes per frame + // frame0: ch0=(1,2) ch1=(3,4) ch2=(5,6) | frame1: ch0=(7,8) ch1=(9,10) ch2=(11,12) + val pcm = byteArrayOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12) + val stereo = PCMUtil.pcmToStereo(pcm, 3) + // expected: L=ch0, R=ch1 per frame + assertArrayEquals(byteArrayOf(1, 2, 3, 4, 7, 8, 9, 10), stereo) + } + + @Test + fun `GIVEN a large multichannel frame WHEN pcmToStereo THEN output size is exact and does not overflow`() { + // 4 channels, 2000 frames -> input 16000 bytes, stereo output 8000 bytes (old static buffer was 4096) + val channels = 4 + val frames = 2000 + val pcm = ByteArray(frames * channels * 2) + val stereo = PCMUtil.pcmToStereo(pcm, channels) + assertEquals(frames * 4, stereo.size) + } +} diff --git a/extra-sources/src/main/java/com/pedro/extrasources/Media3AudioSource.kt b/extra-sources/src/main/java/com/pedro/extrasources/Media3AudioSource.kt index 311c2237a3..e2f40c3eac 100644 --- a/extra-sources/src/main/java/com/pedro/extrasources/Media3AudioSource.kt +++ b/extra-sources/src/main/java/com/pedro/extrasources/Media3AudioSource.kt @@ -47,6 +47,11 @@ class Media3AudioSource( throw IllegalArgumentException("Audio file isStereo (${audioInfo.channels > 1}) is different than the configured: $isStereo") } mediaExtractor.release() + return true + } + + override fun start(getMicrophoneData: GetMicrophoneData) { + this.getMicrophoneData = getMicrophoneData player = ExoPlayer.Builder(context, TracksRenderersFactory(context, MediaFrame.Type.AUDIO, processor)).build().also { exoPlayer -> val mediaItem = MediaItem.fromUri(path) exoPlayer.setMediaItem(mediaItem) @@ -59,11 +64,6 @@ class Media3AudioSource( if (state == Player.STATE_ENDED) onFinish(loopMode) } }) - return true - } - - override fun start(getMicrophoneData: GetMicrophoneData) { - this.getMicrophoneData = getMicrophoneData player?.play() } diff --git a/extra-sources/src/main/java/com/pedro/extrasources/Media3VideoSource.kt b/extra-sources/src/main/java/com/pedro/extrasources/Media3VideoSource.kt index e9ca5f7285..8534db1ac2 100644 --- a/extra-sources/src/main/java/com/pedro/extrasources/Media3VideoSource.kt +++ b/extra-sources/src/main/java/com/pedro/extrasources/Media3VideoSource.kt @@ -37,6 +37,11 @@ class Media3VideoSource( throw IllegalArgumentException("Video file track not found") } mediaExtractor.release() + return true + } + + override fun start(surfaceTexture: SurfaceTexture) { + surface = Surface(surfaceTexture) player = ExoPlayer.Builder(context, TracksRenderersFactory(context, MediaFrame.Type.VIDEO)).build().also { exoPlayer -> exoPlayer.setVideoSurface(surface) val mediaItem = MediaItem.fromUri(path) @@ -50,11 +55,6 @@ class Media3VideoSource( if (state == Player.STATE_ENDED) onFinish(loopMode) } }) - return true - } - - override fun start(surfaceTexture: SurfaceTexture) { - surface = Surface(surfaceTexture) player?.play() } diff --git a/library/src/main/java/com/pedro/library/view/GlStreamInterface.kt b/library/src/main/java/com/pedro/library/view/GlStreamInterface.kt index 3de9ee39cc..72d4d03d25 100644 --- a/library/src/main/java/com/pedro/library/view/GlStreamInterface.kt +++ b/library/src/main/java/com/pedro/library/view/GlStreamInterface.kt @@ -227,8 +227,15 @@ class GlStreamInterface(private val context: Context): OnFrameAvailableListener, surfaceHandlerThread?.quitSafely() surfaceHandlerThread = null threadQueue.clear() - executor?.shutdownNow() - executor = null + val executor = this.executor + if (executor != null) { + executor.secureSubmit(100) { releaseSurfaceManagers() } + executor.shutdownNow() + this.executor = null + } else releaseSurfaceManagers() + } + + private fun releaseSurfaceManagers() { sensorRotationManager.stop() surfaceManagerPhoto.release() surfaceManagerEncoder.release() @@ -237,6 +244,7 @@ class GlStreamInterface(private val context: Context): OnFrameAvailableListener, info.surfaceManager.release() } multiPreviewSurfaceManagers.clear() + surfaceManagerPreview.release() surfaceManager.release() mainRender.release() } diff --git a/library/src/main/java/com/pedro/library/view/OpenGlView.kt b/library/src/main/java/com/pedro/library/view/OpenGlView.kt index 81ae5a8beb..4824ec216d 100644 --- a/library/src/main/java/com/pedro/library/view/OpenGlView.kt +++ b/library/src/main/java/com/pedro/library/view/OpenGlView.kt @@ -349,8 +349,15 @@ open class OpenGlView : SurfaceView, GlInterface, OnFrameAvailableListener, Surf surfaceHandlerThread?.quitSafely() surfaceHandlerThread = null threadQueue.clear() - executor?.shutdownNow() - executor = null + val executor = this.executor + if (executor != null) { + executor.secureSubmit(100) { releaseSurfaceManagers() } + executor.shutdownNow() + this.executor = null + } else releaseSurfaceManagers() + } + + private fun releaseSurfaceManagers() { surfaceManagerPhoto.release() surfaceManagerEncoder.release() surfaceManagerEncoderRecord.release() diff --git a/udp/src/main/java/com/pedro/udp/UdpClient.kt b/udp/src/main/java/com/pedro/udp/UdpClient.kt index 42d06a6550..815bc3f66e 100644 --- a/udp/src/main/java/com/pedro/udp/UdpClient.kt +++ b/udp/src/main/java/com/pedro/udp/UdpClient.kt @@ -162,6 +162,7 @@ class UdpClient(private val connectChecker: ConnectChecker) { val host = urlParser.host val port = urlParser.port if (port == null) { + isStreaming = false onMainThread { connectChecker.onConnectionFailed("Endpoint malformed, port is required") } From cae15e3a7306689dbe8acb219a8e90f8aa4ea7d9 Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Thu, 2 Jul 2026 18:01:08 +0200 Subject: [PATCH 04/10] finish fix gl --- .../socket/java/TcpStreamSocketJavaBase.kt | 2 +- .../filters/AndroidViewFilterRender.java | 19 ++++++++++++++----- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/common/src/main/java/com/pedro/common/socket/java/TcpStreamSocketJavaBase.kt b/common/src/main/java/com/pedro/common/socket/java/TcpStreamSocketJavaBase.kt index 10bff3975e..0a0d6942ed 100644 --- a/common/src/main/java/com/pedro/common/socket/java/TcpStreamSocketJavaBase.kt +++ b/common/src/main/java/com/pedro/common/socket/java/TcpStreamSocketJavaBase.kt @@ -73,7 +73,7 @@ abstract class TcpStreamSocketJavaBase: TcpStreamSocket() { return String(bytes) } - override fun isConnected(): Boolean = socket.isConnected + override fun isConnected(): Boolean = socket.isConnected && !socket.isClosed override fun isReachable(): Boolean = socket.inetAddress?.isReachable(timeout.toInt()) ?: false } \ No newline at end of file diff --git a/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/AndroidViewFilterRender.java b/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/AndroidViewFilterRender.java index 83e73bca22..0a64ed77bd 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/AndroidViewFilterRender.java +++ b/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/AndroidViewFilterRender.java @@ -251,6 +251,12 @@ private void startRender() { thread = Executors.newSingleThreadExecutor(); thread.execute(() -> { while (running) { + Surface surface = this.surface; + Surface surface2 = this.surface2; + if (surface == null || this.surface2 == null) { + sleep(); + continue; + } final Status status = renderingStatus; if (status == Status.RENDER1 || status == Status.RENDER2) { final Canvas canvas; @@ -261,6 +267,7 @@ private void startRender() { canvas = status == Status.RENDER1 ? surface.lockCanvas(null) : surface2.lockCanvas(null); } } catch (IllegalStateException e) { + sleep(); continue; } @@ -299,16 +306,18 @@ private void startRender() { } else { // not rendering, no need to try again immediately - try { - Thread.sleep(10); - } catch (InterruptedException e) { - - } + sleep(); } } }); } + private void sleep() { + try { + Thread.sleep(10); + } catch (InterruptedException ignored) {} + } + private void stopRender() { running = false; if (thread != null) { From f9656b0d672c3d4e566c7e23d2224ccd1afb376b Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Thu, 2 Jul 2026 19:01:40 +0200 Subject: [PATCH 05/10] Adding RTMP fixes --- .../main/java/com/pedro/common/Extensions.kt | 6 +- .../main/java/com/pedro/common/UrlParser.kt | 2 +- .../java/com/pedro/rtmp/amf/v0/AmfObject.kt | 75 ++++++------------- .../java/com/pedro/rtmp/amf/v0/AmfString.kt | 15 ++++ .../video/config/VideoSpecificConfigAV1.kt | 10 +-- .../java/com/pedro/rtmp/rtmp/Handshake.kt | 1 + .../java/com/pedro/rtmp/rtmp/RtmpClient.kt | 2 +- .../java/com/pedro/rtmp/amf/AmfObjectTest.kt | 17 +++++ 8 files changed, 65 insertions(+), 63 deletions(-) diff --git a/common/src/main/java/com/pedro/common/Extensions.kt b/common/src/main/java/com/pedro/common/Extensions.kt index 499b97260b..8d441a6eb8 100644 --- a/common/src/main/java/com/pedro/common/Extensions.kt +++ b/common/src/main/java/com/pedro/common/Extensions.kt @@ -125,7 +125,7 @@ fun ExecutorService.secureSubmit(timeout: Long = 5000, code: () -> Unit) { try { if (isTerminated || isShutdown) return submit { code() }.get(timeout, TimeUnit.MILLISECONDS) - } catch (ignored: InterruptedException) {} + } catch (_: Exception) {} } fun String.getMd5Hash(): String { @@ -133,8 +133,8 @@ fun String.getMd5Hash(): String { try { md = MessageDigest.getInstance("MD5") return md.digest(toByteArray()).bytesToHex() - } catch (ignore: NoSuchAlgorithmException) { - } catch (ignore: UnsupportedEncodingException) { + } catch (_: NoSuchAlgorithmException) { + } catch (_: UnsupportedEncodingException) { } return "" } diff --git a/common/src/main/java/com/pedro/common/UrlParser.kt b/common/src/main/java/com/pedro/common/UrlParser.kt index d8b0dd3909..275c6e7a0e 100644 --- a/common/src/main/java/com/pedro/common/UrlParser.kt +++ b/common/src/main/java/com/pedro/common/UrlParser.kt @@ -57,7 +57,7 @@ class UrlParser private constructor( fun getQuery(key: String): String? = getAllQueries()[key] fun getAuthUser(): String? { - val userInfo = auth?.split(":") ?: return null + val userInfo = auth?.split(":", limit = 2) ?: return null return if (userInfo.size == 2) userInfo[0] else null } diff --git a/rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfObject.kt b/rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfObject.kt index ecb2a66906..a6d0a93380 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfObject.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfObject.kt @@ -49,68 +49,37 @@ open class AmfObject(private val properties: LinkedHashMap = return null } - open fun setProperty(name: String, data: String) { - val key = AmfString(name) - val value = AmfString(data) - properties[key] = value - bodySize += key.getSize() - bodySize += value.getSize() + 1 - } + open fun setProperty(name: String, data: String) = putProperty(name, AmfString(data)) - open fun setProperty(name: String, data: Boolean) { - val key = AmfString(name) - val value = AmfBoolean(data) - properties[key] = value - bodySize += key.getSize() - bodySize += value.getSize() + 1 - } + open fun setProperty(name: String, data: Boolean) = putProperty(name, AmfBoolean(data)) - open fun setProperty(name: String) { - val key = AmfString(name) - val value = AmfNull() - properties[key] = value - bodySize += key.getSize() - bodySize += value.getSize() + 1 - } + open fun setProperty(name: String) = putProperty(name, AmfNull()) - open fun setProperty(name: String, data: Double) { - val key = AmfString(name) - val value = AmfNumber(data) - properties[key] = value - bodySize += key.getSize() - bodySize += value.getSize() + 1 - } + open fun setProperty(name: String, data: Double) = putProperty(name, AmfNumber(data)) - open fun setProperty(name: String, data: AmfData) { - val key = AmfString(name) - properties[key] = data - bodySize += key.getSize() - bodySize += data.getSize() + 1 - } + open fun setProperty(name: String, data: AmfData) = putProperty(name, data) open fun setProperty(name: String, data: Any) { - val newValue: AmfData = when (data) { - is String -> AmfString(data) - is Int -> AmfNumber(data.toDouble()) - is Long -> AmfNumber(data.toDouble()) - is Double -> AmfNumber(data) - is Float -> AmfNumber(data.toDouble()) - is Boolean -> AmfBoolean(data) - is AmfData -> data - else -> throw IllegalArgumentException( - "Unsupported value type: ${data::class.java.name}" - ) + val newValue= when (data) { + is String -> AmfString(data) + is Int -> AmfNumber(data.toDouble()) + is Long -> AmfNumber(data.toDouble()) + is Double -> AmfNumber(data) + is Float -> AmfNumber(data.toDouble()) + is Boolean -> AmfBoolean(data) + is AmfData -> data + else -> throw IllegalArgumentException("Unsupported value type: ${data::class.java.name}") } + putProperty(name, newValue) + } - val existingEntry = properties.entries.find { it.key.value == name } - - if (existingEntry != null) { - properties[existingEntry.key] = newValue - bodySize += newValue.getSize() - existingEntry.value.getSize() + private fun putProperty(name: String, value: AmfData) { + val key = AmfString(name) + val previous = properties.put(key, value) + bodySize += if (previous != null) { + value.getSize() - previous.getSize() } else { - val key = AmfString(name) - properties[key] = newValue - bodySize += key.getSize() + newValue.getSize() + 1 + key.getSize() + value.getSize() + 1 } } diff --git a/rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfString.kt b/rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfString.kt index 5b0e7637bd..8b2ec564d2 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfString.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfString.kt @@ -59,4 +59,19 @@ class AmfString(var value: String = ""): AmfData() { override fun toString(): String { return "AmfString value: $value" } + + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (javaClass != other?.javaClass) return false + other as AmfString + if (bodySize != other.bodySize) return false + if (value != other.value) return false + return true + } + + override fun hashCode(): Int { + var result = bodySize + result = 31 * result + value.hashCode() + return result + } } \ No newline at end of file diff --git a/rtmp/src/main/java/com/pedro/rtmp/flv/video/config/VideoSpecificConfigAV1.kt b/rtmp/src/main/java/com/pedro/rtmp/flv/video/config/VideoSpecificConfigAV1.kt index 0311466a49..ffd036e83a 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/flv/video/config/VideoSpecificConfigAV1.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/flv/video/config/VideoSpecificConfigAV1.kt @@ -159,12 +159,12 @@ class VideoSpecificConfigAV1(private val sequenceObu: ByteArray) { } val subsamplingX: Boolean val subsamplingY: Boolean - var samplePosition = false + var samplePosition = 0 if (monochrome) { bitBuffer.getBool() subsamplingX = true subsamplingY = true - } else if (colorPrimaries == 1 && transferCharacteristics == 1 && matrixCoefficients == 1) { + } else if (colorPrimaries == 1 && transferCharacteristics == 13 && matrixCoefficients == 0) { subsamplingX = false subsamplingY = false } else { @@ -187,9 +187,9 @@ class VideoSpecificConfigAV1(private val sequenceObu: ByteArray) { subsamplingX = true subsamplingY = false } - if (subsamplingX && subsamplingY) { - samplePosition = bitBuffer.getBool() - } + } + if (subsamplingX && subsamplingY) { + samplePosition = bitBuffer.getInt(2) } } //finish config color diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/Handshake.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/Handshake.kt index b17da83907..908223cf58 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/Handshake.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/Handshake.kt @@ -134,6 +134,7 @@ class Handshake { private suspend fun readS0(socket: RtmpSocket): ByteArray { Log.i(TAG, "reading S0") val response = socket.read() + //few servers send 72 as response of c0 + c1. Assuming correct the connection worked. if (response == protocolVersion || response == 72) { Log.i(TAG, "read S0 successful") return byteArrayOf(response.toByte()) diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt index a2fc8e9881..e69bd82bc6 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt @@ -141,7 +141,7 @@ class RtmpClient(private val connectChecker: ConnectChecker) { if (!isStreaming) { commandsManager = when (amfVersion) { AmfVersion.VERSION_0 -> CommandsManagerAmf0() - AmfVersion.VERSION_3 -> CommandsManagerAmf3() + AmfVersion.VERSION_3 -> TODO("Not yet implemented") } } } diff --git a/rtmp/src/test/java/com/pedro/rtmp/amf/AmfObjectTest.kt b/rtmp/src/test/java/com/pedro/rtmp/amf/AmfObjectTest.kt index b88cd194ab..2ce9bbc636 100644 --- a/rtmp/src/test/java/com/pedro/rtmp/amf/AmfObjectTest.kt +++ b/rtmp/src/test/java/com/pedro/rtmp/amf/AmfObjectTest.kt @@ -72,4 +72,21 @@ class AmfObjectTest { assertArrayEquals(expectedBuffer, output.toByteArray()) } + + @Test + fun `GIVEN a property set twice with the same name WHEN written THEN it is not duplicated and size stays consistent`() { + val amfObject = AmfObject() + amfObject.setProperty("name", "first") + amfObject.setProperty("name", "second") //re-set same name via typed overload + + //no duplicated key and last value wins + assertEquals(1, amfObject.getProperties().size) + val value = amfObject.getProperty("name") + assertTrue(value is AmfString && value.value == "second") + + //getSize() must match the bytes actually written by writeBody() + val body = ByteArrayOutputStream() + amfObject.writeBody(body) + assertEquals(body.size(), amfObject.getSize()) + } } \ No newline at end of file From eaa6ea663baf0912f3e6d20f99a1fb80874e7e19 Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Thu, 2 Jul 2026 19:27:43 +0200 Subject: [PATCH 06/10] add PitchAudioFilter --- .../main/java/com/pedro/common/UrlParser.kt | 2 +- .../encoder/input/audio/PitchShiftEffect.kt | 90 +++++++++++++++++++ .../com/pedro/encoder/PitchShiftEffectTest.kt | 63 +++++++++++++ 3 files changed, 154 insertions(+), 1 deletion(-) create mode 100644 encoder/src/main/java/com/pedro/encoder/input/audio/PitchShiftEffect.kt create mode 100644 encoder/src/test/java/com/pedro/encoder/PitchShiftEffectTest.kt diff --git a/common/src/main/java/com/pedro/common/UrlParser.kt b/common/src/main/java/com/pedro/common/UrlParser.kt index 275c6e7a0e..1b572b5395 100644 --- a/common/src/main/java/com/pedro/common/UrlParser.kt +++ b/common/src/main/java/com/pedro/common/UrlParser.kt @@ -62,7 +62,7 @@ class UrlParser private constructor( } fun getAuthPassword(): String? { - val userInfo = auth?.split(":") ?: return null + val userInfo = auth?.split(":", limit = 2) ?: return null return if (userInfo.size == 2) userInfo[1] else null } diff --git a/encoder/src/main/java/com/pedro/encoder/input/audio/PitchShiftEffect.kt b/encoder/src/main/java/com/pedro/encoder/input/audio/PitchShiftEffect.kt new file mode 100644 index 0000000000..82c2e38a95 --- /dev/null +++ b/encoder/src/main/java/com/pedro/encoder/input/audio/PitchShiftEffect.kt @@ -0,0 +1,90 @@ +/* + * Copyright (C) 2024 pedroSG94. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.pedro.encoder.input.audio + +import kotlin.math.PI +import kotlin.math.cos + +/** + * [CustomAudioEffect] that shifts the pitch up (or down) keeping the output buffer + * exactly the same size as the input, so it never breaks the audio pipeline. + * With [pitch] > 1 you get the typical "chipmunk"/squirrel voice. + * + * Unlike a plain resample (which would change the number of samples and therefore + * the buffer size), this is a time-domain pitch shifter: a delay line whose read + * position drifts against the write position at the [pitch] ratio. Two read taps + * offset by half the window are Hann-crossfaded so the wrap-around is inaudible + * (the two Hann windows offset by half a period sum to 1 → constant amplitude). + * Because it produces one output sample per input sample, the length is preserved. + * + * Expects 16 bit little endian PCM (the format the microphone delivers). It is + * designed for mono; interleaved stereo still sounds fine but bleeds slightly + * between channels at the tap boundary. + * + */ +class PitchShiftEffect(windowSize: Int = 1024) : CustomAudioEffect() { + + /** + * Pitch ratio. 1f keeps the original pitch, values > 1f raise it (chipmunk), + * values < 1f lower it. Clamped to a sane range to avoid heavy aliasing. + */ + var pitch = 1.8f + set(value) { field = value.coerceIn(0.5f, 3f) } + + private val size = windowSize.coerceAtLeast(2) + private val ring = FloatArray(size) + private var writePos = 0 + private var delay = 0f + + override fun process(pcmBuffer: ByteArray): ByteArray { + val samples = pcmBuffer.size / 2 + val step = pitch - 1f + val half = size / 2f + for (i in 0 until samples) { + val idx = i * 2 + //read one 16 bit little endian sample and store it in the delay line + val input = ((pcmBuffer[idx].toInt() and 0xFF) or (pcmBuffer[idx + 1].toInt() shl 8)).toShort() + ring[writePos] = input.toFloat() + + //two taps offset by half the window, Hann-crossfaded (the two windows sum to 1) + var out = 0f + for (tap in 0 until 2) { + var d = delay + tap * half + if (d >= size) d -= size + //read position "d" samples behind the write head, wrapped into the ring + var readPos = writePos - d + if (readPos < 0f) readPos += size + val i0 = readPos.toInt() + val i1 = if (i0 + 1 >= size) 0 else i0 + 1 + val frac = readPos - i0 + val sample = ring[i0] * (1f - frac) + ring[i1] * frac + val window = 0.5f * (1f - cos(2.0 * PI * (d / size)).toFloat()) + out += sample * window + } + + //advance write head and drift the delay at the pitch ratio (wraps around the ring) + if (++writePos >= size) writePos = 0 + delay -= step + if (delay < 0f) delay += size else if (delay >= size) delay -= size + + //write the result back as 16 bit little endian, clamped to avoid overflow + val s = out.toInt().coerceIn(-32768, 32767) + pcmBuffer[idx] = (s and 0xFF).toByte() + pcmBuffer[idx + 1] = ((s shr 8) and 0xFF).toByte() + } + return pcmBuffer + } +} diff --git a/encoder/src/test/java/com/pedro/encoder/PitchShiftEffectTest.kt b/encoder/src/test/java/com/pedro/encoder/PitchShiftEffectTest.kt new file mode 100644 index 0000000000..283f659195 --- /dev/null +++ b/encoder/src/test/java/com/pedro/encoder/PitchShiftEffectTest.kt @@ -0,0 +1,63 @@ +package com.pedro.encoder + +import com.pedro.encoder.input.audio.PitchShiftEffect +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Test +import kotlin.math.PI +import kotlin.math.sin + +class PitchShiftEffectTest { + + @Test + fun `keeps the same buffer size for several sizes and pitches`() { + for (window in intArrayOf(256, 1024, 2048)) { + val effect = PitchShiftEffect(window) + for (pitch in floatArrayOf(0.5f, 1f, 1.8f, 3f)) { + effect.pitch = pitch + for (bytes in intArrayOf(64, 1024, 8192)) { + val out = effect.process(ByteArray(bytes) { (it and 0xFF).toByte() }) + assertEquals(bytes, out.size) + } + } + } + } + + @Test + fun `raises pitch (about twice the zero crossings) when pitch is 2`() { + val fs = 44100.0 + val freq = 440.0 + val n = 16384 + val inCrossings = zeroCrossings(sineBytes(n, freq, fs)) + + val effect = PitchShiftEffect().apply { pitch = 2f } + effect.process(sineBytes(n, freq, fs)) //warm up the delay line + val outCrossings = zeroCrossings(effect.process(sineBytes(n, freq, fs))) + + assertTrue( + "expected more zero crossings after pitch up: in=$inCrossings out=$outCrossings", + outCrossings > inCrossings * 1.5 + ) + } + + private fun sineBytes(samples: Int, freq: Double, fs: Double): ByteArray { + val out = ByteArray(samples * 2) + for (i in 0 until samples) { + val v = (sin(2 * PI * freq * i / fs) * 12000).toInt() + out[i * 2] = (v and 0xFF).toByte() + out[i * 2 + 1] = ((v shr 8) and 0xFF).toByte() + } + return out + } + + private fun zeroCrossings(bytes: ByteArray): Int { + var count = 0 + var prev = 0 + for (i in 0 until bytes.size / 2) { + val s = ((bytes[i * 2].toInt() and 0xFF) or (bytes[i * 2 + 1].toInt() shl 8)).toShort().toInt() + if (i > 0 && ((prev < 0 && s >= 0) || (prev >= 0 && s < 0))) count++ + prev = s + } + return count + } +} From 1ba6fa886b69abb5b02b17b7058f409a638a62f4 Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Fri, 3 Jul 2026 21:19:17 +0200 Subject: [PATCH 07/10] finx encoder module --- .../com/pedro/streamer/utils/FilterMenu.kt | 2 +- .../java/com/pedro/encoder/audio/G711Codec.kt | 2 +- .../encoder/input/gl/AndroidViewSprite.java | 2 +- .../com/pedro/encoder/input/gl/Sprite.java | 2 +- .../input/gl/SpriteGestureController.java | 8 ++++---- .../pedro/encoder/input/gl/TextureLoader.java | 1 - .../encoder/input/gl/render/CameraRender.java | 5 ++--- .../gl/render/filters/CropFilterRender.kt | 4 ++-- .../encoder/input/video/Camera2ApiManager.kt | 2 +- .../encoder/utils/gl/GifStreamObject.java | 6 ++++-- .../com/pedro/encoder/utils/gl/GlUtil.java | 11 ++++++++-- .../encoder/utils/gl/TextStreamObject.java | 4 ++-- .../java/com/pedro/rtmp/rtmp/RtmpClient.kt | 12 +++++++---- .../com/pedro/rtmp/rtmp/message/RtmpHeader.kt | 2 ++ .../java/com/pedro/rtmp/utils/AuthUtil.kt | 20 ++++++++++--------- 15 files changed, 49 insertions(+), 34 deletions(-) diff --git a/app/src/main/java/com/pedro/streamer/utils/FilterMenu.kt b/app/src/main/java/com/pedro/streamer/utils/FilterMenu.kt index 48090b6827..2d05fa87b9 100644 --- a/app/src/main/java/com/pedro/streamer/utils/FilterMenu.kt +++ b/app/src/main/java/com/pedro/streamer/utils/FilterMenu.kt @@ -225,7 +225,7 @@ class FilterMenu(private val context: Context) { BitmapFactory.decodeResource(context.resources, R.mipmap.ic_launcher) ) imageObjectFilterRender.setScale(50f, 50f) - imageObjectFilterRender.setPosition(TranslateTo.RIGHT) + imageObjectFilterRender.setPosition(TranslateTo.CENTER) spriteGestureController.setBaseObjectFilterRender(imageObjectFilterRender) //Optional spriteGestureController.setPreventMoveOutside(false) //Optional return true diff --git a/encoder/src/main/java/com/pedro/encoder/audio/G711Codec.kt b/encoder/src/main/java/com/pedro/encoder/audio/G711Codec.kt index b904c7d868..a5af77fa85 100644 --- a/encoder/src/main/java/com/pedro/encoder/audio/G711Codec.kt +++ b/encoder/src/main/java/com/pedro/encoder/audio/G711Codec.kt @@ -41,7 +41,7 @@ class G711Codec { fun decode(src: ByteArray, offset: Int, len: Int): ByteArray { var j = 0 - val out = ByteArray(src.size * 2) + val out = ByteArray(len * 2) for (i in 0 until len) { val s = aLawDecompressTable[src[i + offset].toInt() and 0xff] out[j++] = s.toByte() diff --git a/encoder/src/main/java/com/pedro/encoder/input/gl/AndroidViewSprite.java b/encoder/src/main/java/com/pedro/encoder/input/gl/AndroidViewSprite.java index 377b3dbb06..62e6db744a 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/gl/AndroidViewSprite.java +++ b/encoder/src/main/java/com/pedro/encoder/input/gl/AndroidViewSprite.java @@ -57,7 +57,7 @@ public void translate(TranslateTo translation) { switch (translation) { case CENTER: this.position.x = 50f - scale.x / 2f; - this.position.y = 50f - scale.x / 2f; + this.position.y = 50f - scale.y / 2f; break; case BOTTOM: this.position.x = 50f - scale.x / 2f; diff --git a/encoder/src/main/java/com/pedro/encoder/input/gl/Sprite.java b/encoder/src/main/java/com/pedro/encoder/input/gl/Sprite.java index 009a8c2005..1a6eb93055 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/gl/Sprite.java +++ b/encoder/src/main/java/com/pedro/encoder/input/gl/Sprite.java @@ -71,7 +71,7 @@ public void translate(TranslateTo translation) { switch (translation) { case CENTER: this.position.x = 50f - scale.x / 2f; - this.position.y = 50f - scale.x / 2f; + this.position.y = 50f - scale.y / 2f; break; case BOTTOM: this.position.x = 50f - scale.x / 2f; diff --git a/encoder/src/main/java/com/pedro/encoder/input/gl/SpriteGestureController.java b/encoder/src/main/java/com/pedro/encoder/input/gl/SpriteGestureController.java index 338e3c7664..2dc3b6c111 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/gl/SpriteGestureController.java +++ b/encoder/src/main/java/com/pedro/encoder/input/gl/SpriteGestureController.java @@ -144,12 +144,12 @@ public void scaleSprite(MotionEvent motionEvent) { } else { scale = androidViewFilterRender.getScale(); } - scale.x += percent; - scale.y += percent; + float newScaleX = scale.x + percent; + float newScaleY = scale.y + percent; if (baseObjectFilterRender != null) { - baseObjectFilterRender.setScale(scale.x, scale.y); + baseObjectFilterRender.setScale(newScaleX, newScaleY); } else { - androidViewFilterRender.setScale(scale.x, scale.y); + androidViewFilterRender.setScale(newScaleX, newScaleY); } lastDistance = distance; } diff --git a/encoder/src/main/java/com/pedro/encoder/input/gl/TextureLoader.java b/encoder/src/main/java/com/pedro/encoder/input/gl/TextureLoader.java index 7ae3ecf0c5..b4c2e35eb3 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/gl/TextureLoader.java +++ b/encoder/src/main/java/com/pedro/encoder/input/gl/TextureLoader.java @@ -39,7 +39,6 @@ public int[] load(Bitmap[] bitmaps) { if (bitmaps[i] != null && !bitmaps[i].isRecycled()) { GLES20.glBindTexture(GLES20.GL_TEXTURE_2D, textureId[i]); GLUtils.texImage2D(GLES20.GL_TEXTURE_2D, 0, bitmaps[i], 0); - bitmaps[i].recycle(); } bitmaps[i] = null; } diff --git a/encoder/src/main/java/com/pedro/encoder/input/gl/render/CameraRender.java b/encoder/src/main/java/com/pedro/encoder/input/gl/render/CameraRender.java index 226cf41659..82e5a629ac 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/gl/render/CameraRender.java +++ b/encoder/src/main/java/com/pedro/encoder/input/gl/render/CameraRender.java @@ -50,8 +50,8 @@ public class CameraRender extends BaseRenderOffScreen { private int aPositionHandle = -1; private int aTextureCameraHandle = -1; - private SurfaceTexture surfaceTexture = new SurfaceTexture(textureID[0]); - private Surface surface = new Surface(surfaceTexture); + private SurfaceTexture surfaceTexture; + private Surface surface; private int width; private int height; @@ -81,7 +81,6 @@ public void initGl(int width, int height, Context context, int previewWidth, aTextureCameraHandle = GLES20.glGetAttribLocation(program, "aTextureCoord"); uMVPMatrixHandle = GLES20.glGetUniformLocation(program, "uMVPMatrix"); uSTMatrixHandle = GLES20.glGetUniformLocation(program, "uSTMatrix"); - uSTMatrixHandle = GLES20.glGetUniformLocation(program, "uSTMatrix"); //camera texture GlUtil.createExternalTextures(textureID.length, textureID, 0); diff --git a/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/CropFilterRender.kt b/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/CropFilterRender.kt index 9d8aaff04c..20d7d154cb 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/CropFilterRender.kt +++ b/encoder/src/main/java/com/pedro/encoder/input/gl/render/filters/CropFilterRender.kt @@ -112,8 +112,8 @@ class CropFilterRender: BaseFilterRender() { val initialX = 1f - (1f / scaleX) val initialY = -(1f - (1f / scaleY)) - val percentX = initialX / (50f - (width / 2f)) - val percentY = -initialY / (50f - (height / 2f)) + val percentX = if (width == 100f) 0f else initialX / (50f - (width / 2f)) + val percentY = if (height == 100f) 0f else -initialY / (50f - (height / 2f)) val oX = initialX - (offsetX * percentX) val oY = initialY + (offsetY * percentY) diff --git a/encoder/src/main/java/com/pedro/encoder/input/video/Camera2ApiManager.kt b/encoder/src/main/java/com/pedro/encoder/input/video/Camera2ApiManager.kt index 9b1a962191..ab729425f3 100644 --- a/encoder/src/main/java/com/pedro/encoder/input/video/Camera2ApiManager.kt +++ b/encoder/src/main/java/com/pedro/encoder/input/video/Camera2ApiManager.kt @@ -276,7 +276,7 @@ class Camera2ApiManager(context: Context) : CameraDevice.StateCallback() { val streamConfigurationMap = characteristics.get(CameraCharacteristics.SCALER_STREAM_CONFIGURATION_MAP) val list = mutableListOf>() val fd = streamConfigurationMap?.getOutputMinFrameDuration(SurfaceTexture::class.java, size) ?: return emptyList() - val maxFPS = (10f / "0.$fd".toFloat()).toInt() + val maxFPS = (1e9 / fd).toInt() for (r in fpsSupported) { if (r.upper <= maxFPS) { list.add(r) } } list } else listOf(*fpsSupported) diff --git a/encoder/src/main/java/com/pedro/encoder/utils/gl/GifStreamObject.java b/encoder/src/main/java/com/pedro/encoder/utils/gl/GifStreamObject.java index 9d026c7e91..02979ca4da 100644 --- a/encoder/src/main/java/com/pedro/encoder/utils/gl/GifStreamObject.java +++ b/encoder/src/main/java/com/pedro/encoder/utils/gl/GifStreamObject.java @@ -44,12 +44,12 @@ public GifStreamObject() { @Override public int getWidth() { - return gifBitmaps != null ? gifBitmaps[0].getWidth() : 0; + return gifBitmaps != null && gifBitmaps.length > 0 && gifBitmaps[0] != null ? gifBitmaps[0].getWidth() : 0; } @Override public int getHeight() { - return gifBitmaps != null ? gifBitmaps[0].getHeight() : 0; + return gifBitmaps != null && gifBitmaps.length > 0 && gifBitmaps[0] != null ? gifBitmaps[0].getHeight() : 0; } public void load(InputStream inputStreamGif) throws IOException { @@ -57,6 +57,7 @@ public void load(InputStream inputStreamGif) throws IOException { if (gifDecoder.read(inputStreamGif, inputStreamGif.available()) == 0) { Log.i(TAG, "read gif ok"); numFrames = gifDecoder.getFrameCount(); + if (numFrames <= 0) throw new IOException("Read gif error: no frames"); gifDelayFrames = new int[numFrames]; gifBitmaps = new Bitmap[numFrames]; for (int i = 0; i < numFrames; i++) { @@ -99,6 +100,7 @@ public int updateFrame(int size) { @Override public int updateFrame() { + if (gifDelayFrames == null || numFrames <= 0) return 0; if (startDelayFrame == 0) { startDelayFrame = TimeUtils.getCurrentTimeMillis(); } diff --git a/encoder/src/main/java/com/pedro/encoder/utils/gl/GlUtil.java b/encoder/src/main/java/com/pedro/encoder/utils/gl/GlUtil.java index 6fe23ae255..eb768e8566 100644 --- a/encoder/src/main/java/com/pedro/encoder/utils/gl/GlUtil.java +++ b/encoder/src/main/java/com/pedro/encoder/utils/gl/GlUtil.java @@ -81,7 +81,7 @@ public static int createProgram(String vertexSource, String fragmentSource) { public static void createTextures(int quantity, int[] texturesId, int offset, int filter, boolean isExternal) { GLES20.glGenTextures(quantity, texturesId, offset); final int type = isExternal ? GLES11Ext.GL_TEXTURE_EXTERNAL_OES : GLES20.GL_TEXTURE_2D; - for (int i = offset; i < quantity; i++) { + for (int i = offset; i < offset + quantity; i++) { GLES20.glActiveTexture(GLES20.GL_TEXTURE0 + i); GLES20.glBindTexture(type, texturesId[i]); GLES20.glTexParameterf(type, GLES20.GL_TEXTURE_MIN_FILTER, filter); @@ -156,9 +156,16 @@ private static Bitmap flipVerticalBitmap(Bitmap bitmap, int width, int height) { return Bitmap.createBitmap(bitmap, 0, 0, width, height, matrix, true); } + private static int maxVertexAttribs = -1; + public static void disableResources(int... vertex) { + if (maxVertexAttribs < 0) { + int[] value = new int[1]; + GLES20.glGetIntegerv(GLES20.GL_MAX_VERTEX_ATTRIBS, value, 0); + maxVertexAttribs = value[0]; + } for (int v: vertex) { - if (v >= 0 && v < GLES20.GL_MAX_VERTEX_ATTRIBS) GLES20.glDisableVertexAttribArray(v); + if (v >= 0 && v < maxVertexAttribs) GLES20.glDisableVertexAttribArray(v); } GLES20.glBindTexture(GLES11Ext.GL_TEXTURE_EXTERNAL_OES, 0); GLES20.glBindTexture(GLES20.GL_TEXTURE_2D, 0); diff --git a/encoder/src/main/java/com/pedro/encoder/utils/gl/TextStreamObject.java b/encoder/src/main/java/com/pedro/encoder/utils/gl/TextStreamObject.java index 1615b935af..6841363c29 100644 --- a/encoder/src/main/java/com/pedro/encoder/utils/gl/TextStreamObject.java +++ b/encoder/src/main/java/com/pedro/encoder/utils/gl/TextStreamObject.java @@ -66,8 +66,8 @@ private Bitmap textAsBitmap(String text, float textSize, int textColor, int back paint.setTextAlign(Paint.Align.LEFT); float baseline = -paint.ascent(); // ascent() is negative - int width = (int) (paint.measureText(text) + 0.5f); // round - int height = (int) (baseline + paint.descent() + 0.5f); + int width = Math.max(1, (int) (paint.measureText(text) + 0.5f)); // round + int height = Math.max(1, (int) (baseline + paint.descent() + 0.5f)); Bitmap image = Bitmap.createBitmap(width, height, Bitmap.Config.ARGB_8888); Canvas canvas = new Canvas(image); canvas.drawColor(backgroundColor); diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt index e69bd82bc6..0a292df6d8 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt @@ -61,6 +61,7 @@ import java.net.URISyntaxException import java.nio.ByteBuffer import java.util.concurrent.atomic.AtomicLong import javax.net.ssl.TrustManager +import kotlin.time.Duration.Companion.milliseconds /** * Created by pedro on 8/04/21. @@ -302,7 +303,10 @@ class RtmpClient(private val connectChecker: ConnectChecker) { //Handle all command received and send response for it. handleMessages() } - if (shouldSendPings) commandsManager.sendPing(socket) + if (shouldSendPings) { + pingTs.set(TimeUtils.getCurrentTimeMicro()) + commandsManager.sendPing(socket) + } //read packet because maybe server want send you something while streaming handleServerPackets() }.exceptionOrNull() @@ -409,7 +413,7 @@ class RtmpClient(private val connectChecker: ConnectChecker) { if (shouldSendPings) { rtt = (TimeUtils.getCurrentTimeMicro() - pingTs.get()).toInt() scopePing.launch { - delay(1000) + delay(1000.milliseconds) if (isStreaming) { pingTs.set(TimeUtils.getCurrentTimeMicro()) commandsManager.sendPing(socket) @@ -557,7 +561,7 @@ class RtmpClient(private val connectChecker: ConnectChecker) { jobRetry = scopeRetry.launch { reTries-- disconnect(false) - delay(delay) + delay(delay.milliseconds) val reconnectUrl = backupUrl ?: url connect(reconnectUrl, true) } @@ -572,7 +576,7 @@ class RtmpClient(private val connectChecker: ConnectChecker) { private suspend fun disconnect(clear: Boolean) { if (isStreaming) rtmpSender.stop(clear) runCatching { - withTimeoutOrNull(100) { + withTimeoutOrNull(100.milliseconds) { socket?.let { commandsManager.sendClose(it) } } } diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/RtmpHeader.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/RtmpHeader.kt index a9bf2cc06f..bd063a269a 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/RtmpHeader.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/RtmpHeader.kt @@ -66,6 +66,7 @@ class RtmpHeader(var basicHeader: BasicHeader) { messageType = RtmpMessage.getMarkType(socket.read()) //extended timestamp if (timeStamp >= 0xffffff) timeStamp = socket.readUInt32() + if (lastHeader != null) timeStamp += lastHeader.timeStamp } ChunkType.TYPE_2 -> { if (lastHeader != null) { @@ -76,6 +77,7 @@ class RtmpHeader(var basicHeader: BasicHeader) { timeStamp = socket.readUInt24() //extended timestamp if (timeStamp >= 0xffffff) timeStamp = socket.readUInt32() + if (lastHeader != null) timeStamp += lastHeader.timeStamp } ChunkType.TYPE_3 -> { if (lastHeader != null) { diff --git a/rtmp/src/main/java/com/pedro/rtmp/utils/AuthUtil.kt b/rtmp/src/main/java/com/pedro/rtmp/utils/AuthUtil.kt index 60c574d825..13b472925a 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/utils/AuthUtil.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/utils/AuthUtil.kt @@ -68,11 +68,11 @@ object AuthUtil { return "?authmod=$authMod&user=$user&nonce=$nonce&cnonce=$cNonce&nc=$ncHex&response=$hash3" } - fun getSalt(description: String): String = findDescriptionValue("salt=", description) + fun getSalt(description: String): String = findDescriptionValue("salt", description) - fun getChallenge(description: String): String = findDescriptionValue("challenge=", description) + fun getChallenge(description: String): String = findDescriptionValue("challenge", description) - fun getOpaque(description: String): String = findDescriptionValue("opaque=", description) + fun getOpaque(description: String): String = findDescriptionValue("opaque", description) @OptIn(ExperimentalEncodingApi::class) fun stringToMd5Base64(s: String): String { @@ -81,21 +81,23 @@ object AuthUtil { md.update(s.toByteArray()) val md5hash = md.digest() return Base64.encode(md5hash) - } catch (ignore: Exception) { } + } catch (_: Exception) { } return "" } /** * Limelight auth utils */ - fun getNonce(description: String): String = findDescriptionValue("nonce=", description) + fun getNonce(description: String): String = findDescriptionValue("nonce", description) - private fun findDescriptionValue(value: String, description: String): String { + private fun findDescriptionValue(keyToFind: String, description: String): String { val data = description.split("&").toTypedArray() for (s in data) { - if (s.contains(value)) { - return s.substring(value.length) - } + val param = s.split("=", limit = 2) + if (param.size < 2) continue + val key = param[0] + val value = param[1] + if (key.contains(keyToFind, ignoreCase = true)) return value } return "" } From 907eabf2a695353c13611676278a221e3a25da48 Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Fri, 3 Jul 2026 21:49:29 +0200 Subject: [PATCH 08/10] finish fix rtsp --- .../main/java/com/pedro/rtsp/rtsp/RtspClient.kt | 12 ++++++------ .../pedro/rtsp/rtsp/commands/CommandsManager.kt | 14 ++++++-------- whip/src/main/java/com/pedro/whip/WhipClient.kt | 1 + .../java/com/pedro/whip/webrtc/CommandsManager.kt | 12 +++++------- 4 files changed, 18 insertions(+), 21 deletions(-) diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt b/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt index 8e1a1c4636..9065a26515 100644 --- a/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt +++ b/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt @@ -33,7 +33,6 @@ import com.pedro.common.toMediaFrameInfo import com.pedro.common.validMessage import com.pedro.rtsp.rtsp.commands.CommandsManager import com.pedro.rtsp.rtsp.commands.Method -import com.pedro.rtsp.utils.RtpConstants import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job @@ -47,6 +46,7 @@ import kotlinx.coroutines.withTimeoutOrNull import java.net.URISyntaxException import java.nio.ByteBuffer import javax.net.ssl.TrustManager +import kotlin.time.Duration.Companion.milliseconds /** * Created by pedro on 10/02/17. @@ -228,7 +228,7 @@ class RtspClient(private val connectChecker: ConnectChecker) { if (user != null && password != null) setAuthorization(user, password) val error = runCatching { - commandsManager.updateNtpTimestamp() + commandsManager.updateTimestamp() commandsManager.setUrl(host, port, "/$path") if (!commandsManager.audioDisabled) { rtspSender.setAudioInfo(commandsManager.sampleRate, commandsManager.isStereo) @@ -236,7 +236,7 @@ class RtspClient(private val connectChecker: ConnectChecker) { if (!commandsManager.videoDisabled) { if (!commandsManager.videoInfoReady()) { Log.i(TAG, "waiting for sps and pps") - withTimeoutOrNull(5000) { + withTimeoutOrNull(5000.milliseconds) { mutex.lock() } if (!commandsManager.videoInfoReady()) { @@ -383,7 +383,7 @@ class RtspClient(private val connectChecker: ConnectChecker) { while (scope.isActive && isStreaming) { val error = runCatching { if (isAlive()) { - delay(2000) + delay(2000.milliseconds) socket?.let { socket -> if (socket.isConnected()) { val command = commandsManager.getResponse(socket) @@ -423,7 +423,7 @@ class RtspClient(private val connectChecker: ConnectChecker) { private suspend fun disconnect(clear: Boolean) { if (isStreaming) rtspSender.stop() val error = runCatching { - withTimeoutOrNull(100) { + withTimeoutOrNull(100.milliseconds) { socket?.write(commandsManager.createTeardown()) socket?.flush() } @@ -479,7 +479,7 @@ class RtspClient(private val connectChecker: ConnectChecker) { jobRetry = scopeRetry.launch { reTries-- disconnect(false) - delay(delay) + delay(delay.milliseconds) val reconnectUrl = backupUrl ?: url connect(reconnectUrl, true) } diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandsManager.kt b/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandsManager.kt index d6b2d66dac..dc9fe406d7 100644 --- a/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandsManager.kt +++ b/rtsp/src/main/java/com/pedro/rtsp/rtsp/commands/CommandsManager.kt @@ -29,7 +29,6 @@ import com.pedro.rtsp.rtsp.commands.SdpBody.createG711Body import com.pedro.rtsp.rtsp.commands.SdpBody.createH264Body import com.pedro.rtsp.rtsp.commands.SdpBody.createH265Body import com.pedro.rtsp.rtsp.commands.SdpBody.createOpusBody -import com.pedro.rtsp.utils.RtpConstants import com.pedro.rtsp.utils.RtpTracks import com.pedro.rtsp.utils.encodeToString import com.pedro.rtsp.utils.getData @@ -58,7 +57,7 @@ open class CommandsManager { private set private var cSeq = 0 private var sessionId: String? = null - private var timeStamp: ULong = 0uL + private var timeStamp = 0L var sampleRate = 32000 var isStereo = true var protocol: Protocol = Protocol.TCP @@ -266,9 +265,11 @@ open class CommandsManager { var line: String? while (socket.readLine().also { line = it } != null) { response += "${line ?: ""}\n" - //end of response if ((line?.length ?: 0) < 3) break } + val contentLength = Regex("Content-Length:\\s*(\\d+)", RegexOption.IGNORE_CASE) + .find(response)?.groupValues?.getOrNull(1)?.toIntOrNull() ?: 0 + if (contentLength > 0) response += socket.read(contentLength).decodeToString() Log.i(TAG, response) return if (method == Method.UNKNOWN) { commandParser.parseCommand(response) @@ -305,10 +306,7 @@ open class CommandsManager { return "" } - fun updateNtpTimestamp() { - val uptime = TimeUtils.getCurrentTimeMillis() - val seconds = (uptime / 1000).toULong() - val fraction = ((uptime % 1000) * (1L shl 32) / 1000).toULong() - timeStamp = (seconds shl 32) or fraction + fun updateTimestamp() { + timeStamp = TimeUtils.getCurrentTimeNano() } } \ No newline at end of file diff --git a/whip/src/main/java/com/pedro/whip/WhipClient.kt b/whip/src/main/java/com/pedro/whip/WhipClient.kt index b008214337..43eef9e730 100644 --- a/whip/src/main/java/com/pedro/whip/WhipClient.kt +++ b/whip/src/main/java/com/pedro/whip/WhipClient.kt @@ -207,6 +207,7 @@ class WhipClient(private val connectChecker: ConnectChecker) { } val error = runCatching { + commandsManager.updateTimestamp() commandsManager.setUrl(host, port, path, tlsEnabled) if (!commandsManager.audioDisabled) { whipSender.setAudioInfo(commandsManager.sampleRate, commandsManager.isStereo) diff --git a/whip/src/main/java/com/pedro/whip/webrtc/CommandsManager.kt b/whip/src/main/java/com/pedro/whip/webrtc/CommandsManager.kt index b56dc4d8e6..5100064ceb 100644 --- a/whip/src/main/java/com/pedro/whip/webrtc/CommandsManager.kt +++ b/whip/src/main/java/com/pedro/whip/webrtc/CommandsManager.kt @@ -57,7 +57,7 @@ class CommandsManager { var videoCodec = VideoCodec.H264 var audioCodec = AudioCodec.OPUS private val timeout = 5000 - private val timeStamp: Long + private var timeStamp = 0L private val secureRandom = SecureRandom() val rtpTracks = RtpTracks() private var certificates: TrustManager? = null @@ -85,12 +85,6 @@ class CommandsManager { private const val TAG = "CommandsManager" } - init { - val uptime = TimeUtils.getCurrentTimeMillis() - timeStamp = uptime / 1000 shl 32 and ((uptime - uptime / 1000 * 1000 shr 32) - / 1000) // NTP timestamp - } - fun addCertificates(certificates: TrustManager?) { this.certificates = certificates } @@ -335,4 +329,8 @@ class CommandsManager { fun generateTransactionId(): ByteArray { return secureRandom.nextBytes(12) } + + fun updateTimestamp() { + timeStamp = TimeUtils.getCurrentTimeNano() + } } \ No newline at end of file From 0cca95ad81ea4ef7ed5bed8468a35f5d55d2fa9b Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Fri, 3 Jul 2026 23:22:20 +0200 Subject: [PATCH 09/10] finish fixes in SRT --- .../java/com/pedro/srt/mpeg2ts/AdaptationField.kt | 14 +++++--------- srt/src/main/java/com/pedro/srt/mpeg2ts/Pes.kt | 7 +------ .../com/pedro/srt/mpeg2ts/packets/AacPacket.kt | 2 +- .../com/pedro/srt/mpeg2ts/packets/OpusPacket.kt | 2 +- .../main/java/com/pedro/srt/srt/CommandsManager.kt | 8 ++++++++ srt/src/main/java/com/pedro/srt/srt/SrtClient.kt | 11 +++++------ srt/src/main/java/com/pedro/srt/srt/SrtSender.kt | 2 +- .../com/pedro/srt/srt/packets/ControlPacket.kt | 6 +++--- .../java/com/pedro/srt/srt/packets/control/Ack.kt | 12 ++++++------ .../java/com/pedro/srt/srt/packets/control/Nak.kt | 2 +- .../srt/srt/packets/control/handshake/Handshake.kt | 8 +++----- srt/src/main/java/com/pedro/srt/utils/Constants.kt | 1 - .../test/java/com/pedro/srt/srt/control/NakTest.kt | 11 +++++++++++ udp/src/main/java/com/pedro/udp/UdpSender.kt | 2 +- 14 files changed, 47 insertions(+), 41 deletions(-) diff --git a/srt/src/main/java/com/pedro/srt/mpeg2ts/AdaptationField.kt b/srt/src/main/java/com/pedro/srt/mpeg2ts/AdaptationField.kt index 8d96ebee34..25cba4695d 100644 --- a/srt/src/main/java/com/pedro/srt/mpeg2ts/AdaptationField.kt +++ b/srt/src/main/java/com/pedro/srt/mpeg2ts/AdaptationField.kt @@ -16,10 +16,8 @@ package com.pedro.srt.mpeg2ts -import com.pedro.srt.utils.Constants import com.pedro.srt.utils.toInt import java.nio.ByteBuffer -import kotlin.math.pow /** * Created by pedro on 24/8/23. @@ -58,9 +56,9 @@ data class AdaptationField( private val adaptationExtension: ByteArray? = null, private val stuffingBytes: ByteArray? = null, ) { - private val length: Int = calculateSize() private val transportPrivateDataLength: Int = transportPrivateData?.size ?: 0 - + private val length: Int = calculateSize() + fun getData(): ByteArray { val buffer = ByteBuffer.allocate(length) //size after put length @@ -86,6 +84,7 @@ data class AdaptationField( buffer.put(it.size.toByte()) buffer.put(it) } + adaptationExtension?.let { buffer.put(it) } stuffingBytes?.let { buffer.put(it) } return buffer.array() } @@ -100,11 +99,8 @@ data class AdaptationField( } private fun addClockReference(buffer: ByteBuffer, timestamp: Long) { - val pcrBase = - (Constants.SYSTEM_CLOCK_FREQ * timestamp / 1000000 /* µs -> s */ / 300) % 2.toDouble() - .pow(33) - .toLong() - val pcrExt = (Constants.SYSTEM_CLOCK_FREQ * timestamp / 1000000 /* µs -> s */) % 300 + val pcrBase = (timestamp * 9 / 100) % (1L shl 33) + val pcrExt = (timestamp * 27) % 300 /** * PCR Base -> 33 bits diff --git a/srt/src/main/java/com/pedro/srt/mpeg2ts/Pes.kt b/srt/src/main/java/com/pedro/srt/mpeg2ts/Pes.kt index 0c7aa7b748..02c36c86f6 100644 --- a/srt/src/main/java/com/pedro/srt/mpeg2ts/Pes.kt +++ b/srt/src/main/java/com/pedro/srt/mpeg2ts/Pes.kt @@ -16,11 +16,9 @@ package com.pedro.srt.mpeg2ts -import com.pedro.srt.utils.Constants import com.pedro.srt.utils.toInt import java.nio.ByteBuffer import kotlin.experimental.and -import kotlin.math.pow /** * Created by pedro on 28/8/23. @@ -62,10 +60,7 @@ class Pes( } private fun addTimestamp(buffer: ByteBuffer, timestamp: Long, fourBits: Byte) { - val pts = - (Constants.SYSTEM_CLOCK_FREQ * timestamp / 1000000 /* µs -> s */ / 300) % 2.toDouble() - .pow(33) - .toLong() + val pts = (timestamp * 9 / 100) % (1L shl 33) buffer.put((((fourBits and 0xF).toInt() shl 4) or ((pts shr 29) and 0xE).toInt() or 1).toByte()) buffer.putShort((((pts shr 14) and 0xFFFE) or 1).toShort()) diff --git a/srt/src/main/java/com/pedro/srt/mpeg2ts/packets/AacPacket.kt b/srt/src/main/java/com/pedro/srt/mpeg2ts/packets/AacPacket.kt index 8e3e52d275..d5e02f970f 100644 --- a/srt/src/main/java/com/pedro/srt/mpeg2ts/packets/AacPacket.kt +++ b/srt/src/main/java/com/pedro/srt/mpeg2ts/packets/AacPacket.kt @@ -54,7 +54,7 @@ class AacPacket( adts.get(payload, 0, headerSize) fixedBuffer.get(payload, headerSize, length) - val pes = Pes(psiManager.getAudioPid().toInt(), false, PesType.AUDIO, mediaFrame.info.timestamp, ByteBuffer.wrap(payload)) + val pes = Pes(psiManager.getAudioPid().toInt(), true, PesType.AUDIO, mediaFrame.info.timestamp, ByteBuffer.wrap(payload)) val mpeg2tsPackets = mpegTsPacketizer.write(listOf(pes)).chunkPackets(chunkSize).map { buffer -> MpegTsPacket(buffer, MpegType.AUDIO, PacketPosition.SINGLE, isKey = false) } diff --git a/srt/src/main/java/com/pedro/srt/mpeg2ts/packets/OpusPacket.kt b/srt/src/main/java/com/pedro/srt/mpeg2ts/packets/OpusPacket.kt index b22cf173ef..fe4b19e62e 100644 --- a/srt/src/main/java/com/pedro/srt/mpeg2ts/packets/OpusPacket.kt +++ b/srt/src/main/java/com/pedro/srt/mpeg2ts/packets/OpusPacket.kt @@ -51,7 +51,7 @@ class OpusPacket( val pes = Pes(psiManager.getAudioPid().toInt(), true, PesType.PRIVATE_STREAM_1, mediaFrame.info.timestamp, ByteBuffer.wrap(payload)) val mpeg2tsPackets = mpegTsPacketizer.write(listOf(pes)).chunkPackets(chunkSize).map { buffer -> - MpegTsPacket(buffer, MpegType.AUDIO, PacketPosition.SINGLE, true) + MpegTsPacket(buffer, MpegType.AUDIO, PacketPosition.SINGLE, false) } if (mpeg2tsPackets.isNotEmpty()) callback(mpeg2tsPackets) } diff --git a/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt b/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt index f65afd5ce2..e0799a862c 100644 --- a/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt +++ b/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt @@ -55,6 +55,7 @@ class CommandsManager { var audioDisabled = false var videoDisabled = false var host = "" + var latency = 120 //in millis //Avoid write a packet in middle of other. private val writeSync = Mutex(locked = false) private var encryptor: EncryptionUtil? = null @@ -121,6 +122,7 @@ class CommandsManager { ) sequenceNumber++ packetHandlingQueue.add(dataPacket) + dropTooLatePackets(dataPacket.ts) dataPacket.write() socket?.write(dataPacket) return dataPacket.getSize() @@ -153,6 +155,12 @@ class CommandsManager { } } + private fun dropTooLatePackets(nowTs: Int) { + val thresholdUs = latency * 1000 + val firstKept = packetHandlingQueue.indexOfFirst { (nowTs - it.ts) <= thresholdUs } + if (firstKept > 0) packetHandlingQueue.subList(0, firstKept).clear() + } + @Throws(IOException::class) suspend fun writeAck2(ackSequence: Int, socket: SrtSocket?) { writeSync.withLock { diff --git a/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt b/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt index 71ab6af468..30634c4a7e 100644 --- a/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt +++ b/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt @@ -105,7 +105,6 @@ class SrtClient(private val connectChecker: ConnectChecker) { private set var packetsLost = 0 private set - private var latency = 120 //in millis var socketType = SocketType.JAVA var socketTimeout = StreamSocket.DEFAULT_TIMEOUT @@ -128,7 +127,7 @@ class SrtClient(private val connectChecker: ConnectChecker) { } fun setLatency(latency: Int) { - this.latency = latency + commandsManager.latency = latency } fun setDelay(millis: Long) { @@ -211,8 +210,8 @@ class SrtClient(private val connectChecker: ConnectChecker) { val host = urlParser.host val port = urlParser.port ?: 8888 - val path = urlParser.getQuery("streamid") ?: urlParser.getFullPath() - latency = urlParser.getQuery("latency")?.toIntOrNull() ?: latency + val path = urlParser.getQuery("streamid") ?: urlParser.path + commandsManager.latency = urlParser.getQuery("latency")?.toIntOrNull() ?: commandsManager.latency val passphrase = urlParser.getQuery("passphrase") ?: "" if (passphrase.isNotEmpty() && passphrase.length in 10..79) { val encryptionType = when (urlParser.getQuery("pbkeylen")?.toIntOrNull()) { @@ -247,8 +246,8 @@ class SrtClient(private val connectChecker: ConnectChecker) { flags = ExtensionContentFlag.TSBPDSND.value or ExtensionContentFlag.TSBPDRCV.value or ExtensionContentFlag.CRYPT.value or ExtensionContentFlag.TLPKTDROP.value or ExtensionContentFlag.PERIODICNAK.value or ExtensionContentFlag.REXMITFLG.value, - receiverDelay = latency, - senderDelay = latency, + receiverDelay = commandsManager.latency, + senderDelay = commandsManager.latency, path = path, encryptInfo = commandsManager.getEncryptInfo() ))) diff --git a/srt/src/main/java/com/pedro/srt/srt/SrtSender.kt b/srt/src/main/java/com/pedro/srt/srt/SrtSender.kt index 1c9404900d..28b16bc8db 100644 --- a/srt/src/main/java/com/pedro/srt/srt/SrtSender.kt +++ b/srt/src/main/java/com/pedro/srt/srt/SrtSender.kt @@ -112,7 +112,7 @@ class SrtSender( setTrackConfig(!commandsManager.videoDisabled, !commandsManager.audioDisabled) //send config val psiList = mutableListOf(psiManager.getPat()) - psiManager.getPmt()?.let { psiList.add(0, it) } + psiManager.getPmt()?.let { psiList.add(it) } psiList.add(psiManager.getSdt()) val psiPacketsConfig = mpegTsPacketizer.write(psiList).chunkPackets(chunkSize).map { buffer -> MpegTsPacket(buffer, MpegType.PSI, PacketPosition.SINGLE, isKey = false) diff --git a/srt/src/main/java/com/pedro/srt/srt/packets/ControlPacket.kt b/srt/src/main/java/com/pedro/srt/srt/packets/ControlPacket.kt index bed4aa890a..82b563e936 100644 --- a/srt/src/main/java/com/pedro/srt/srt/packets/ControlPacket.kt +++ b/srt/src/main/java/com/pedro/srt/srt/packets/ControlPacket.kt @@ -50,7 +50,7 @@ abstract class ControlPacket( ): SrtPacket() { protected fun writeHeader(ts: Int, socketId: Int) { - val headerData = PacketType.CONTROL.value and 0xff shl 31 or (controlType.value and 0xff shl 16) or subtype.value + val headerData = PacketType.CONTROL.value and 0xff shl 31 or (controlType.value and 0x7fff shl 16) or subtype.value buffer.writeUInt32(headerData) buffer.writeUInt32(typeSpecificInformation) buffer.writeUInt32(ts) @@ -63,7 +63,7 @@ abstract class ControlPacket( if (packetType != PacketType.CONTROL) { throw IOException("error, parsing data packet as control packet") } - controlType = ControlType.from((headerData ushr 16) and 0xFF) + controlType = ControlType.from((headerData ushr 16) and 0x7FFF) val subtypeValue = headerData and 0xFFFF if (subtypeValue == 0) subtype = ControlType.SUB_TYPE else throw IOException("unknown subtype: $subtypeValue") @@ -80,7 +80,7 @@ abstract class ControlPacket( companion object { fun getType(input: InputStream): ControlType { val headerData = input.readUInt32() - return ControlType.from((headerData ushr 16) and 0xFF) + return ControlType.from((headerData ushr 16) and 0x7FFF) } } } \ No newline at end of file diff --git a/srt/src/main/java/com/pedro/srt/srt/packets/control/Ack.kt b/srt/src/main/java/com/pedro/srt/srt/packets/control/Ack.kt index 9de010ed3d..0403e1a9c3 100644 --- a/srt/src/main/java/com/pedro/srt/srt/packets/control/Ack.kt +++ b/srt/src/main/java/com/pedro/srt/srt/packets/control/Ack.kt @@ -56,12 +56,12 @@ class Ack( private fun readBody(input: InputStream) { lastAcknowledgedPacketSequenceNumber = input.readUInt32() and 0x7FFFFFFF //31 bits - rtt = input.readUInt32() - rttVariance = input.readUInt32() - availableBufferSize = input.readUInt32() - packetReceivingRate = input.readUInt32() - estimatedLinkCapacity = input.readUInt32() - receivingRate = input.readUInt32() + if (input.available() >= 4) rtt = input.readUInt32() + if (input.available() >= 4) rttVariance = input.readUInt32() + if (input.available() >= 4) availableBufferSize = input.readUInt32() + if (input.available() >= 4) packetReceivingRate = input.readUInt32() + if (input.available() >= 4) estimatedLinkCapacity = input.readUInt32() + if (input.available() >= 4) receivingRate = input.readUInt32() } override fun toString(): String { diff --git a/srt/src/main/java/com/pedro/srt/srt/packets/control/Nak.kt b/srt/src/main/java/com/pedro/srt/srt/packets/control/Nak.kt index 9738afe96b..f8f4bc5fbf 100644 --- a/srt/src/main/java/com/pedro/srt/srt/packets/control/Nak.kt +++ b/srt/src/main/java/com/pedro/srt/srt/packets/control/Nak.kt @@ -77,7 +77,7 @@ class Nak: ControlPacket(ControlType.NAK) { } fun getNakRanges(): List> { - return cifLostList.chunked(2).map { ranges -> + return cifLostList.chunked(2).filter { it.size == 2 }.map { ranges -> (ranges[0] and 0x7FFFFFFF) to (ranges[1] and 0x7FFFFFFF) } } diff --git a/srt/src/main/java/com/pedro/srt/srt/packets/control/handshake/Handshake.kt b/srt/src/main/java/com/pedro/srt/srt/packets/control/handshake/Handshake.kt index 9dcfc4b881..7645b70f2f 100644 --- a/srt/src/main/java/com/pedro/srt/srt/packets/control/handshake/Handshake.kt +++ b/srt/src/main/java/com/pedro/srt/srt/packets/control/handshake/Handshake.kt @@ -136,11 +136,9 @@ data class Handshake( } private fun readAddress(input: InputStream): String { - val num1 = input.readUInt32LittleEndian() - val num2 = input.readUInt32LittleEndian() - val num3 = input.readUInt32LittleEndian() - val num4 = input.readUInt32LittleEndian() - return "$num1.$num2.$num3.$num4" + val ip = input.readUInt32LittleEndian() + repeat(3) { input.readUInt32LittleEndian() } + return "${(ip ushr 24) and 0xFF}.${(ip ushr 16) and 0xFF}.${(ip ushr 8) and 0xFF}.${ip and 0xFF}" } fun isErrorType(): Boolean { diff --git a/srt/src/main/java/com/pedro/srt/utils/Constants.kt b/srt/src/main/java/com/pedro/srt/utils/Constants.kt index a050775728..ba45b3da55 100644 --- a/srt/src/main/java/com/pedro/srt/utils/Constants.kt +++ b/srt/src/main/java/com/pedro/srt/utils/Constants.kt @@ -21,5 +21,4 @@ package com.pedro.srt.utils */ object Constants { const val MTU = 1500 - const val SYSTEM_CLOCK_FREQ = 27000000 } \ No newline at end of file diff --git a/srt/src/test/java/com/pedro/srt/srt/control/NakTest.kt b/srt/src/test/java/com/pedro/srt/srt/control/NakTest.kt index 658c447b63..b1799acd8e 100644 --- a/srt/src/test/java/com/pedro/srt/srt/control/NakTest.kt +++ b/srt/src/test/java/com/pedro/srt/srt/control/NakTest.kt @@ -77,4 +77,15 @@ class NakTest { assertEquals(listOf(0x7FFFFFFE to 1), packet.getNakRanges()) assertEquals(4, packet.getLostCount()) } + + @Test + fun `GIVEN a truncated nak ending in a range start WHEN get ranges THEN drop the incomplete range without crashing`() { + //header (16 bytes) + a single range-start word (0x80000007) with no range-end → odd-sized cif list + val buffer = byteArrayOf(-128, 3, 0, 0, 0, 0, 0, 0, 0, 0, 9, -60, 0, 0, 0, 64, -128, 0, 0, 7) + val packet = Nak() + packet.read(ByteArrayInputStream(buffer)) + + assertEquals(emptyList>(), packet.getNakRanges()) + assertEquals(0, packet.getLostCount()) + } } \ No newline at end of file diff --git a/udp/src/main/java/com/pedro/udp/UdpSender.kt b/udp/src/main/java/com/pedro/udp/UdpSender.kt index f8dc8d3353..18bf4b1542 100644 --- a/udp/src/main/java/com/pedro/udp/UdpSender.kt +++ b/udp/src/main/java/com/pedro/udp/UdpSender.kt @@ -94,7 +94,7 @@ class UdpSender( setTrackConfig(!commandManager.videoDisabled, !commandManager.audioDisabled) //send config val psiList = mutableListOf(psiManager.getPat()) - psiManager.getPmt()?.let { psiList.add(0, it) } + psiManager.getPmt()?.let { psiList.add(it) } psiList.add(psiManager.getSdt()) val psiPacketsConfig = mpegTsPacketizer.write(psiList).chunkPackets(chunkSize).map { b -> MpegTsPacket(b, MpegType.PSI, PacketPosition.SINGLE, isKey = false) From 2662daf358c2013c1ebada654d8e9135055332e4 Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Fri, 3 Jul 2026 23:38:26 +0200 Subject: [PATCH 10/10] fix tests --- .../com/pedro/rtmp/flv/audio/AacPacketTest.kt | 23 +++++++++++-------- .../pedro/rtmp/flv/audio/OpusPacketTest.kt | 22 ++++++++++-------- 2 files changed, 26 insertions(+), 19 deletions(-) diff --git a/rtmp/src/test/java/com/pedro/rtmp/flv/audio/AacPacketTest.kt b/rtmp/src/test/java/com/pedro/rtmp/flv/audio/AacPacketTest.kt index 03a0cf35cf..35a72cf6ea 100644 --- a/rtmp/src/test/java/com/pedro/rtmp/flv/audio/AacPacketTest.kt +++ b/rtmp/src/test/java/com/pedro/rtmp/flv/audio/AacPacketTest.kt @@ -17,6 +17,7 @@ package com.pedro.rtmp.flv.audio import com.pedro.common.frame.MediaFrame +import com.pedro.rtmp.flv.FlvPacket import com.pedro.rtmp.flv.FlvType import com.pedro.rtmp.flv.audio.packet.AacPacket import kotlinx.coroutines.test.runTest @@ -30,23 +31,25 @@ import java.nio.ByteBuffer class AacPacketTest { @Test - fun `GIVEN an AAC buffer WHEN call create an AAC packet 2 times THEN return config and expected buffer`() = runTest { + fun `GIVEN an AAC buffer WHEN call create an AAC packet THEN return config and expected buffer`() = runTest { val timestamp = 123456789L val buffer = ByteArray(256) { 0x00 } val info = MediaFrame.Info(0, buffer.size, timestamp, false) val mediaFrame = MediaFrame(ByteBuffer.wrap(buffer), info, MediaFrame.Type.AUDIO) val aacPacket = AacPacket() aacPacket.sendAudioInfo(32000, true) + val packets = mutableListOf() aacPacket.createFlvPacket(mediaFrame) { flvPacket -> - assertEquals(FlvType.AUDIO, flvPacket.type) - assertEquals((-81).toByte(), flvPacket.buffer[0]) - assertEquals(AacPacket.Type.SEQUENCE.mark, flvPacket.buffer[1]) - } - aacPacket.createFlvPacket(mediaFrame) { flvPacket -> - assertEquals(FlvType.AUDIO, flvPacket.type) - assertEquals((-81).toByte(), flvPacket.buffer[0]) - assertEquals(AacPacket.Type.RAW.mark, flvPacket.buffer[1]) - assertEquals(buffer.size + 2, flvPacket.length) + packets.add(flvPacket) } + assertEquals(FlvType.AUDIO, packets[0].type) + assertEquals((-81).toByte(), packets[0].buffer[0]) + assertEquals(AacPacket.Type.SEQUENCE.mark, packets[0].buffer[1]) + assertEquals(4, packets[0].length) + + assertEquals(FlvType.AUDIO, packets[1].type) + assertEquals((-81).toByte(), packets[1].buffer[0]) + assertEquals(AacPacket.Type.RAW.mark, packets[1].buffer[1]) + assertEquals(buffer.size + 2, packets[1].length) } } \ No newline at end of file diff --git a/rtmp/src/test/java/com/pedro/rtmp/flv/audio/OpusPacketTest.kt b/rtmp/src/test/java/com/pedro/rtmp/flv/audio/OpusPacketTest.kt index 5a0bdfb485..209eba9223 100644 --- a/rtmp/src/test/java/com/pedro/rtmp/flv/audio/OpusPacketTest.kt +++ b/rtmp/src/test/java/com/pedro/rtmp/flv/audio/OpusPacketTest.kt @@ -17,6 +17,7 @@ package com.pedro.rtmp.flv.audio import com.pedro.common.frame.MediaFrame +import com.pedro.rtmp.flv.FlvPacket import com.pedro.rtmp.flv.FlvType import com.pedro.rtmp.flv.audio.packet.OpusPacket import kotlinx.coroutines.test.runTest @@ -30,22 +31,25 @@ import java.nio.ByteBuffer class OpusPacketTest { @Test - fun `GIVEN an Opus buffer WHEN call create an Opus packet 2 times THEN return config and expected buffer`() = runTest { + fun `GIVEN an Opus buffer WHEN call create an Opus packet THEN return config and expected buffer`() = runTest { val timestamp = 123456789L val buffer = ByteArray(256) { 0x00 } val info = MediaFrame.Info(0, buffer.size, timestamp, false) val mediaFrame = MediaFrame(ByteBuffer.wrap(buffer), info, MediaFrame.Type.AUDIO) val opusPacket = OpusPacket() opusPacket.sendAudioInfo(48000, true) + val packets = mutableListOf() opusPacket.createFlvPacket(mediaFrame) { flvPacket -> - assertEquals(FlvType.AUDIO, flvPacket.type) - assertEquals(((AudioFormat.EX_HEADER.value shl 4) or (AudioFourCCPacketType.SEQUENCE_START.value and 0x0F)).toByte(), flvPacket.buffer[0]) - assertEquals(5 + 19, flvPacket.length) - } - opusPacket.createFlvPacket(mediaFrame) { flvPacket -> - assertEquals(FlvType.AUDIO, flvPacket.type) - assertEquals(((AudioFormat.EX_HEADER.value shl 4) or (AudioFourCCPacketType.CODED_FRAMES.value and 0x0F)).toByte(), flvPacket.buffer[0]) - assertEquals(5 + buffer.size, flvPacket.length) + packets.add(flvPacket) } + assertEquals(2, packets.size) + + assertEquals(FlvType.AUDIO, packets[0].type) + assertEquals(((AudioFormat.EX_HEADER.value shl 4) or (AudioFourCCPacketType.SEQUENCE_START.value and 0x0F)).toByte(), packets[0].buffer[0]) + assertEquals(5 + 19, packets[0].length) + + assertEquals(FlvType.AUDIO, packets[1].type) + assertEquals(((AudioFormat.EX_HEADER.value shl 4) or (AudioFourCCPacketType.CODED_FRAMES.value and 0x0F)).toByte(), packets[1].buffer[0]) + assertEquals(5 + buffer.size, packets[1].length) } } \ No newline at end of file