Expose kernel TCP RTT for Java RTMP sockets

.gitignore

@@ -5,4 +5,5 @@
 /build
 .DS_Store
 /.kotlin
-/.claude
+/.claude
+/common/.cxx/

common/build.gradle.kts

@@ -6,13 +6,23 @@ plugins {
 
 android {
     namespace = "com.pedro.common"
+    // The selected NDK supports API 21+, while common retains minSdk 16.
+    // This is safe because the JNI shim is loaded only on API 29+.
+    ndkVersion = "28.2.13676358"
+    experimentalProperties["android.ndk.suppressMinSdkVersionError"] = 21
     //noinspection GradleDependency
     compileSdk = 35
 
     defaultConfig {
         minSdk = 16
         lint.targetSdk = 37
     }
+    externalNativeBuild {
+        cmake {
+            path = file("src/main/cpp/CMakeLists.txt")
+            version = "3.22.1"
+        }
+    }
     testOptions {
         unitTests.isReturnDefaultValues = true
     }

common/src/main/cpp/CMakeLists.txt

@@ -0,0 +1,9 @@
+cmake_minimum_required(VERSION 3.22.1)
+
+project(rootencoder_common)
+
+add_library(
+    rootencoder_common
+    SHARED
+    tcp_info.cpp
+)

common/src/main/cpp/tcp_info.cpp

@@ -0,0 +1,28 @@
+#include <cstddef>
+#include <jni.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <sys/socket.h>
+
+extern "C" JNIEXPORT jlong JNICALL
+Java_com_pedro_common_socket_java_TcpInfo_nativeGetRttMicros(
+        JNIEnv*, jobject, jint fd) {
+    tcp_info info{};
+    socklen_t length = sizeof(info);
+
+    if (getsockopt(fd, IPPROTO_TCP, TCP_INFO, &info, &length) != 0) {
+        return -1;
+    }
+
+    constexpr size_t requiredLength =
+        offsetof(tcp_info, tcpi_rtt) + sizeof(info.tcpi_rtt);
+    if (length < requiredLength) {
+        return -1;
+    }
+
+    if (info.tcpi_state != TCP_ESTABLISHED) {
+        return -1;
+    }
+
+    return static_cast<jlong>(info.tcpi_rtt);
+}

common/src/main/java/com/pedro/common/socket/base/TcpStreamSocket.kt

@@ -1,6 +1,13 @@
 package com.pedro.common.socket.base
 
 abstract class TcpStreamSocket: StreamSocket() {
+    /**
+     * Returns the smoothed TCP round-trip time reported by the kernel, in microseconds.
+     *
+     * This is only available for eligible socket implementations that expose TCP_INFO.
+     */
+    open fun getTcpRtt(): Long? = null
+
     abstract suspend fun write(bytes: ByteArray)
     abstract suspend fun write(bytes: ByteArray, offset: Int, size: Int)
     abstract suspend fun write(b: Int)

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

@@ -0,0 +1,20 @@
+package com.pedro.common.socket.java
+
+import androidx.annotation.Keep
+
+@Keep
+internal object TcpInfo {
+
+    private val loaded by lazy {
+        runCatching { System.loadLibrary("rootencoder_common") }.isSuccess
+    }
+
+    fun getRttMicros(fd: Int): Long? {
+        if (!loaded) return null
+        return runCatching { nativeGetRttMicros(fd) }
+            .getOrNull()
+            ?.takeIf { it > 0 }
+    }
+
+    private external fun nativeGetRttMicros(fd: Int): Long
+}

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

@@ -1,5 +1,7 @@
 package com.pedro.common.socket.java
 
+import android.os.Build
+import android.os.ParcelFileDescriptor
 import com.pedro.common.readUntil
 import com.pedro.common.socket.base.TcpStreamSocket
 import java.io.ByteArrayInputStream
@@ -8,26 +10,44 @@ import java.net.Socket
 
 abstract class TcpStreamSocketJavaBase: TcpStreamSocket() {
 
+    private val socketLock = Any()
     private var socket = Socket()
     private var input = ByteArrayInputStream(byteArrayOf()).buffered()
     private var output = ByteArrayOutputStream().buffered()
 
     abstract fun onConnectSocket(timeout: Long): Socket
 
     override suspend fun connect() {
-        socket = onConnectSocket(timeout)
-        output = socket.getOutputStream().buffered()
-        input = socket.getInputStream().buffered()
+        val connectedSocket = onConnectSocket(timeout)
+        synchronized(socketLock) {
+            socket = connectedSocket
+            output = socket.getOutputStream().buffered()
+            input = socket.getInputStream().buffered()
+        }
     }
 
     override suspend fun close() {
-        if (socket.isConnected) {
-            runCatching { socket.shutdownOutput() }
-            runCatching { socket.shutdownInput() }
-            runCatching { socket.close() }
+        synchronized(socketLock) {
+            if (socket.isConnected) {
+                runCatching { socket.shutdownOutput() }
+                runCatching { socket.shutdownInput() }
+                runCatching { socket.close() }
+            }
         }
     }
 
+    override fun getTcpRtt(): Long? {
+        // Before API 29, closing the ParcelFileDescriptor can close the live stream socket.
+        if (Build.VERSION.SDK_INT < Build.VERSION_CODES.Q) return null
+        val descriptor = synchronized(socketLock) {
+            if (!socket.isConnected || socket.isClosed) return null
+            runCatching { ParcelFileDescriptor.fromSocket(socket) }.getOrNull()
+        } ?: return null
+        return runCatching {
+            descriptor.use { TcpInfo.getRttMicros(it.fd) }
+        }.getOrNull()
+    }
+
     override suspend fun write(bytes: ByteArray) {
         output.write(bytes)
     }

library/src/main/java/com/pedro/library/util/streamclient/GenericStreamClient.kt

@@ -123,6 +123,13 @@ class GenericStreamClient(
     rtmpClient.setWriteChunkSize(chunkSize)
   }
 
+  /**
+   * Smoothed TCP round-trip time reported by the kernel, in microseconds.
+   *
+   * Returns null unless the active client is an eligible RTMP/RTMPS connection.
+   */
+  fun getTcpRtt(): Long? = (connectedStreamClient as? RtmpStreamClient)?.getTcpRtt()
+
   override fun setAuthorization(user: String?, password: String?) {
     rtmpClient.setAuthorization(user, password)
     rtspClient.setAuthorization(user, password)

library/src/main/java/com/pedro/library/util/streamclient/RtmpStreamClient.kt

@@ -88,6 +88,15 @@ class RtmpStreamClient(
    */
   fun getRtt() = rtmpClient.rtt
 
+  /**
+   * Smoothed TCP round-trip time reported by the kernel, in microseconds.
+   *
+   * Returns null when no eligible local socket exists, the platform is unsupported, or the query
+   * fails. This is currently available for direct RTMP/RTMPS connections using [SocketType.JAVA]
+   * on Android 10 or newer.
+   */
+  fun getTcpRtt(): Long? = rtmpClient.getTcpRtt()
+
   /**
    * Send ping commands each second to server.
    * This allow get a RTT and keep alive the read channel in servers that close it due to inactivity.

rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt

@@ -69,6 +69,7 @@ class RtmpClient(private val connectChecker: ConnectChecker) {
 
   private val validSchemes = arrayOf("rtmp", "rtmps", "rtmpt", "rtmpts")
 
+  @Volatile
   private var socket: RtmpSocket? = null
   private var scope = CoroutineScope(Dispatchers.IO)
   private var scopeRetry = CoroutineScope(Dispatchers.IO)
@@ -116,6 +117,19 @@ class RtmpClient(private val connectChecker: ConnectChecker) {
     private set
   private val pingTs = AtomicLong(0)
 
+  /**
+   * Smoothed TCP round-trip time reported by the kernel, in microseconds.
+   *
+   * Returns null when no eligible local socket exists, the platform is unsupported, or the query
+   * fails. This is currently available for direct RTMP/RTMPS connections using [SocketType.JAVA]
+   * on Android 10 or newer.
+   */
+  fun getTcpRtt(): Long? {
+    val currentSocket = socket as? TcpSocket ?: return null
+    val rtt = currentSocket.getTcpRtt()
+    return rtt.takeIf { socket === currentSocket }
+  }
+
   /**
    * Add certificates for TLS connection
    */

rtmp/src/main/java/com/pedro/rtmp/utils/socket/TcpSocket.kt

@@ -52,6 +52,8 @@ class TcpSocket(
 
   override fun isReachable(): Boolean = socket.isReachable()
 
+  fun getTcpRtt(): Long? = socket.getTcpRtt()
+
   override suspend fun write(b: Int) {
     socket.write(b)
   }