diff --git a/.gitignore b/.gitignore index 7552331786..cf033405f8 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,5 @@ /build .DS_Store /.kotlin -/.claude \ No newline at end of file +/.claude +/common/.cxx/ \ No newline at end of file diff --git a/common/build.gradle.kts b/common/build.gradle.kts index a3142e73d5..8c4efc4c0e 100644 --- a/common/build.gradle.kts +++ b/common/build.gradle.kts @@ -6,6 +6,10 @@ 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 @@ -13,6 +17,12 @@ android { minSdk = 16 lint.targetSdk = 37 } + externalNativeBuild { + cmake { + path = file("src/main/cpp/CMakeLists.txt") + version = "3.22.1" + } + } testOptions { unitTests.isReturnDefaultValues = true } diff --git a/common/src/main/cpp/CMakeLists.txt b/common/src/main/cpp/CMakeLists.txt new file mode 100644 index 0000000000..52ac9eccd0 --- /dev/null +++ b/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 +) diff --git a/common/src/main/cpp/tcp_info.cpp b/common/src/main/cpp/tcp_info.cpp new file mode 100644 index 0000000000..3a5da2e68b --- /dev/null +++ b/common/src/main/cpp/tcp_info.cpp @@ -0,0 +1,28 @@ +#include +#include +#include +#include +#include + +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(info.tcpi_rtt); +} diff --git a/common/src/main/java/com/pedro/common/socket/base/TcpStreamSocket.kt b/common/src/main/java/com/pedro/common/socket/base/TcpStreamSocket.kt index b4ddef4ab8..dffcb834fa 100644 --- a/common/src/main/java/com/pedro/common/socket/base/TcpStreamSocket.kt +++ b/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) diff --git a/common/src/main/java/com/pedro/common/socket/java/TcpInfo.kt b/common/src/main/java/com/pedro/common/socket/java/TcpInfo.kt new file mode 100644 index 0000000000..c2cddde9ed --- /dev/null +++ b/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 +} 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 0a0d6942ed..5d017b3bc9 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 @@ -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,6 +10,7 @@ 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() @@ -15,19 +18,36 @@ abstract class TcpStreamSocketJavaBase: TcpStreamSocket() { 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) } diff --git a/library/src/main/java/com/pedro/library/util/streamclient/GenericStreamClient.kt b/library/src/main/java/com/pedro/library/util/streamclient/GenericStreamClient.kt index 1df58c19b3..9ddfc2a54d 100644 --- a/library/src/main/java/com/pedro/library/util/streamclient/GenericStreamClient.kt +++ b/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) diff --git a/library/src/main/java/com/pedro/library/util/streamclient/RtmpStreamClient.kt b/library/src/main/java/com/pedro/library/util/streamclient/RtmpStreamClient.kt index e36a8e3de2..d85eda9e11 100644 --- a/library/src/main/java/com/pedro/library/util/streamclient/RtmpStreamClient.kt +++ b/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. 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 43f3d30ce8..97401d19fb 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt +++ b/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 */ diff --git a/rtmp/src/main/java/com/pedro/rtmp/utils/socket/TcpSocket.kt b/rtmp/src/main/java/com/pedro/rtmp/utils/socket/TcpSocket.kt index 58186e1558..fd0b2920ec 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/utils/socket/TcpSocket.kt +++ b/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) }