fixing common and rtsp errors

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

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() {

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) {
@@ -123,16 +125,16 @@ 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 {
   val md: MessageDigest
   try {
     md = MessageDigest.getInstance("MD5")
     return md.digest(toByteArray()).bytesToHex()
-  } catch (ignore: NoSuchAlgorithmException) {
-  } catch (ignore: UnsupportedEncodingException) {
+  } catch (_: NoSuchAlgorithmException) {
+  } catch (_: UnsupportedEncodingException) {
   }
   return ""
 }
@@ -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()
 }
 

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>): 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 }

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<MediaFrame>((cache / 5).toInt()) { p0, p1 ->
+        if (cacheTime == 0L) return
+        cacheQueue = PriorityBlockingQueue<MediaFrame>(maxOf(1, (cache / 5).toInt())) { p0, p1 ->
             p0.info.timestamp.compare(p1.info.timestamp)
         }
     }

common/src/main/java/com/pedro/common/UrlParser.kt

@@ -57,12 +57,12 @@ 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
   }
 
   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
   }
 

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,21 +91,21 @@ 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
         return size >= capacity * (percentUsed / 100f)
     }
 
     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
     }
 
-    fun getCacheSize(): Int = cacheSize
+    fun getCacheSize(): Int = queue.capacity
 
     fun getItemsInCache(): Int = queue.getSize()
 

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
   }
 

common/src/main/java/com/pedro/common/socket/java/TcpStreamSocketJavaBase.kt

@@ -2,24 +2,20 @@ 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() {
 
     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,9 +58,22 @@ 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
+    override fun isConnected(): Boolean = socket.isConnected && !socket.isClosed
 
     override fun isReachable(): Boolean = socket.inetAddress?.isReachable(timeout.toInt()) ?: false
 }

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
 

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 }

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
 }

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)

common/src/test/java/com/pedro/common/BitBufferTest.kt

@@ -182,39 +182,39 @@ 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)
   }
 
   @Test
   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)
   }
 
   @Test
   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)
   }
 

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())
+  }
 }