From 2fdac7567cc8374f95519c0fa9df3800a783a1f1 Mon Sep 17 00:00:00 2001 From: Aubrey Haskett Date: Wed, 14 Oct 2026 14:19:47 -0500 Subject: [PATCH 1/2] Expose active RTMP Java socket --- .../common/socket/java/TcpStreamSocketJavaBase.kt | 9 +++++++++ .../library/util/streamclient/GenericStreamClient.kt | 12 ++++++++++++ .../library/util/streamclient/RtmpStreamClient.kt | 4 ++++ rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt | 8 ++++++++ .../java/com/pedro/rtmp/utils/socket/TcpSocket.kt | 5 +++++ 5 files changed, 38 insertions(+) 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..bd31653287 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 @@ -8,12 +8,21 @@ import java.net.Socket abstract class TcpStreamSocketJavaBase: TcpStreamSocket() { + @Volatile private var socket = Socket() private var input = ByteArrayInputStream(byteArrayOf()).buffered() private var output = ByteArrayOutputStream().buffered() abstract fun onConnectSocket(timeout: Long): Socket + /** + * Returns the current Java socket for read-only diagnostics. + * + * The socket can be replaced or closed concurrently during a reconnect. Callers must not + * close it or change its configuration. + */ + fun getJavaSocket(): Socket? = socket.takeIf { it.isConnected && !it.isClosed } + override suspend fun connect() { socket = onConnectSocket(timeout) output = socket.getOutputStream().buffered() 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..2f956fb6ab 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 @@ -19,6 +19,7 @@ package com.pedro.library.util.streamclient import com.pedro.common.socket.base.SocketType import com.pedro.rtsp.rtsp.Protocol import com.pedro.srt.srt.packets.control.handshake.EncryptionType +import java.net.Socket import javax.net.ssl.TrustManager /** @@ -180,6 +181,17 @@ class GenericStreamClient( override fun getBytesSend(): Long = connectedStreamClient?.getBytesSend() ?: 0 + /** + * Returns the active direct RTMP Java socket for read-only diagnostics. + * + * Returns null when another protocol or socket implementation is active. + */ + fun getJavaSocket(): Socket? = if (connectedStreamClient === rtmpClient) { + rtmpClient.getJavaSocket() + } else { + null + } + override fun getDroppedAudioFrames(): Long = connectedStreamClient?.getDroppedAudioFrames() ?: 0 override fun getDroppedVideoFrames(): Long = connectedStreamClient?.getDroppedVideoFrames() ?: 0 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..f60c7775d7 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 @@ -18,6 +18,7 @@ package com.pedro.library.util.streamclient import com.pedro.common.socket.base.SocketType import com.pedro.rtmp.rtmp.RtmpClient +import java.net.Socket import javax.net.ssl.TrustManager /** @@ -88,6 +89,9 @@ class RtmpStreamClient( */ fun getRtt() = rtmpClient.rtt + /** Returns the active direct RTMP Java socket for read-only diagnostics. */ + fun getJavaSocket(): Socket? = rtmpClient.getJavaSocket() + /** * 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..5bb75c4df8 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt @@ -55,6 +55,7 @@ import kotlinx.coroutines.launch import kotlinx.coroutines.withTimeoutOrNull import java.io.IOException import java.net.URISyntaxException +import java.net.Socket import java.nio.ByteBuffer import java.util.concurrent.atomic.AtomicLong import javax.net.ssl.TrustManager @@ -150,6 +151,13 @@ class RtmpClient(private val connectChecker: ConnectChecker) { shouldSendPings = enabled } + /** + * Returns the active direct RTMP Java socket for read-only diagnostics. + * + * The socket can become stale during reconnect. Returns null for tunneled or non-Java sockets. + */ + fun getJavaSocket(): Socket? = (socket as? TcpSocket)?.getJavaSocket() + /** * Must be called before connect */ 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..4e117805c8 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 @@ -18,10 +18,12 @@ package com.pedro.rtmp.utils.socket import com.pedro.common.socket.base.SocketType import com.pedro.common.socket.base.StreamSocket +import com.pedro.common.socket.java.TcpStreamSocketJavaBase import com.pedro.common.toUInt16 import com.pedro.common.toUInt24 import com.pedro.common.toUInt32 import com.pedro.common.toUInt32LittleEndian +import java.net.Socket import javax.net.ssl.TrustManager /** @@ -52,6 +54,9 @@ class TcpSocket( override fun isReachable(): Boolean = socket.isReachable() + /** Returns the active Java socket for read-only diagnostics, or null for non-Java sockets. */ + fun getJavaSocket(): Socket? = (socket as? TcpStreamSocketJavaBase)?.getJavaSocket() + override suspend fun write(b: Int) { socket.write(b) } From 535677ce273e669a8efd9f7b15d38a52fc395cc8 Mon Sep 17 00:00:00 2001 From: Aubrey Haskett Date: Wed, 14 Oct 2026 14:43:28 -0500 Subject: [PATCH 2/2] Address Java socket API review feedback --- .../socket/java/TcpStreamSocketJavaBase.kt | 5 +- .../java/TcpStreamSocketJavaBaseTest.kt | 59 +++++++++++++++++++ .../util/streamclient/GenericStreamClient.kt | 4 +- .../util/streamclient/RtmpStreamClient.kt | 8 ++- .../java/com/pedro/rtmp/rtmp/RtmpClient.kt | 5 +- .../com/pedro/rtmp/utils/socket/TcpSocket.kt | 7 ++- 6 files changed, 82 insertions(+), 6 deletions(-) create mode 100644 common/src/test/java/com/pedro/common/socket/java/TcpStreamSocketJavaBaseTest.kt 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 bd31653287..439ef68a89 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 @@ -18,8 +18,9 @@ abstract class TcpStreamSocketJavaBase: TcpStreamSocket() { /** * Returns the current Java socket for read-only diagnostics. * - * The socket can be replaced or closed concurrently during a reconnect. Callers must not - * close it or change its configuration. + * For TLS connections, the returned object is an SSLSocket. The socket can be replaced or + * closed concurrently during a reconnect. Callers must not read from, write to, close, or + * reconfigure it because this stream owns all socket I/O. */ fun getJavaSocket(): Socket? = socket.takeIf { it.isConnected && !it.isClosed } diff --git a/common/src/test/java/com/pedro/common/socket/java/TcpStreamSocketJavaBaseTest.kt b/common/src/test/java/com/pedro/common/socket/java/TcpStreamSocketJavaBaseTest.kt new file mode 100644 index 0000000000..ce90c33f4d --- /dev/null +++ b/common/src/test/java/com/pedro/common/socket/java/TcpStreamSocketJavaBaseTest.kt @@ -0,0 +1,59 @@ +/* + * 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.common.socket.java + +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertNull +import org.junit.Assert.assertSame +import org.junit.Test +import java.net.InetAddress +import java.net.ServerSocket +import java.net.Socket + +class TcpStreamSocketJavaBaseTest { + + @Test + fun `GIVEN a Java socket WHEN connecting and closing THEN expose only the active socket`() = runTest { + ServerSocket(0, 1, InetAddress.getLoopbackAddress()).use { serverSocket -> + val streamSocket = TestTcpStreamSocket(serverSocket) + + assertNull(streamSocket.getJavaSocket()) + + streamSocket.connect() + serverSocket.accept().use { + assertSame(streamSocket.connectedSocket, streamSocket.getJavaSocket()) + + streamSocket.close() + assertNull(streamSocket.getJavaSocket()) + } + } + } + + private class TestTcpStreamSocket( + private val serverSocket: ServerSocket, + ): TcpStreamSocketJavaBase() { + + lateinit var connectedSocket: Socket + private set + + override fun onConnectSocket(timeout: Long): Socket { + return Socket(serverSocket.inetAddress, serverSocket.localPort).also { + connectedSocket = it + } + } + } +} 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 2f956fb6ab..7032c1a2e9 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 @@ -184,7 +184,9 @@ class GenericStreamClient( /** * Returns the active direct RTMP Java socket for read-only diagnostics. * - * Returns null when another protocol or socket implementation is active. + * For RTMPS connections, the returned object is an SSLSocket. Callers must not read from, write + * to, close, or reconfigure it because the stream owns all socket I/O. The socket can become + * stale during reconnect. Returns null when another protocol or socket implementation is active. */ fun getJavaSocket(): Socket? = if (connectedStreamClient === rtmpClient) { rtmpClient.getJavaSocket() 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 f60c7775d7..62dd455c20 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 @@ -89,7 +89,13 @@ class RtmpStreamClient( */ fun getRtt() = rtmpClient.rtt - /** Returns the active direct RTMP Java socket for read-only diagnostics. */ + /** + * Returns the active direct RTMP Java socket for read-only diagnostics. + * + * For RTMPS connections, the returned object is an SSLSocket. Callers must not read from, write + * to, close, or reconfigure it because the stream owns all socket I/O. The socket can become + * stale during reconnect. + */ fun getJavaSocket(): Socket? = rtmpClient.getJavaSocket() /** 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 5bb75c4df8..f69ad27563 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt @@ -70,6 +70,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) @@ -154,7 +155,9 @@ class RtmpClient(private val connectChecker: ConnectChecker) { /** * Returns the active direct RTMP Java socket for read-only diagnostics. * - * The socket can become stale during reconnect. Returns null for tunneled or non-Java sockets. + * For RTMPS connections, the returned object is an SSLSocket. The socket can become stale + * during reconnect. Callers must not read from, write to, close, or reconfigure it because the + * RTMP client owns all socket I/O. Returns null for tunneled or non-Java sockets. */ fun getJavaSocket(): Socket? = (socket as? TcpSocket)?.getJavaSocket() 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 4e117805c8..7590259373 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 @@ -54,7 +54,12 @@ class TcpSocket( override fun isReachable(): Boolean = socket.isReachable() - /** Returns the active Java socket for read-only diagnostics, or null for non-Java sockets. */ + /** + * Returns the active Java socket for read-only diagnostics, or null for non-Java sockets. + * + * For TLS connections, the returned object is an SSLSocket. Callers must not read from, write + * to, close, or reconfigure it because this RTMP socket owns all socket I/O. + */ fun getJavaSocket(): Socket? = (socket as? TcpStreamSocketJavaBase)?.getJavaSocket() override suspend fun write(b: Int) {