Skip to content

Refactoring srt commands #2205

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pedroSG94 merged 4 commits into from
Sep 28, 2026
Merged

Refactoring srt commands #2205

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a218145
remove redundant code
pedroSG94 Sep 22, 2026
d26c52d
refactoring srt commands
pedroSG94 Sep 22, 2026
c72bf1b
move all retransmission to other class
pedroSG94 Sep 22, 2026
5f23924
move all retransmission to other class 2
pedroSG94 Sep 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 1 addition & 5 deletions encoder/src/main/java/com/pedro/encoder/audio/AudioEncoder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,11 +167,7 @@ protected long calculatePts(Frame frame, long presentTimeUs) {
@Override
protected boolean checkBuffer(@NonNull ByteBuffer byteBuffer,
@NonNull MediaCodec.BufferInfo bufferInfo) {
// Codec config (e.g. AAC AudioSpecificConfig) is signalled per protocol and via
// MediaFormat csd-0 for the muxer; do not forward it as an encoded audio frame.
if ((bufferInfo.flags & MediaCodec.BUFFER_FLAG_CODEC_CONFIG) != 0) {
return false;
}
if ((bufferInfo.flags & MediaCodec.BUFFER_FLAG_CODEC_CONFIG) != 0) return false;
return checkValidTimeStamp(bufferInfo);
}

Expand Down
14 changes: 6 additions & 8 deletions library/src/main/java/com/pedro/library/util/AacMuxerRecordController.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,14 +119,12 @@ class AacMuxerRecordController : AsyncBaseRecordController() {

private suspend fun write(byteBuffer: ByteBuffer, info: MediaCodec.BufferInfo) {
try {
if ((info.flags and MediaCodec.BUFFER_FLAG_CODEC_CONFIG) != MediaCodec.BUFFER_FLAG_CODEC_CONFIG) {
val header = createAdtsHeader(2, info.size - info.offset, sampleRate, channels).array()
outputStream?.write(header)
val data = ByteArray(byteBuffer.remaining())
byteBuffer.get(data)
outputStream?.write(data)
bitrateManager?.calculateBitrate(info.size * 8L)
}
val header = createAdtsHeader(2, info.size - info.offset, sampleRate, channels).array()
outputStream?.write(header)
val data = ByteArray(byteBuffer.remaining())
byteBuffer.get(data)
outputStream?.write(data)
bitrateManager?.calculateBitrate(info.size * 8L)
} catch (e: Exception) {
listener?.onError(e)
}
Expand Down
169 changes: 16 additions & 153 deletions srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,18 +36,14 @@ import com.pedro.srt.utils.SrtSocket
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import java.io.IOException
import kotlin.math.max
import kotlin.math.min
import kotlin.random.Random

/**
* Created by pedro on 23/8/23.
*/
class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurrentTimeMicro() }) {
class CommandsManager {

private val TAG = "CommandsManager"
//used for packet lost
private val packetHandlingQueue = mutableListOf<DataPacket>()

var sequenceNumber: Int = generateInitialSequence()
var messageNumber = 1
Expand All @@ -60,26 +56,17 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr
var videoDisabled = false
var host = ""
var latency = 120 //in millis
/**
* Max retransmit bandwidth as a percentage of the estimated media rate.
* Values <= 0 disable the limit (legacy behavior).
*/
var retransmitOverheadPercent: Int = 25
//Avoid write a packet in middle of other.
private val writeSync = Mutex(locked = false)
private var encryptor: EncryptionUtil? = null
var videoCodec = VideoCodec.H264
var audioCodec = AudioCodec.AAC
private val packetRetransmission = PacketRetransmission()

private var rtt = 0
private var rttVariance = 0
private var mediaBytesPerSecond = 0.0
private var mediaWindowStartUs = 0L
private var mediaWindowBytes = 0L
private var retransmitTokens = 0.0
private var lastTokenRefillUs = 0L
private var bucketInitialized = false

fun setRetransmitOverhead(percent: Int) {
packetRetransmission.retransmitOverheadPercent = percent
}

fun setPassphrase(passphrase: String, type: EncryptionType) {
encryptor = if (passphrase.isEmpty() || type == EncryptionType.NONE) null else EncryptionUtil(type, passphrase)
}
Expand All @@ -95,19 +82,16 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr
fun encryptionEnabled() = encryptor != null

fun loadStartTs() {
startTS = timeProvider()
startTS = TimeUtils.getCurrentTimeMicro()
localSocketId = generateSocketId()
}

fun getTs(): Int {
return (timeProvider() - startTS).toInt()
return (TimeUtils.getCurrentTimeMicro() - startTS).toInt()
}

suspend fun updateRtt(rtt: Int, rttVariance: Int) {
writeSync.withLock {
this.rtt = rtt
this.rttVariance = rttVariance
}
fun updateRtt(rtt: Int, rttVariance: Int) {
packetRetransmission.updateRtt(rtt, rttVariance, latency)
}

@Throws(IOException::class)
Expand Down Expand Up @@ -148,10 +132,10 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr
socketId = socketId
)
sequenceNumber++
packetHandlingQueue.add(dataPacket)
dropTooLatePackets(dataPacket.ts)
trackMediaBytes(packet.buffer.size, timeProvider())
packetRetransmission.addPacket(dataPacket)
packetRetransmission.dropTooLatePackets(dataPacket.ts, latency)
dataPacket.write()
packetRetransmission.trackMediaBytes(dataPacket.getSize(), TimeUtils.getCurrentTimeMicro())
socket?.write(dataPacket)
return dataPacket.getSize()
}
Expand All @@ -160,119 +144,16 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr
@Throws(IOException::class)
suspend fun reSendPackets(lostRanges: List<Pair<Int, Int>>, socket: SrtSocket?): Int {
writeSync.withLock {
val unlimited = retransmitOverheadPercent <= 0
val nowTs = getTs()
val nowUs = timeProvider()
if (!unlimited) {
refillRetransmitTokens(nowUs)
}
val latencyUs = latency * 1000
val minResendInterval = if (unlimited) 0 else {
min(max(rtt + 4 * rttVariance, MIN_RESEND_INTERVAL_US), latencyUs / 4)
}
var newlyReported = 0
var budgetExhausted = false
for (packet in packetHandlingQueue) {
if (!isInLostRange(packet.sequenceNumber, lostRanges)) continue
if (!packet.nakReported) {
packet.nakReported = true
newlyReported++
}
if (!unlimited) {
if ((nowTs - packet.ts + rtt / 2) >= latencyUs) continue
// The gate only suppresses repeated reports of a packet that was already retransmitted;
// the first NAK is always honored even when it arrives within minResendInterval of the original send.
if (packet.retransmitted && (nowTs - packet.lastSentTs) < minResendInterval) continue
if (budgetExhausted) continue
val packetSize = dataPacketWireSize(packet)
if (retransmitTokens < packetSize) {
budgetExhausted = true
continue
}
retransmitTokens -= packetSize
}
packet.retransmitted = true
packet.write()
socket?.write(packet)
packet.lastSentTs = nowTs
}
return newlyReported
return packetRetransmission.reSendPackets(lostRanges, socket, getTs(), latency, MTU)
}
}

suspend fun updateHandlingQueue(lastPacketSequence: Int) {
writeSync.withLock {
packetHandlingQueue.removeAll {
//discard confirmed packets
val diff = (lastPacketSequence - it.sequenceNumber) and 0x7FFFFFFF
diff in 1 until 0x40000000
}
}
}

private fun isInLostRange(sequenceNumber: Int, lostRanges: List<Pair<Int, Int>>): Boolean {
return lostRanges.any { (min, max) ->
((sequenceNumber - min) and 0x7FFFFFFF) <= ((max - min) and 0x7FFFFFFF)
packetRetransmission.updateHandlingQueue(lastPacketSequence)
}
}

private fun trackMediaBytes(bytes: Int, nowUs: Long) {
if (mediaWindowStartUs == 0L) mediaWindowStartUs = nowUs
mediaWindowBytes += bytes
val elapsed = nowUs - mediaWindowStartUs
if (elapsed >= MEDIA_RATE_WINDOW_US) {
val rate = mediaWindowBytes.toDouble() * MEDIA_RATE_WINDOW_US / elapsed
mediaBytesPerSecond = if (mediaBytesPerSecond == 0.0) rate else {
mediaBytesPerSecond * MEDIA_RATE_EWMA_OLD + rate * MEDIA_RATE_EWMA_NEW
}
mediaWindowStartUs = nowUs
mediaWindowBytes = 0
}
}

private fun getRetransmitRate(): Double {
val percent = retransmitOverheadPercent
val mediaRate = if (mediaBytesPerSecond > 0.0) mediaBytesPerSecond else MIN_RETRANSMIT_BYTES_PER_SECOND.toDouble()
return max(mediaRate * percent / 100.0, MIN_RETRANSMIT_BYTES_PER_SECOND.toDouble())
}

private fun getRetransmitCapacity(rate: Double): Int {
// Allow an immediate burst up to half a second of media so short loss events
// (e.g. a brief link flap) are not retried one packet at a time on healthy links.
val burstCapacity = if (mediaBytesPerSecond > 0.0) {
(mediaBytesPerSecond * RETRANSMIT_BURST_WINDOW_US / MEDIA_RATE_WINDOW_US).toInt()
} else {
0
}
return max(burstCapacity, max((rate * latency / 1000.0).toInt(), MTU))
}

private fun refillRetransmitTokens(nowUs: Long) {
if (!bucketInitialized) {
val rate = getRetransmitRate()
retransmitTokens = getRetransmitCapacity(rate).toDouble()
lastTokenRefillUs = nowUs
bucketInitialized = true
return
}
val elapsedUs = nowUs - lastTokenRefillUs
if (elapsedUs <= 0) return
val rate = getRetransmitRate()
val capacity = getRetransmitCapacity(rate)
retransmitTokens = min(retransmitTokens + rate * elapsedUs / MEDIA_RATE_WINDOW_US, capacity.toDouble())
lastTokenRefillUs = nowUs
}

private fun dataPacketWireSize(packet: DataPacket): Int {
return packet.payload.size + DATA_HEADER_SIZE
}

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 {
Expand Down Expand Up @@ -307,15 +188,7 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr
socketId = 0
startTS = 0L
host = ""
packetHandlingQueue.clear()
rtt = 0
rttVariance = 0
mediaBytesPerSecond = 0.0
mediaWindowStartUs = 0L
mediaWindowBytes = 0L
retransmitTokens = 0.0
lastTokenRefillUs = 0L
bucketInitialized = false
packetRetransmission.reset()
}

private fun generateInitialSequence(): Int {
Expand All @@ -325,14 +198,4 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr
private fun generateSocketId(): Int {
return Random.nextInt(1, Int.MAX_VALUE)
}

companion object {
private const val DATA_HEADER_SIZE = 16
private const val MIN_RETRANSMIT_BYTES_PER_SECOND = 8_000
private const val MIN_RESEND_INTERVAL_US = 20_000
private const val MEDIA_RATE_WINDOW_US = 1_000_000L
private const val RETRANSMIT_BURST_WINDOW_US = 500_000L
private const val MEDIA_RATE_EWMA_OLD = 0.8
private const val MEDIA_RATE_EWMA_NEW = 0.2
}
}
Loading
Loading