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..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 @@ -8,12 +8,22 @@ 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. + * + * 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 } + override suspend fun connect() { socket = onConnectSocket(timeout) output = socket.getOutputStream().buffered() 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 1df58c19b3..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 @@ -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,19 @@ class GenericStreamClient( override fun getBytesSend(): Long = connectedStreamClient?.getBytesSend() ?: 0 + /** + * 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. 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..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 @@ -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,15 @@ class RtmpStreamClient( */ fun getRtt() = rtmpClient.rtt + /** + * 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() + /** * 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..f69ad27563 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 @@ -69,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) @@ -150,6 +152,15 @@ class RtmpClient(private val connectChecker: ConnectChecker) { shouldSendPings = enabled } + /** + * Returns the active direct RTMP Java socket for read-only diagnostics. + * + * 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() + /** * 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..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 @@ -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,14 @@ class TcpSocket( override fun isReachable(): Boolean = socket.isReachable() + /** + * 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) { socket.write(b) }