Expose active Java socket for RTMP diagnostics

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

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

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

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.

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
    */

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