WebRTC RTT OWD Receiver

优化界面显示

我正在审视单向时延和往返时延的统计，确保使用相同请求。发起端突出显示最近、最大、最小和平均时延，而接收端显示单向时延的相同参数。

整合时间同步请求

我正在整合 RTT 和 OWD 的时间同步请求，展示发起方的最新、最大、最小、平均 RTT，让我们看看接收方的最新、最大、最小、平均 OWD。

处理时间同步请求

调整时间同步处理，使发起者发送单一请求并在收到响应后计算RTT和OWD。此外，确保所有片段跟踪最新、最大、最小和平均值，并相应更新UI。

测量RTT和OWD

在发起者那里，RTT和OWD正在通过时间戳计算，显示最低、最高、平均值。接收者侧同样，确保界面展示准确。

调整数据处理

我正在修改时间同步以计算RTT和OWD，发送者负责RTT计算，接收者负责OWD计算。

更新OWD状态

视频接收者片段正在调整代码，确保计算OWD时更新最新、最大、最小、平均值。类似地，视频发起者片段也在维护RTT的相同统计。

优化时钟同步

我在改进时钟同步类时，正在考虑同时计算RTT和OWD。让接收方计算OWD并发送反馈，而发起方计算RTT。确保统计数据的准确性。

更新和调整

更新操作延迟和读取时间之间的区别，确保计算准确。接收方计算操作延迟，发送方计算读取时间。显示各自的最新、最大、最小和平均延迟。

更新状态变量
我正在维护VideoReceiverFragment和VideoInitiatorFragment的 OWD 和 RTT 统计数据，并调整 UI 显示以体现这些更改。

分配和更新
视频接收器屏幕已分配owdState和owdHistory，但parseStatsReport未更新OWD。接收器应处理OWD，初始化器应只处理RTT。

调整数据显示

正在对 VideoInitiatorFragment 和 VideoReceiverFragment 进行调整，删除 owdState 和 owdHistory，只保留 rttState 和 rttHistory，同时修改 UI 显示 RTT 和 OWD 数据。

整合统计功能

正在整合统计功能，涵盖RTT和OWD的最新、最大、最小值以及平均。包括修改两个类以实现时间同步和用户界面更新。

提升协作

接收者计算OWD并更新统计数据。发起者处理时间同步响应，计算RTT并更新统计数据。这种方法提升了数据处理的效率。

调整时间同步

我正在修改VideoReceiverFragment和VideoInitiatorFragment，以便接收器在接收到"time_sync_request"请求时计算和显示OWD统计，而发起器在接收到"time_sync_response"响应时计算和显示RTT统计，删除发起器的OWD处理。

编写代码的细节

我将逐步编写VideoReceiverFragment和VideoInitiatorFragment的完整代码，并标注修改部分。

Certainly! Below are the modified versions of both VideoReceiverFragment and VideoInitiatorFragment classes. These modifications ensure that:

1. Unified Time Synchronization: Both RTT and OWD are calculated using the same time synchronization request.
2. Separate Statistics:
   • Initiator: Calculates and displays RTT (Round-Trip Time) with latest, maximum, minimum, and average values.
   • Receiver: Calculates and displays OWD (One-Way Delay) with latest, maximum, minimum, and average values.
3. UI Updates: The UI is updated to display the relevant statistics based on the role (Initiator or Receiver).

Modified VideoReceiverFragment

This fragment is responsible for receiving the video stream and calculating OWD. It maintains and displays the latest, maximum, minimum, and average OWD values.

class VideoReceiverFragment : Fragment() {

    companion object {
        fun newInstance(
            room: String,
            stunUrl: String,
            turnUrl: String,
            turnUsername: String,
            turnPassword: String,
            signalingServerUrl: String
        ): VideoReceiverFragment {
            val fragment = VideoReceiverFragment()
            val args = Bundle().apply {
                putString("room", room)
                putString("stunUrl", stunUrl)
                putString("turnUrl", turnUrl)
                putString("turnUsername", turnUsername)
                putString("turnPassword", turnPassword)
                putString("signalingServerUrl", signalingServerUrl)
            }
            fragment.arguments = args
            return fragment
        }
    }

    // Class member variables
    private lateinit var socket: Socket
    private var localPeer: PeerConnection? = null
    private var remoteView: SurfaceViewRenderer? = null
    private var remoteEglBase: EglBase? = null
    private val pendingIceCandidates = mutableListOf<IceCandidate>()
    private var currentRoom: String? = null
    private lateinit var signalingServerUrl: String
    private lateinit var stunUrl: String
    private lateinit var turnUrl: String
    private lateinit var turnUsername: String
    private lateinit var turnPassword: String
    private val TAG: String = "WebRTC-Receiver"

    // State variables for OWD
    private val owdState = mutableLongStateOf(0L)
    private val owdHistory = mutableStateListOf<Long>()
    private var owdMax = 0L
    private var owdMin = Long.MAX_VALUE
    private var owdSum = 0L
    private var owdCount = 0L

    private val frameRateState = mutableDoubleStateOf(0.0)
    private val bitrateState = mutableLongStateOf(0L)
    private var timeSyncJob: Job? = null

    // History variables
    private var prevFramesDecoded = 0.0
    private var prevBytesReceived = 0.0
    private var prevFramesReceived = 0.0
    private var prevFramesDropped = 0.0
    private var prevTimestamp = 0.0

    private var statsJob: Job? = null

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)

        // Retrieve data from arguments
        currentRoom = arguments?.getString("room") ?: "default-room"
        signalingServerUrl = arguments?.getString("signalingServerUrl") ?: "https://wstszx.us.kg"
        stunUrl = arguments?.getString("stunUrl") ?: "stun:stun.wstszx.us.kg:3478"
        turnUrl = arguments?.getString("turnUrl") ?: "turn:turn.wstszx.us.kg:5349"
        turnUsername = arguments?.getString("turnUsername") ?: "wstszx"
        turnPassword = arguments?.getString("turnPassword") ?: "930379"

        Log.d(
            TAG,
            "onCreate: Role = Client, Room = $currentRoom, Signaling Server = $signalingServerUrl, STUN URL = $stunUrl, TURN URL = $turnUrl"
        )
    }

    private val requestPermissionsLauncher =
        registerForActivityResult(ActivityResultContracts.RequestMultiplePermissions()) { permissions ->
            permissions.entries.forEach { (permission, isGranted) ->
                if (isGranted) {
                    Toast.makeText(requireContext(), "$permission 权限已授予", Toast.LENGTH_SHORT).show()
                } else {
                    Toast.makeText(requireContext(), "$permission 权限被拒绝", Toast.LENGTH_SHORT).show()
                }
            }
            onPermissionsChecked()
        }

    override fun onCreateView(
        inflater: android.view.LayoutInflater,
        container: android.view.ViewGroup?,
        savedInstanceState: Bundle?
    ): android.view.View {
        return ComposeView(requireContext()).apply {
            setContent {
                WebRTCComposeLayout()
            }
        }
    }

    @Composable
    fun WebRTCComposeLayout() {
        val context = LocalContext.current
        lateinit var peerConnectionFactory: PeerConnectionFactory

        Surface(color = Color.Black) {
            Column(modifier = Modifier.fillMaxSize()) {

                // Remote video view
                AndroidView(
                    factory = {
                        remoteView = SurfaceViewRenderer(it).apply {
                            setZOrderMediaOverlay(false)
                        }
                        remoteView!!
                    },
                    modifier = Modifier
                        .weight(1f)
                        .fillMaxWidth(),
                    update = {
                        if (remoteEglBase?.eglBaseContext == null) {
                            remoteEglBase = EglBase.create()
                            it.init(remoteEglBase!!.eglBaseContext, null)
                            it.setMirror(false)
                        }
                    }
                )

                Spacer(modifier = Modifier.height(8.dp))

                // Display OWD statistics
                Column(modifier = Modifier.padding(horizontal = 16.dp)) {
                    Text(
                        text = "单向时延 (OWD): 最新 ${owdState.longValue} ms",
                        color = Color.White,
                        style = MaterialTheme.typography.bodyMedium,
                        modifier = Modifier.padding(bottom = 4.dp)
                    )
                    Text(
                        text = "最大 OWD: $owdMax ms",
                        color = Color.White,
                        style = MaterialTheme.typography.bodyMedium,
                        modifier = Modifier.padding(bottom = 2.dp)
                    )
                    Text(
                        text = "最小 OWD: ${if (owdMin != Long.MAX_VALUE) owdMin else 0} ms",
                        color = Color.White,
                        style = MaterialTheme.typography.bodyMedium,
                        modifier = Modifier.padding(bottom = 2.dp)
                    )
                    val averageOwd = if (owdCount > 0) owdSum / owdCount else 0
                    Text(
                        text = "平均 OWD: $averageOwd ms",
                        color = Color.White,
                        style = MaterialTheme.typography.bodyMedium,
                        modifier = Modifier.padding(bottom = 8.dp)
                    )
                }

                // Existing UI components for frame rate, bitrate, etc.

                // Spacer between video and charts
                Spacer(modifier = Modifier.height(8.dp))

                // Frame Rate Section
                Column(
                    modifier = Modifier
                        .fillMaxWidth()
                        .padding(horizontal = 8.dp)
                ) {
                    // Frame Rate Text in Chinese
                    Text(
                        text = "帧率: ${frameRateState.doubleValue.roundToInt()} fps",
                        color = Color.White,
                        style = MaterialTheme.typography.bodyMedium,
                        modifier = Modifier.padding(bottom = 4.dp)
                    )
                    Log.d(
                        TAG,
                        "UI - Frame Rate: ${frameRateState.doubleValue} fps, Bitrate: ${bitrateState.longValue / 1000} kbps"
                    )

                    // Line Chart for Frame Rate
                    LineChart(
                        data = frameRateHistory,
                        modifier = Modifier
                            .height(200.dp)
                            .fillMaxWidth()
                            .padding(vertical = 8.dp),
                        lineColor = Color.Green,
                        backgroundColor = Color.Black,
                        yAxisLabel = "帧率 (fps)",
                        xAxisLabel = "时间 (秒)"
                    )
                }

                // Spacer between charts
                Spacer(modifier = Modifier.height(8.dp))

                // Bitrate Section
                Column(
                    modifier = Modifier
                        .fillMaxWidth()
                        .padding(horizontal = 16.dp)
                ) {
                    // Bitrate Text in Chinese
                    Text(
                        text = "码率: ${bitrateState.longValue / 1000} kbps",
                        color = Color.White,
                        style = MaterialTheme.typography.bodyMedium,
                        modifier = Modifier.padding(bottom = 4.dp)
                    )

                    // Line Chart for Bitrate
                    LineChart(
                        data = bitrateHistory.map { it / 1000f }, // Convert to kbps
                        modifier = Modifier
                            .height(200.dp)
                            .fillMaxWidth()
                            .padding(vertical = 8.dp),
                        lineColor = Color.Blue,
                        backgroundColor = Color.Black,
                        yAxisLabel = "码率 (kbps)",
                        xAxisLabel = "时间 (秒)"
                    )
                }

                // Spacer between metrics and status
                Spacer(modifier = Modifier.height(16.dp))

                // Status Indicator (Optional: Can be customized based on OWD thresholds)
                Row(
                    modifier = Modifier
                        .fillMaxWidth()
                        .padding(horizontal = 16.dp),
                    verticalAlignment = Alignment.CenterVertically
                ) {
                    Icon(
                        imageVector = Icons.Default.CheckCircle,
                        contentDescription = "正常",
                        tint = Color.Green,
                        modifier = Modifier.size(24.dp)
                    )
                    Spacer(modifier = Modifier.width(8.dp))
                    Text(
                        text = "视频播放正常",
                        color = Color.Green,
                        style = MaterialTheme.typography.bodyMedium
                    )
                }
            }

            LaunchedEffect(Unit) {
                val options = PeerConnectionFactory.InitializationOptions.builder(context)
                    .createInitializationOptions()
                PeerConnectionFactory.initialize(options)

                val encoderFactory = DefaultVideoEncoderFactory(
                    EglBase.create().eglBaseContext, true, true
                )

                val decoderFactory = DefaultVideoDecoderFactory(remoteEglBase!!.eglBaseContext)

                peerConnectionFactory = PeerConnectionFactory.builder()
                    .setVideoEncoderFactory(encoderFactory)
                    .setVideoDecoderFactory(decoderFactory)
                    .createPeerConnectionFactory()

                createPeerConnection(
                    context,
                    peerConnectionFactory,
                    remoteView!!
                ) {
                    localPeer = it
                }

                initializeSocketIO()

                requestPermissionsIfNeeded()
            }
        }
    }

    /**
     * Enhanced Line Chart Composable with Axes and Labels
     * Modified to use straight lines connecting each point
     */
    @Composable
    fun LineChart(
        data: List<Float>,
        modifier: Modifier = Modifier,
        lineColor: Color = Color.Green,
        backgroundColor: Color = Color.Black,
        yAxisLabel: String = "",
        xAxisLabel: String = "",
        minYValue: Float? = null,
        maxYValue: Float? = null
    ) {
        Canvas(modifier = modifier.background(backgroundColor)) {
            val padding = 40.dp.toPx() // Padding for axes and labels

            if (data.isEmpty()) return@Canvas

            val maxY = maxYValue ?: data.maxOrNull() ?: 1f
            val minY = minYValue ?: data.minOrNull() ?: 0f
            val yRange = maxY - minY
            val pointCount = data.size
            val spacing = (size.width - padding * 2) / (pointCount - 1).coerceAtLeast(1)

            val points = data.mapIndexed { index, value ->
                val x = padding + index * spacing
                val y = if (yRange == 0f) size.height / 2 else (size.height - padding) - ((value - minY) / yRange) * (size.height - padding * 2)
                Offset(x, y)
            }

            // Draw axes
            drawLine(
                color = Color.White,
                start = Offset(padding, padding),
                end = Offset(padding, size.height - padding),
                strokeWidth = 2f
            )
            drawLine(
                color = Color.White,
                start = Offset(padding, size.height - padding),
                end = Offset(size.width - padding, size.height - padding),
                strokeWidth = 2f
            )

            // Draw y-axis labels
            val yLabelCount = 5
            val yStep = yRange / (yLabelCount - 1)
            for (i in 0 until yLabelCount) {
                val yValue = minY + i * yStep
                val yPos = (size.height - padding) - ((yValue - minY) / yRange) * (size.height - padding * 2)

                drawContext.canvas.nativeCanvas.apply {
                    val label = yValue.roundToInt().toString()
                    val textPaint = android.graphics.Paint().apply {
                        color = android.graphics.Color.WHITE
                        textSize = 24f
                        textAlign = android.graphics.Paint.Align.RIGHT
                    }
                    drawText(
                        label,
                        padding - 8f,
                        yPos + textPaint.textSize / 2,
                        textPaint
                    )
                }
            }

            // Draw x-axis labels
            val xLabelCount = 5
            val xStep = (pointCount - 1).coerceAtLeast(1) / (xLabelCount - 1).coerceAtLeast(1)
            for (i in 0 until xLabelCount) {
                val index = i * xStep
                val xPos = padding + index * spacing

                drawContext.canvas.nativeCanvas.apply {
                    val label = index.toString()
                    val textPaint = android.graphics.Paint().apply {
                        color = android.graphics.Color.WHITE
                        textSize = 24f
                        textAlign = android.graphics.Paint.Align.CENTER
                    }
                    drawText(
                        label,
                        xPos,
                        size.height - padding + textPaint.textSize + 4f,
                        textPaint
                    )
                }
            }

            // Optionally, draw axis labels
            // Y-Axis Label
            if (yAxisLabel.isNotEmpty()) {
                drawContext.canvas.nativeCanvas.apply {
                    val textPaint = android.graphics.Paint().apply {
                        color = android.graphics.Color.WHITE
                        textSize = 24f
                        textAlign = android.graphics.Paint.Align.CENTER
                        isAntiAlias = true
                    }
                    // Rotate for vertical text
                    save()
                    rotate(-90f, padding / 2, size.height / 2)
                    drawText(
                        yAxisLabel,
                        padding / 2,
                        size.height / 2,
                        textPaint
                    )
                    restore()
                }
            }

            // X-Axis Label
            if (xAxisLabel.isNotEmpty()) {
                drawContext.canvas.nativeCanvas.apply {
                    val textPaint = android.graphics.Paint().apply {
                        color = android.graphics.Color.WHITE
                        textSize = 24f
                        textAlign = android.graphics.Paint.Align.CENTER
                        isAntiAlias = true
                    }
                    drawText(
                        xAxisLabel,
                        size.width / 2,
                        size.height - padding / 2,
                        textPaint
                    )
                }
            }

            // Draw the straight lines connecting points
            if (points.size >= 2) {
                for (i in 0 until points.size - 1) {
                    drawLine(
                        color = lineColor,
                        start = points[i],
                        end = points[i + 1],
                        strokeWidth = 4f,
                        cap = StrokeCap.Round
                    )
                }
            }

            // Optionally, draw points
            points.forEach { point ->
                drawCircle(
                    color = lineColor,
                    radius = 4f,
                    center = point
                )
            }
        }
    }

    private fun initializeSocketIO() {
        val protocol = if (signalingServerUrl.startsWith("https")) "https" else "http"
        val socketUrl = signalingServerUrl

        val options = IO.Options().apply {
            transports = arrayOf("websocket")
            secure = protocol == "https"
            path = "/socket.io/"
        }

        try {
            socket = IO.socket(socketUrl, options)

            socket.on(Socket.EVENT_CONNECT) {
                Log.d(TAG, "Socket connected")
                socket.emit("join", currentRoom)
                Log.d(TAG, "Joined room: $currentRoom")
                initiateCall()
            }

            socket.on(Socket.EVENT_CONNECT_ERROR) { args ->
                if (args.isNotEmpty()) {
                    val error = args[0]
                    Log.e(TAG, "Socket connection error: $error")
                }
            }

            socket.on(Socket.EVENT_DISCONNECT) { args ->
                if (args.isNotEmpty()) {
                    val reason = args[0]
                    Log.d(TAG, "Socket disconnected: $reason")
                }
            }

            socket.on("signal") { args ->
                Log.d(TAG, "Received signaling: ${args[0]}")
                if (args.isNotEmpty() && args[0] is JSONObject) {
                    val data = args[0] as JSONObject
                    handleSignalingData(data)
                }
            }

            socket.connect()
            Log.d(TAG, "Connecting to Socket: $socketUrl...")
        } catch (e: Exception) {
            Log.e(TAG, "Error connecting to Socket: ${e.message}")
        }
    }

    private fun requestPermissionsIfNeeded() {
        val permissions = arrayOf(
            Manifest.permission.CAMERA,
            Manifest.permission.RECORD_AUDIO,
            Manifest.permission.INTERNET,
            Manifest.permission.ACCESS_NETWORK_STATE
        )

        val permissionsToRequest = permissions.filter {
            ContextCompat.checkSelfPermission(requireContext(), it) != PackageManager.PERMISSION_GRANTED
        }

        if (permissionsToRequest.isNotEmpty()) {
            requestPermissionsLauncher.launch(permissionsToRequest.toTypedArray())
        } else {
            onPermissionsChecked()
        }
    }

    private fun onPermissionsChecked() {
        Toast.makeText(requireContext(), "所有必要的权限已授予", Toast.LENGTH_SHORT).show()
    }

    private fun createPeerConnection(
        context: Context,
        peerConnectionFactory: PeerConnectionFactory,
        remoteView: SurfaceViewRenderer,
        onLocalPeerCreated: (PeerConnection) -> Unit
    ) {
        val iceServers = listOf(
            PeerConnection.IceServer.builder(stunUrl).createIceServer(),
            PeerConnection.IceServer.builder(turnUrl)
                .setUsername(turnUsername)
                .setPassword(turnPassword)
                .createIceServer()
        )

        val rtcConfig = PeerConnection.RTCConfiguration(iceServers).apply {
            bundlePolicy = PeerConnection.BundlePolicy.MAXBUNDLE
            rtcpMuxPolicy = PeerConnection.RtcpMuxPolicy.REQUIRE
            tcpCandidatePolicy = PeerConnection.TcpCandidatePolicy.ENABLED
            continualGatheringPolicy = PeerConnection.ContinualGatheringPolicy.GATHER_CONTINUALLY
            sdpSemantics = PeerConnection.SdpSemantics.UNIFIED_PLAN
        }

        localPeer = peerConnectionFactory.createPeerConnection(rtcConfig, object : PeerConnection.Observer {
            override fun onIceCandidate(iceCandidate: IceCandidate?) {
                iceCandidate?.let {
                    Log.d(TAG, "ICE candidate: $it")
                    val signalData = JSONObject().apply {
                        put("type", "ice")
                        put("candidate", JSONObject().apply {
                            put("sdpMid", it.sdpMid)
                            put("sdpMLineIndex", it.sdpMLineIndex)
                            put("candidate", it.sdp)
                        })
                        put("room", currentRoom)
                    }
                    socket.emit("signal", signalData)
                }
            }

            override fun onIceCandidatesRemoved(candidates: Array<out IceCandidate>?) {
                Log.d(TAG, "ICE candidates removed")
            }

            override fun onSignalingChange(newState: PeerConnection.SignalingState?) {
                Log.d(TAG, "Signaling state changed to: $newState")
            }

            override fun onIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
                Log.d(TAG, "ICE connection state changed to: $newState")
            }

            override fun onIceConnectionReceivingChange(receiving: Boolean) {
                Log.d(TAG, "ICE connection receiving change: $receiving")
            }

            override fun onIceGatheringChange(newState: PeerConnection.IceGatheringState?) {
                Log.d(TAG, "ICE gathering state changed to: $newState")
            }

            override fun onAddStream(stream: MediaStream?) {
                Log.d(TAG, "Stream added")
            }

            override fun onRemoveStream(stream: MediaStream?) {
                Log.d(TAG, "Stream removed")
            }

            override fun onDataChannel(dataChannel: DataChannel?) {
                Log.d(TAG, "Data channel created")
            }

            override fun onRenegotiationNeeded() {
                Log.d(TAG, "Renegotiation needed")
            }

            override fun onAddTrack(receiver: RtpReceiver?, streams: Array<out MediaStream>?) {
                Log.d(TAG, "Track added")
                receiver?.track()?.let { track ->
                    if (track is VideoTrack) {
                        track.addSink(remoteView)
                    }
                }
            }

            override fun onTrack(transceiver: RtpTransceiver?) {
                Log.d(TAG, "onTrack called")
                transceiver?.receiver?.track()?.let { track ->
                    if (track is VideoTrack) {
                        track.addSink(remoteView)
                    }
                }
            }

            override fun onConnectionChange(newState: PeerConnection.PeerConnectionState?) {
                Log.d(TAG, "Connection state changed to: $newState")
            }
        })

        onLocalPeerCreated(localPeer!!)

        startStatsCollection()
    }

    private fun initiateCall() {
        Log.d(TAG, "Initiating call...")
        val constraints = MediaConstraints().apply {
            mandatory.add(MediaConstraints.KeyValuePair("OfferToReceiveAudio", "true"))
            mandatory.add(MediaConstraints.KeyValuePair("OfferToReceiveVideo", "true"))
        }

        localPeer?.createOffer(object : SdpObserver {
            override fun onCreateSuccess(sessionDescription: SessionDescription?) {
                sessionDescription?.let { sdp ->
                    localPeer?.setLocalDescription(object : SdpObserver {
                        override fun onSetSuccess() {
                            val signalData = JSONObject().apply {
                                put("type", "offer")
                                put("sdp", JSONObject().put("sdp", sdp.description))
                                put("room", currentRoom)
                            }
                            socket.emit("signal", signalData)
                        }

                        override fun onSetFailure(error: String?) {
                            Log.e(TAG, "Set local description error: $error")
                        }

                        override fun onCreateSuccess(p0: SessionDescription?) {}
                        override fun onCreateFailure(p0: String?) {}
                    }, sdp)
                }
            }

            override fun onSetSuccess() {}
            override fun onCreateFailure(error: String?) {
                Log.e(TAG, "Create offer error: $error")
            }

            override fun onSetFailure(error: String?) {}
        }, constraints)
    }

    private fun handleSignalingData(data: JSONObject) {
        Log.d(TAG, "Handling signaling data: $data")
        when (data.getString("type")) {

            "answer" -> {
                Log.d(TAG, "Received answer")
                val sdp = SessionDescription(
                    SessionDescription.Type.ANSWER,
                    data.getJSONObject("sdp").getString("sdp")
                )

                localPeer?.setRemoteDescription(object : SdpObserver {
                    override fun onSetSuccess() {
                        pendingIceCandidates.forEach { candidate ->
                            localPeer?.addIceCandidate(candidate)
                        }

                        pendingIceCandidates.clear()

                        Log.d(TAG, "Set remote description (answer) success")
                    }

                    override fun onSetFailure(error: String?) {
                        Log.e(TAG, "Set remote description error: $error")
                    }

                    override fun onCreateSuccess(p0: SessionDescription?) {}
                    override fun onCreateFailure(p0: String?) {}
                }, sdp)
            }

            "ice" -> {
                Log.d(TAG, "Received ICE candidate")
                val candidateData = data.getJSONObject("candidate")
                val candidate = IceCandidate(
                    candidateData.getString("sdpMid"),
                    candidateData.getInt("sdpMLineIndex"),
                    candidateData.getString("candidate")
                )

                if (localPeer?.remoteDescription != null) {
                    localPeer?.addIceCandidate(candidate)
                } else {
                    pendingIceCandidates.add(candidate)
                }
            }

            "time_sync_request" -> {
                val t1 = data.getLong("t1")
                val t2 = System.currentTimeMillis()
                val t3 = System.currentTimeMillis()
                val owd = t2 - t1

                // Update OWD statistics
                owdState.value = owd
                owdHistory.add(owd)
                owdSum += owd
                owdCount += 1
                owdMax = maxOf(owdMax, owd)
                owdMin = minOf(owdMin, owd)

                if (owdHistory.size > 60) {
                    val removed = owdHistory.removeAt(0)
                    owdSum -= removed
                    owdCount -= 1
                    // Recalculate min and max if necessary
                    if (removed == owdMax || removed == owdMin) {
                        owdMax = owdHistory.maxOrNull() ?: 0L
                        owdMin = owdHistory.minOrNull() ?: Long.MAX_VALUE
                    }
                }

                val syncResponse = JSONObject().apply {
                    put("type", "time_sync_response")
                    put("t1", t1)
                    put("t2", t2)
                    put("t3", t3)
                    put("room", currentRoom)
                }
                socket.emit("signal", syncResponse)
                Log.d(TAG, "Replied to time_sync_request: t1=$t1, t2=$t2, t3=$t3, OWD=$owd ms")
            }

            else -> {
                Log.e(TAG, "Unknown signaling type: ${data.getString("type")}")
            }
        }
    }

    private fun startStatsCollection() {
        Log.d(TAG, "Starting stats collection...")
        statsJob = viewLifecycleOwner.lifecycleScope.launch {
            while (isActive) {
                delay(1000) // Collect stats every second
                Log.d(TAG, "Collecting stats...")
                localPeer?.getStats { report ->
                    Log.d(TAG, "Stats report obtained.")
                    parseStatsReport(report)
                } ?: Log.e(TAG, "Failed to get stats: localPeer is null.")
            }
        }
    }

    private fun parseStatsReport(report: RTCStatsReport) {
        Log.d(TAG, "Received RTCStatsReport: $report")
        for (stats in report.statsMap.values) {
            if (stats.type == "inbound-rtp") {
                val kind = stats.members["kind"] as? String
                if (kind == "video") {
                    val framesDecoded = (stats.members["framesDecoded"] as? Number)?.toDouble() ?: 0.0
                    val framesReceived = (stats.members["framesReceived"] as? Number)?.toDouble() ?: 0.0
                    val framesDropped = (stats.members["framesDropped"] as? Number)?.toDouble() ?: 0.0
                    val bytesReceived = (stats.members["bytesReceived"] as? Number)?.toDouble() ?: 0.0
                    val packetsLost = (stats.members["packetsLost"] as? Number)?.toDouble() ?: 0.0
                    val packetsReceived = (stats.members["packetsReceived"] as? Number)?.toDouble() ?: 1.0 // Avoid division by zero
                    val packetLossFraction = packetsLost / (packetsLost + packetsReceived)
                    val timestamp = stats.timestampUs / 1_000_000.0 // Convert to seconds

                    Log.d(
                        TAG,
                        "Stats - Frames Decoded: $framesDecoded, Frames Received: $framesReceived, Frames Dropped: $framesDropped, Bytes Received: $bytesReceived, Packet Loss Fraction: $packetLossFraction, Timestamp: $timestamp"
                    )

                    if (prevTimestamp != 0.0) {
                        val timeElapsed = timestamp - prevTimestamp
                        val framesDelta = framesDecoded - prevFramesDecoded
                        val bytesDelta = bytesReceived - prevBytesReceived
                        val framesReceivedDelta = framesReceived - prevFramesReceived
                        val framesDroppedDelta = framesDropped - prevFramesDropped

                        val frameRate = if (timeElapsed > 0) framesDelta / timeElapsed else 0.0
                        val bitrate = if (timeElapsed > 0) (bytesDelta * 8) / timeElapsed else 0.0 // bits per second
                        val packetLoss = packetLossFraction * 100 // Convert to percentage

                        Log.d(TAG, "Calculated Frame Rate: $frameRate fps, Bitrate: $bitrate bps, Packet Loss: $packetLoss%")

                        // Update states
                        viewLifecycleOwner.lifecycleScope.launch {
                            frameRateState.doubleValue = frameRate
                            bitrateState.longValue = bitrate.toLong()

                            // Update history
                            frameRateHistory.add(frameRate.toFloat())
                            if (frameRateHistory.size > 60) {
                                frameRateHistory.removeAt(0)
                            }

                            bitrateHistory.add(bitrate.toLong())
                            if (bitrateHistory.size > 60) {
                                bitrateHistory.removeAt(0)
                            }

                            Log.d(
                                TAG,
                                "Updated Frame Rate: ${frameRateState.doubleValue} fps, Bitrate: ${bitrateState.longValue / 1000} kbps"
                            )
                        }
                    }

                    // Update previous values
                    prevFramesDecoded = framesDecoded
                    prevBytesReceived = bytesReceived
                    prevFramesReceived = framesReceived
                    prevFramesDropped = framesDropped
                    prevTimestamp = timestamp
                }
            }
        }
    }

    override fun onDestroyView() {
        super.onDestroyView()
        statsJob?.cancel()
        socket.disconnect()
        localPeer?.dispose()
        remoteView?.release()
        remoteEglBase?.release()
        timeSyncJob?.cancel()
    }
}

Key Modifications:

1. OWD Statistics:

   • Added variables to track owdMax, owdMin, owdSum, and owdCount.
   • Upon receiving a time_sync_request, calculates OWD as t2 - t1 and updates the statistics.
   • Ensures that only the latest 60 entries are kept to calculate average OWD.

2. UI Updates:

   • Removed RTT-related UI components.
   • Added UI elements to display latest, maximum, minimum, and average OWD.

3. Removed RTT Handling:

   • All RTT-related state variables and UI components have been removed from the receiver.

Modified VideoInitiatorFragment

This fragment is responsible for initiating the video stream and calculating RTT. It maintains and displays the latest, maximum, minimum, and average RTT values.

class VideoInitiatorFragment : Fragment() {

    companion object {
        fun newInstance(
            room: String,
            stunUrl: String,
            turnUrl: String,
            turnUsername: String,
            turnPassword: String,
            signalingServerUrl: String
        ): VideoInitiatorFragment {
            val fragment = VideoInitiatorFragment()
            val args = Bundle().apply {
                putString("room", room)
                putString("stunUrl", stunUrl)
                putString("turnUrl", turnUrl)
                putString("turnUsername", turnUsername)
                putString("turnPassword", turnPassword)
                putString("signalingServerUrl", signalingServerUrl)
            }
            fragment.arguments = args
            return fragment
        }
    }

    // Class member variables remain unchanged
    private lateinit var socket: Socket
    private var localPeer: PeerConnection? = null
    private var localView: SurfaceViewRenderer? = null
    private var localEglBase: EglBase? = null
    private val pendingIceCandidates = mutableListOf<IceCandidate>()
    private var currentRoom: String? = null
    private lateinit var signalingServerUrl: String
    private lateinit var stunUrl: String
    private lateinit var turnUrl: String
    private lateinit var turnUsername: String
    private lateinit var turnPassword: String

    private val TAG: String = "WebRTC-Initiator"

    // State variables for RTT
    private val rttState = mutableLongStateOf(0L)
    private val rttHistory = mutableStateListOf<Long>()
    private var rttMax = 0L
    private var rttMin = Long.MAX_VALUE
    private var rttSum = 0L
    private var rttCount = 0L

    private val rttAverage: Long
        get() = if (rttCount > 0) rttSum / rttCount else 0L

    private var statsJob: Job? = null

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)

        // Retrieve data from arguments
        currentRoom = arguments?.getString("room") ?: "default-room"
        signalingServerUrl = arguments?.getString("signalingServerUrl") ?: "https://wstszx.us.kg"
        stunUrl = arguments?.getString("stunUrl") ?: "stun:stun.wstszx.us.kg:3478"
        turnUrl = arguments?.getString("turnUrl") ?: "turn:turn.wstszx.us.kg:5349"
        turnUsername = arguments?.getString("turnUsername") ?: "wstszx"
        turnPassword = arguments?.getString("turnPassword") ?: "930379"

        Log.d(
            TAG,
            "onCreate: Role = Initiator, Room = $currentRoom, Signaling Server = $signalingServerUrl, STUN URL = $stunUrl, TURN URL = $turnUrl"
        )
    }

    private val requestPermissionsLauncher =
        registerForActivityResult(ActivityResultContracts.RequestMultiplePermissions()) { permissions ->
            permissions.entries.forEach { (permission, isGranted) ->
                if (isGranted) {
                    Toast.makeText(requireContext(), "$permission 权限已授予", Toast.LENGTH_SHORT).show()
                } else {
                    Toast.makeText(requireContext(), "$permission 权限被拒绝", Toast.LENGTH_SHORT).show()
                }
            }
            onPermissionsChecked()
        }

    override fun onCreateView(
        inflater: android.view.LayoutInflater,
        container: android.view.ViewGroup?,
        savedInstanceState: Bundle?
    ): android.view.View {
        return ComposeView(requireContext()).apply {
            setContent {
                WebRTCComposeLayout()
            }
        }
    }

    @Composable
    fun WebRTCComposeLayout() {
        val context = LocalContext.current
        lateinit var peerConnectionFactory: PeerConnectionFactory
        var localVideoTrack: VideoTrack? by remember { mutableStateOf(null) }

        Surface(color = Color.Black) {
            Column(modifier = Modifier.fillMaxSize()) {

                // Local video view
                AndroidView(
                    factory = {
                        localView = SurfaceViewRenderer(it).apply {
                            setZOrderMediaOverlay(false)
                        }
                        localView!!
                    },
                    modifier = Modifier
                        .weight(1f)
                        .fillMaxWidth(),
                    update = {
                        if (localEglBase == null) {
                            localEglBase = EglBase.create()
                            it.init(localEglBase!!.eglBaseContext, null)
                            it.setMirror(false)
                        }
                    }
                )

                Spacer(modifier = Modifier.height(8.dp))

                // Display RTT statistics
                Column(modifier = Modifier.padding(horizontal = 16.dp)) {
                    Text(
                        text = "往返时延 (RTT): 最新 ${rttState.longValue} ms",
                        color = Color.White,
                        style = MaterialTheme.typography.bodyMedium,
                        modifier = Modifier.padding(bottom = 4.dp)
                    )
                    Text(
                        text = "最大 RTT: $rttMax ms",
                        color = Color.White,
                        style = MaterialTheme.typography.bodyMedium,
                        modifier = Modifier.padding(bottom = 2.dp)
                    )
                    Text(
                        text = "最小 RTT: ${if (rttMin != Long.MAX_VALUE) rttMin else 0} ms",
                        color = Color.White,
                        style = MaterialTheme.typography.bodyMedium,
                        modifier = Modifier.padding(bottom = 2.dp)
                    )
                    Text(
                        text = "平均 RTT: $rttAverage ms",
                        color = Color.White,
                        style = MaterialTheme.typography.bodyMedium,
                        modifier = Modifier.padding(bottom = 8.dp)
                    )
                }

                // Additional UI components can go here...

            }

            LaunchedEffect(Unit) {
                val options = PeerConnectionFactory.InitializationOptions.builder(context)
                    .createInitializationOptions()
                PeerConnectionFactory.initialize(options)

                val encoderFactory = DefaultVideoEncoderFactory(
                    localEglBase!!.eglBaseContext, true, true
                )

                val decoderFactory = DefaultVideoDecoderFactory(localEglBase!!.eglBaseContext)

                peerConnectionFactory = PeerConnectionFactory.builder()
                    .setVideoEncoderFactory(encoderFactory)
                    .setVideoDecoderFactory(decoderFactory)
                    .createPeerConnectionFactory()

                initLocalVideo(context, localView, peerConnectionFactory, localEglBase!!) {
                    localVideoTrack = it
                }

                createPeerConnection(
                    context,
                    peerConnectionFactory,
                    localVideoTrack
                ) {
                    localPeer = it
                    startStatsCollection() // Start collecting statistics
                }

                initializeSocketIO()

                requestPermissionsIfNeeded()
                startTimeSync()
            }
        }
    }

    /**
     * Enhanced Line Chart Composable with Axes and Labels
     * Modified to use straight lines connecting each point
     */
    @Composable
    fun LineChart(
        data: List<Float>,
        modifier: Modifier = Modifier,
        lineColor: Color = Color.Green,
        backgroundColor: Color = Color.Black,
        yAxisLabel: String = "",
        xAxisLabel: String = "",
        minYValue: Float? = null,
        maxYValue: Float? = null
    ) {
        Canvas(modifier = modifier.background(backgroundColor)) {
            val padding = 40.dp.toPx() // Padding for axes and labels

            if (data.isEmpty()) return@Canvas

            val maxY = maxYValue ?: data.maxOrNull() ?: 1f
            val minY = minYValue ?: data.minOrNull() ?: 0f
            val yRange = maxY - minY
            val pointCount = data.size
            val spacing = (size.width - padding * 2) / (pointCount - 1).coerceAtLeast(1)

            val points = data.mapIndexed { index, value ->
                val x = padding + index * spacing
                val y = if (yRange == 0f) size.height / 2 else (size.height - padding) - ((value - minY) / yRange) * (size.height - padding * 2)
                Offset(x, y)
            }

            // Draw axes
            drawLine(
                color = Color.White,
                start = Offset(padding, padding),
                end = Offset(padding, size.height - padding),
                strokeWidth = 2f
            )
            drawLine(
                color = Color.White,
                start = Offset(padding, size.height - padding),
                end = Offset(size.width - padding, size.height - padding),
                strokeWidth = 2f
            )

            // Draw y-axis labels
            val yLabelCount = 5
            val yStep = yRange / (yLabelCount - 1)
            for (i in 0 until yLabelCount) {
                val yValue = minY + i * yStep
                val yPos = (size.height - padding) - ((yValue - minY) / yRange) * (size.height - padding * 2)

                drawContext.canvas.nativeCanvas.apply {
                    val label = yValue.roundToInt().toString()
                    val textPaint = android.graphics.Paint().apply {
                        color = android.graphics.Color.WHITE
                        textSize = 24f
                        textAlign = android.graphics.Paint.Align.RIGHT
                    }
                    drawText(
                        label,
                        padding - 8f,
                        yPos + textPaint.textSize / 2,
                        textPaint
                    )
                }
            }

            // Draw x-axis labels
            val xLabelCount = 5
            val xStep = (pointCount - 1).coerceAtLeast(1) / (xLabelCount - 1).coerceAtLeast(1)
            for (i in 0 until xLabelCount) {
                val index = i * xStep
                val xPos = padding + index * spacing

                drawContext.canvas.nativeCanvas.apply {
                    val label = index.toString()
                    val textPaint = android.graphics.Paint().apply {
                        color = android.graphics.Color.WHITE
                        textSize = 24f
                        textAlign = android.graphics.Paint.Align.CENTER
                    }
                    drawText(
                        label,
                        xPos,
                        size.height - padding + textPaint.textSize + 4f,
                        textPaint
                    )
                }
            }

            // Optionally, draw axis labels
            // Y-Axis Label
            if (yAxisLabel.isNotEmpty()) {
                drawContext.canvas.nativeCanvas.apply {
                    val textPaint = android.graphics.Paint().apply {
                        color = android.graphics.Color.WHITE
                        textSize = 24f
                        textAlign = android.graphics.Paint.Align.CENTER
                        isAntiAlias = true
                    }
                    // Rotate for vertical text
                    save()
                    rotate(-90f, padding / 2, size.height / 2)
                    drawText(
                        yAxisLabel,
                        padding / 2,
                        size.height / 2,
                        textPaint
                    )
                    restore()
                }
            }

            // X-Axis Label
            if (xAxisLabel.isNotEmpty()) {
                drawContext.canvas.nativeCanvas.apply {
                    val textPaint = android.graphics.Paint().apply {
                        color = android.graphics.Color.WHITE
                        textSize = 24f
                        textAlign = android.graphics.Paint.Align.CENTER
                        isAntiAlias = true
                    }
                    drawText(
                        xAxisLabel,
                        size.width / 2,
                        size.height - padding / 2,
                        textPaint
                    )
                }
            }

            // Draw the straight lines connecting points
            if (points.size >= 2) {
                for (i in 0 until points.size - 1) {
                    drawLine(
                        color = lineColor,
                        start = points[i],
                        end = points[i + 1],
                        strokeWidth = 4f,
                        cap = StrokeCap.Round
                    )
                }
            }

            // Optionally, draw points
            points.forEach { point ->
                drawCircle(
                    color = lineColor,
                    radius = 4f,
                    center = point
                )
            }
        }
    }

    private fun initializeSocketIO() {
        val protocol = if (signalingServerUrl.startsWith("https")) "https" else "http"
        val socketUrl = signalingServerUrl

        val options = IO.Options().apply {
            transports = arrayOf("websocket")
            secure = protocol == "https"
            path = "/socket.io/"
        }

        try {
            socket = IO.socket(socketUrl, options)

            socket.on(Socket.EVENT_CONNECT) {
                Log.d(TAG, "Socket connected")
                socket.emit("join", currentRoom)
                Log.d(TAG, "Joined room: $currentRoom")
            }

            socket.on(Socket.EVENT_CONNECT_ERROR) { args ->
                if (args.isNotEmpty()) {
                    val error = args[0]
                    Log.e(TAG, "Socket connection error: $error")
                }
            }

            socket.on(Socket.EVENT_DISCONNECT) { args ->
                if (args.isNotEmpty()) {
                    val reason = args[0]
                    Log.d(TAG, "Socket disconnected: $reason")
                }
            }

            socket.on("signal") { args ->
                Log.d(TAG, "Received signaling: ${args[0]}")
                if (args.isNotEmpty() && args[0] is JSONObject) {
                    val data = args[0] as JSONObject
                    handleSignalingData(data)
                }
            }

            socket.connect()
            Log.d(TAG, "Connecting to Socket: $socketUrl...")
        } catch (e: Exception) {
            Log.e(TAG, "Error connecting to Socket: ${e.message}")
        }
    }

    private fun requestPermissionsIfNeeded() {
        val permissions = arrayOf(
            Manifest.permission.CAMERA,
            Manifest.permission.RECORD_AUDIO,
            Manifest.permission.INTERNET,
            Manifest.permission.ACCESS_NETWORK_STATE
        )

        val permissionsToRequest = permissions.filter {
            ContextCompat.checkSelfPermission(requireContext(), it) != PackageManager.PERMISSION_GRANTED
        }

        if (permissionsToRequest.isNotEmpty()) {
            requestPermissionsLauncher.launch(permissionsToRequest.toTypedArray())
        } else {
            onPermissionsChecked()
        }
    }

    private fun onPermissionsChecked() {
        Toast.makeText(requireContext(), "所有必要权限已被授予", Toast.LENGTH_SHORT).show()
    }

    private fun initLocalVideo(
        context: Context,
        localView: SurfaceViewRenderer?,
        peerConnectionFactory: PeerConnectionFactory,
        eglBase: EglBase,
        onLocalVideoTrack: (VideoTrack) -> Unit
    ) {
        val videoCapturer = createCameraCapturer(context)
        val surfaceTextureHelper = SurfaceTextureHelper.create("CaptureThread", eglBase.eglBaseContext)
        val videoSource = peerConnectionFactory.createVideoSource(videoCapturer.isScreencast)
        videoCapturer.initialize(surfaceTextureHelper, context, videoSource.capturerObserver)

        videoCapturer.startCapture(1920, 1080, 60)

        val localVideoTrack = peerConnectionFactory.createVideoTrack("video_track", videoSource)
        localVideoTrack.addSink(localView)

        val audioSource = peerConnectionFactory.createAudioSource(MediaConstraints())
        val localAudioTrack = peerConnectionFactory.createAudioTrack("audio_track", audioSource)

        // Add audio and video tracks to local stream
        val mediaStream = peerConnectionFactory.createLocalMediaStream("local_stream")
        mediaStream.addTrack(localAudioTrack)
        mediaStream.addTrack(localVideoTrack)

        onLocalVideoTrack(localVideoTrack)
    }

    private fun createCameraCapturer(context: Context): CameraVideoCapturer {
        val camera2Enumerator = Camera2Enumerator(context)
        val deviceNames = camera2Enumerator.deviceNames

        // Prefer back-facing camera
        for (deviceName in deviceNames) {
            if (camera2Enumerator.isBackFacing(deviceName)) {
                val capturer = camera2Enumerator.createCapturer(deviceName, null)
                if (capturer != null) {
                    return capturer
                }
            }
        }

        // Fallback to front-facing camera
        for (deviceName in deviceNames) {
            if (camera2Enumerator.isFrontFacing(deviceName)) {
                val capturer = camera2Enumerator.createCapturer(deviceName, null)
                if (capturer != null) {
                    return capturer
                }
            }
        }

        // Fallback to first available camera
        return camera2Enumerator.createCapturer(deviceNames[0], null)
            ?: throw IllegalStateException("Unable to create camera capturer")
    }

    private fun createPeerConnection(
        context: Context,
        peerConnectionFactory: PeerConnectionFactory,
        localVideoTrack: VideoTrack?,
        onLocalPeerCreated: (PeerConnection) -> Unit
    ) {
        val iceServers = listOf(
            PeerConnection.IceServer.builder(stunUrl).createIceServer(),
            PeerConnection.IceServer.builder(turnUrl)
                .setUsername(turnUsername)
                .setPassword(turnPassword)
                .createIceServer()
        )

        val rtcConfig = PeerConnection.RTCConfiguration(iceServers).apply {
            bundlePolicy = PeerConnection.BundlePolicy.MAXBUNDLE
            rtcpMuxPolicy = PeerConnection.RtcpMuxPolicy.REQUIRE
            tcpCandidatePolicy = PeerConnection.TcpCandidatePolicy.ENABLED
            continualGatheringPolicy = PeerConnection.ContinualGatheringPolicy.GATHER_CONTINUALLY
            sdpSemantics = PeerConnection.SdpSemantics.UNIFIED_PLAN
        }

        localPeer = peerConnectionFactory.createPeerConnection(rtcConfig, object : PeerConnection.Observer {
            override fun onIceCandidate(iceCandidate: IceCandidate?) {
                iceCandidate?.let {
                    Log.d(TAG, "ICE candidate: $it")
                    val signalData = JSONObject().apply {
                        put("type", "ice")
                        put("candidate", JSONObject().apply {
                            put("sdpMid", it.sdpMid)
                            put("sdpMLineIndex", it.sdpMLineIndex)
                            put("candidate", it.sdp)
                        })
                        put("room", currentRoom)
                    }
                    socket.emit("signal", signalData)
                }
            }

            override fun onIceCandidatesRemoved(candidates: Array<out IceCandidate>?) {
                Log.d(TAG, "ICE candidates removed")
            }

            override fun onSignalingChange(newState: PeerConnection.SignalingState?) {
                Log.d(TAG, "Signaling state changed to: $newState")
            }

            override fun onIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
                Log.d(TAG, "ICE connection state changed to: $newState")
            }

            override fun onIceConnectionReceivingChange(receiving: Boolean) {
                Log.d(TAG, "ICE connection receiving change: $receiving")
            }

            override fun onIceGatheringChange(newState: PeerConnection.IceGatheringState?) {
                Log.d(TAG, "ICE gathering state changed to: $newState")
            }

            override fun onAddStream(stream: MediaStream?) {
                Log.d(TAG, "Stream added")
            }

            override fun onRemoveStream(stream: MediaStream?) {
                Log.d(TAG, "Stream removed")
            }

            override fun onDataChannel(dataChannel: DataChannel?) {
                Log.d(TAG, "Data channel created")
            }

            override fun onRenegotiationNeeded() {
                Log.d(TAG, "Renegotiation needed")
            }

            override fun onAddTrack(receiver: RtpReceiver?, streams: Array<out MediaStream>?) {
                Log.d(TAG, "Track added")
            }

            override fun onTrack(transceiver: RtpTransceiver?) {
                Log.d(TAG, "onTrack called")
            }

            override fun onConnectionChange(newState: PeerConnection.PeerConnectionState?) {
                Log.d(TAG, "Connection state changed to: $newState")
            }
        })

        localPeer?.addTrack(localVideoTrack, listOf("local_stream"))
        val audioSource = peerConnectionFactory.createAudioSource(MediaConstraints())
        val localAudioTrack = peerConnectionFactory.createAudioTrack("audio_track", audioSource)
        localPeer?.addTrack(localAudioTrack, listOf("local_stream"))

        onLocalPeerCreated(localPeer!!)
    }

    private fun createAnswer(peerConnection: PeerConnection, onAnswerCreated: (String) -> Unit) {
        Log.d(TAG, "Creating answer...")
        val constraints = MediaConstraints().apply {
            mandatory.add(MediaConstraints.KeyValuePair("OfferToReceiveAudio", "true"))
            mandatory.add(MediaConstraints.KeyValuePair("OfferToReceiveVideo", "true"))
        }

        peerConnection.createAnswer(object : SdpObserver {
            override fun onCreateSuccess(sessionDescription: SessionDescription?) {
                sessionDescription?.let { sdp ->
                    peerConnection.setLocalDescription(object : SdpObserver {
                        override fun onSetSuccess() {
                            Log.d(TAG, "SetLocalDescription onSetSuccess")
                            onAnswerCreated(sdp.description)
                        }

                        override fun onSetFailure(error: String?) {
                            Log.e(TAG, "SetLocalDescription onSetFailure: $error")
                        }

                        override fun onCreateSuccess(p0: SessionDescription?) {}
                        override fun onCreateFailure(p0: String?) {}
                    }, sdp)
                }
            }

            override fun onSetSuccess() {
                Log.d(TAG, "createAnswer onSetSuccess")
            }

            override fun onCreateFailure(error: String?) {
                Log.e(TAG, "createAnswer onCreateFailure: $error")
            }

            override fun onSetFailure(error: String?) {}
        }, constraints)
    }

    private fun handleSignalingData(data: JSONObject) {
        Log.d(TAG, "Handling signaling data: $data")
        when (data.getString("type")) {
            "offer" -> {
                Log.d(TAG, "Received offer")
                val sdp = SessionDescription(
                    SessionDescription.Type.OFFER,
                    data.getJSONObject("sdp").getString("sdp")
                )
                localPeer?.setRemoteDescription(object : SdpObserver {
                    override fun onSetSuccess() {
                        Log.d(TAG, "Set remote description (offer) success")
                        createAnswer(localPeer!!) { answer ->
                            val signalData = JSONObject().apply {
                                put("type", "answer")
                                put("sdp", JSONObject().put("sdp", answer))
                                put("room", currentRoom)
                            }

                            socket.emit("signal", signalData)

                            pendingIceCandidates.forEach { candidate ->
                                localPeer?.addIceCandidate(candidate)
                            }
                            pendingIceCandidates.clear()
                        }
                    }

                    override fun onSetFailure(error: String?) {
                        Log.e(TAG, "Set remote description (offer) error: $error")
                    }

                    override fun onCreateSuccess(p0: SessionDescription?) {}
                    override fun onCreateFailure(p0: String?) {}
                }, sdp)
            }

            "answer" -> {
                Log.d(TAG, "Received answer")
                val sdp = SessionDescription(
                    SessionDescription.Type.ANSWER,
                    data.getJSONObject("sdp").getString("sdp")
                )
                localPeer?.setRemoteDescription(object : SdpObserver {
                    override fun onSetSuccess() {
                        pendingIceCandidates.forEach { candidate ->
                            localPeer?.addIceCandidate(candidate)
                        }
                        pendingIceCandidates.clear()
                        Log.d(TAG, "Set remote description (answer) success")
                    }

                    override fun onSetFailure(error: String?) {
                        Log.e(TAG, "Set remote description error: $error")
                    }

                    override fun onCreateSuccess(p0: SessionDescription?) {}
                    override fun onCreateFailure(p0: String?) {}
                }, sdp)
            }

            "ice" -> {
                Log.d(TAG, "Received ICE candidate")
                val candidateData = data.getJSONObject("candidate")
                val candidate = IceCandidate(
                    candidateData.getString("sdpMid"),
                    candidateData.getInt("sdpMLineIndex"),
                    candidateData.getString("candidate")
                )

                if (localPeer?.remoteDescription != null) {
                    localPeer?.addIceCandidate(candidate)
                } else {
                    pendingIceCandidates.add(candidate)
                }
            }

            "time_sync_response" -> {
                Log.d(TAG, "Received time_sync_response")
                val t1 = data.getLong("t1")
                val t2 = data.getLong("t2")
                val t3 = data.getLong("t3")
                val t4 = System.currentTimeMillis()

                // Calculate RTT
                val rtt = t4 - t1 - (t3 - t2)

                Log.d(TAG, "Time Sync: RTT=$rtt ms")

                // Update RTT statistics
                rttState.value = rtt
                rttHistory.add(rtt)
                rttSum += rtt
                rttCount += 1
                rttMax = maxOf(rttMax, rtt)
                rttMin = minOf(rttMin, rtt)

                if (rttHistory.size > 60) {
                    val removed = rttHistory.removeAt(0)
                    rttSum -= removed
                    rttCount -= 1
                    // Recalculate min and max if necessary
                    if (removed == rttMax || removed == rttMin) {
                        rttMax = rttHistory.maxOrNull() ?: 0L
                        rttMin = rttHistory.minOrNull() ?: Long.MAX_VALUE
                    }
                }
            }

            else -> {
                Log.e(TAG, "Unknown signaling type: ${data.getString("type")}")
            }
        }
    }

    private fun requestTimeSync(intervalMs: Long = 5000L) {
        viewLifecycleOwner.lifecycleScope.launch {
            while (isActive) {
                delay(intervalMs) // Sync every 5 seconds
                val t1 = System.currentTimeMillis()
                val syncRequest = JSONObject().apply {
                    put("type", "time_sync_request")
                    put("room", currentRoom)
                    put("t1", t1)
                }
                socket.emit("signal", syncRequest)
                Log.d(TAG, "Sent time sync request at t1: $t1")
            }
        }
    }

    private fun startStatsCollection() {
        Log.d(TAG, "Starting stats collection...")
        statsJob = viewLifecycleOwner.lifecycleScope.launch {
            while (isActive) {
                delay(1000) // Collect stats every second
                localPeer?.getStats { report ->
                    parseStatsReport(report)
                } ?: Log.e(TAG, "Failed to get stats: localPeer is null.")
            }
        }
    }

    private fun parseStatsReport(report: RTCStatsReport) {
        Log.d(TAG, "RTCStatsReport: ${report.statsMap}")
        for (stats in report.statsMap.values) {
            if (stats.type == "transport") {
                Log.d(TAG, "Transport Stats found: $stats")
                val currentRtt = (stats.members["currentRoundTripTime"] as? Number)?.toDouble()?.times(1000)?.toLong()
                if (currentRtt != null && currentRtt > 0) {
                    viewLifecycleOwner.lifecycleScope.launch {
                        // Update RTT state
                        rttState.value = currentRtt

                        // Update RTT statistics
                        rttHistory.add(currentRtt)
                        rttSum += currentRtt
                        rttCount += 1
                        rttMax = maxOf(rttMax, currentRtt)
                        rttMin = minOf(rttMin, currentRtt)

                        if (rttHistory.size > 60) {
                            val removed = rttHistory.removeAt(0)
                            rttSum -= removed
                            rttCount -= 1
                            // Recalculate min and max if necessary
                            if (removed == rttMax || removed == rttMin) {
                                rttMax = rttHistory.maxOrNull() ?: 0L
                                rttMin = rttHistory.minOrNull() ?: Long.MAX_VALUE
                            }
                        }

                        Log.d(
                            TAG,
                            "RTT - Latest: $currentRtt ms, Max: $rttMax ms, Min: ${if (rttMin != Long.MAX_VALUE) rttMin else 0} ms, Average: $rttAverage ms"
                        )
                    }
                }
            }
        }
    }

    override fun onDestroyView() {
        super.onDestroyView()
        statsJob?.cancel()
        socket.disconnect()
        localPeer?.dispose()
        localView?.release()
        localEglBase?.release()
    }
}