<NME> Endpoint.kt
<BEF> /*
 * Copyright @ 2018 - present 8x8, Inc.
 *
 * 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 org.jitsi.videobridge

import org.jitsi.config.JitsiConfig
import org.jitsi.metaconfig.config
import org.jitsi.nlj.Features
import org.jitsi.nlj.MediaSourceDesc
import org.jitsi.nlj.PacketHandler
import org.jitsi.nlj.PacketInfo
import org.jitsi.nlj.Transceiver
import org.jitsi.nlj.TransceiverEventHandler
import org.jitsi.nlj.format.PayloadType
import org.jitsi.nlj.rtp.AudioRtpPacket
import org.jitsi.nlj.rtp.ParsedVideoPacket
import org.jitsi.nlj.rtp.RtpExtension
import org.jitsi.nlj.rtp.SsrcAssociationType
import org.jitsi.nlj.rtp.VideoRtpPacket
import org.jitsi.nlj.srtp.TlsRole
import org.jitsi.nlj.stats.EndpointConnectionStats
import org.jitsi.nlj.stats.NodeStatsBlock
import org.jitsi.nlj.transform.node.ConsumerNode
import org.jitsi.nlj.util.Bandwidth
import org.jitsi.nlj.util.LocalSsrcAssociation
import org.jitsi.nlj.util.NEVER
import org.jitsi.nlj.util.PacketInfoQueue
import org.jitsi.nlj.util.RemoteSsrcAssociation
import org.jitsi.nlj.util.sumOf
import org.jitsi.rtp.Packet
import org.jitsi.rtp.UnparsedPacket
import org.jitsi.rtp.rtcp.RtcpSrPacket
import org.jitsi.rtp.rtcp.rtcpfb.RtcpFbPacket
import org.jitsi.rtp.rtcp.rtcpfb.payload_specific_fb.RtcpFbFirPacket
import org.jitsi.rtp.rtcp.rtcpfb.payload_specific_fb.RtcpFbPliPacket
import org.jitsi.rtp.rtp.RtpPacket
import org.jitsi.utils.MediaType
import org.jitsi.utils.concurrent.RecurringRunnableExecutor
import org.jitsi.utils.logging2.Logger
import org.jitsi.utils.logging2.cdebug
import org.jitsi.utils.mins
import org.jitsi.utils.queue.CountingErrorHandler
import org.jitsi.videobridge.cc.BandwidthProbing
import org.jitsi.videobridge.cc.allocation.BandwidthAllocation
import org.jitsi.videobridge.cc.allocation.BitrateController
import org.jitsi.videobridge.cc.allocation.EffectiveConstraintsMap
import org.jitsi.videobridge.cc.allocation.VideoConstraints
import org.jitsi.videobridge.datachannel.DataChannelStack
import org.jitsi.videobridge.datachannel.protocol.DataChannelPacket
import org.jitsi.videobridge.datachannel.protocol.DataChannelProtocolConstants
import org.jitsi.videobridge.message.BridgeChannelMessage
import org.jitsi.videobridge.message.ForwardedEndpointsMessage
import org.jitsi.videobridge.message.ForwardedSourcesMessage
import org.jitsi.videobridge.message.ReceiverVideoConstraintsMessage
import org.jitsi.videobridge.message.SenderSourceConstraintsMessage
import org.jitsi.videobridge.message.SenderVideoConstraintsMessage
import org.jitsi.videobridge.relay.AudioSourceDesc
import org.jitsi.videobridge.rest.root.debug.EndpointDebugFeatures
import org.jitsi.videobridge.sctp.SctpConfig
import org.jitsi.videobridge.sctp.SctpManager
import org.jitsi.videobridge.stats.PacketTransitStats
import org.jitsi.videobridge.transport.dtls.DtlsTransport
import org.jitsi.videobridge.transport.ice.IceTransport
import org.jitsi.videobridge.util.ByteBufferPool
import org.jitsi.videobridge.util.TaskPools
import org.jitsi.videobridge.util.looksLikeDtls
import org.jitsi.videobridge.websocket.colibriWebSocketServiceSupplier
import org.jitsi.xmpp.extensions.colibri.WebSocketPacketExtension
import org.jitsi.xmpp.extensions.jingle.DtlsFingerprintPacketExtension
import org.jitsi.xmpp.extensions.jingle.IceUdpTransportPacketExtension
import org.jitsi_modified.sctp4j.SctpDataCallback
import org.jitsi_modified.sctp4j.SctpServerSocket
import org.jitsi_modified.sctp4j.SctpSocket
import org.json.simple.JSONObject
import java.nio.ByteBuffer
import java.security.SecureRandom
import java.time.Clock
import java.time.Duration
import java.time.Instant
import java.util.Optional
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicLong
import java.util.function.Supplier

/**
 * Models a local endpoint (participant) in a [Conference]
 */
class Endpoint @JvmOverloads constructor(
    id: String,
    conference: Conference,
    parentLogger: Logger,
    /**
     * True if the ICE agent for this Endpoint will be initialized to serve
     * as a controlling ICE agent, false otherwise
     */
    iceControlling: Boolean,
    private val isUsingSourceNames: Boolean,
    private val doSsrcRewriting: Boolean,
    /**
     * Whether this endpoint is in "visitor" mode, i.e. should be invisible to other endpoints.
     */
    val visitor: Boolean,
    private val clock: Clock = Clock.systemUTC()
) : AbstractEndpoint(conference, id, parentLogger),
    PotentialPacketHandler,
    EncodingsManager.EncodingsUpdateListener,
    SsrcRewriter {
    /**
     * The time at which this endpoint was created
     */
    val creationTime = clock.instant()

    private val sctpHandler = SctpHandler()
    private val dataChannelHandler = DataChannelHandler()

    /* TODO: do we ever want to support useUniquePort for an Endpoint? */
    private val iceTransport = IceTransport(id, iceControlling, false, logger)
    private val dtlsTransport = DtlsTransport(logger)

    private val diagnosticContext = conference.newDiagnosticContext().apply {
        put("endpoint_id", id)
    }

    private val timelineLogger = logger.createChildLogger("timeline.${this.javaClass.name}")

    /**
     * The [SctpManager] instance we'll use to manage the SCTP connection
     */
    private var sctpManager: SctpManager? = null

    private var dataChannelStack: DataChannelStack? = null

    /**
     * The [SctpSocket] for this endpoint, if an SCTP connection was
     * negotiated.
     */
    private var sctpSocket: Optional<SctpServerSocket> = Optional.empty()

    /**
     * Whether this endpoint should accept audio packets. We set this according
     * to whether the endpoint has an audio Colibri channel whose direction
     * allows sending.
     */
    var acceptAudio = false

    /**
     * Whether this endpoint should accept video packets. We set this according
     * to whether the endpoint has a video Colibri channel whose direction
     * allows sending.
     */
    var acceptVideo = false

    /**
     * The queue we put outgoing SRTP packets onto so they can be sent
     * out via the [IceTransport] on an IO thread.
     */
    private val outgoingSrtpPacketQueue = PacketInfoQueue(
        "${javaClass.simpleName}-outgoing-packet-queue",
        TaskPools.IO_POOL,
        this::doSendSrtp,
        TransportConfig.queueSize
    ).apply {
        setErrorHandler(queueErrorCounter)
    }

        setErrorHandler(queueErrorCounter)
    }

    private val bitrateController = BitrateController(
        object : BitrateController.EventHandler {
            override fun allocationChanged(allocation: BandwidthAllocation) {
            override fun forwardedEndpointsChanged(forwardedEndpoints: Set<String>) =
                sendForwardedEndpointsMessage(forwardedEndpoints)

            override fun forwardedSourcesChanged(forwardedSources: Set<String>) {
                sendForwardedSourcesMessage(forwardedSources)
            }

            override fun effectiveVideoConstraintsChanged(
                oldEffectiveConstraints: EffectiveConstraintsMap,
                newEffectiveConstraints: EffectiveConstraintsMap,
            ) = this@Endpoint.effectiveVideoConstraintsChanged(oldEffectiveConstraints, newEffectiveConstraints)

            override fun keyframeNeeded(endpointId: String?, ssrc: Long) =
                conference.requestKeyframe(endpointId, ssrc)
        },
        Supplier { getOrderedEndpoints() },
        diagnosticContext,
        logger,
        isUsingSourceNames,
    )

    /** Whether any sources are suspended from being sent to this endpoint because of BWE. */
    fun hasSuspendedSources() = bitrateController.hasSuspendedSources()

    /**
     * The instance which manages the Colibri messaging (over a data channel
     * or web sockets).
     */
    private val _messageTransport = EndpointMessageTransport(
        this,
        Supplier { conference.videobridge.statistics },
        conference,
        logger
    )

    override fun getMessageTransport(): EndpointMessageTransport = _messageTransport

    /**
     * Gets the endpoints in the conference in LastN order, with this {@link Endpoint} removed.
     */
    fun getOrderedEndpoints(): List<AbstractEndpoint> =
        conference.orderedEndpoints.filterNot { it == this }

    /**
     * Listen for RTT updates from [transceiver] and update the ICE stats the first time an RTT is available. Note that
     * the RTT is measured via RTCP, since we don't expose response time for STUN requests.
     */
    private val rttListener: EndpointConnectionStats.EndpointConnectionStatsListener =
        object : EndpointConnectionStats.EndpointConnectionStatsListener {
            override fun onRttUpdate(newRttMs: Double) {
                if (newRttMs > 0) {
                    transceiver.removeEndpointConnectionStatsListener(this)
                    iceTransport.updateStatsOnInitialRtt(newRttMs)
                }
            }
        }

    val transceiver = Transceiver(
        id,
        TaskPools.CPU_POOL,
        TaskPools.CPU_POOL,
        TaskPools.SCHEDULED_POOL,
        diagnosticContext,
        logger,
        TransceiverEventHandlerImpl(),
        clock
    ).apply {
        setIncomingPacketHandler(object : ConsumerNode("receiver chain handler") {
            override fun consume(packetInfo: PacketInfo) {
                this@Endpoint.handleIncomingPacket(packetInfo)
            }

            override fun trace(f: () -> Unit) = f.invoke()
        })
        addEndpointConnectionStatsListener(rttListener)
        setLocalSsrc(MediaType.AUDIO, conference.localAudioSsrc)
        setLocalSsrc(MediaType.VIDEO, conference.localVideoSsrc)
    }

    private val bandwidthProbing = BandwidthProbing(
        object : BandwidthProbing.ProbingDataSender {
            override fun sendProbing(mediaSsrcs: Collection<Long>, numBytes: Int): Int {
                return transceiver.sendProbing(mediaSsrcs, numBytes)
            }
        },
        Supplier { bitrateController.getStatusSnapshot() }
    ).apply {
        diagnosticsContext = this@Endpoint.diagnosticContext
        enabled = true
    }.also {
        recurringRunnableExecutor.registerRecurringRunnable(it)
    }

    /**
     * Manages remapping of video SSRCs when enabled.
     */
    private val videoSsrcs = VideoSsrcCache(SsrcLimitConfig.config.maxVideoSsrcs, this, logger)

    /**
     * Manages remapping of audio SSRCs when enabled.
     */
    private val audioSsrcs = AudioSsrcCache(SsrcLimitConfig.config.maxAudioSsrcs, this, logger)

    /**
     * Last advertised forwarded-sources in remapping mode.
     */
    private var activeSources: Set<String> = emptySet()

    /**
     * Track allocated send SSRCs to avoid collisions.
     */
    private val sendSsrcs = mutableSetOf<Long>()

    /**
     * Next allocated send SSRC, when allocating serially (off by default).
     */
    private var nextSendSsrc = 777000001L

    init {
        conference.encodingsManager.subscribe(this)
        setupIceTransport()
        setupDtlsTransport()

        conference.videobridge.statistics.totalEndpoints.inc()
        if (visitor) {
            conference.videobridge.statistics.totalVisitors.inc()
        }

        logger.info("Created new endpoint isUsingSourceNames=$isUsingSourceNames, iceControlling=$iceControlling")
    }

    override var mediaSources: Array<MediaSourceDesc>
        get() = transceiver.getMediaSources()
        set(value) {
            applyVideoTypeCache(value)
            val wasEmpty = transceiver.getMediaSources().isEmpty()
            if (transceiver.setMediaSources(value)) {
                eventEmitter.fireEvent { sourcesChanged() }
            }
            if (wasEmpty) {
                sendAllVideoConstraints()
            }
        }

    /**
     *  Keep track of this endpoint's audio sources.
     */
    var audioSources: ArrayList<AudioSourceDesc> = ArrayList()

    private fun setupIceTransport() {
        iceTransport.incomingDataHandler = object : IceTransport.IncomingDataHandler {
            override fun dataReceived(data: ByteArray, offset: Int, length: Int, receivedTime: Instant) {
                // DTLS data will be handled by the DtlsTransport, but SRTP data can go
                // straight to the transceiver
                if (looksLikeDtls(data, offset, length)) {
                    // DTLS transport is responsible for making its own copy, because it will manage its own
                    // buffers
                    dtlsTransport.dtlsDataReceived(data, offset, length)
                } else {
                    val copy = ByteBufferPool.getBuffer(
                        length +
                            RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET +
                            Packet.BYTES_TO_LEAVE_AT_END_OF_PACKET
                    )
                    System.arraycopy(data, offset, copy, RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET, length)
                    val pktInfo =
                        PacketInfo(UnparsedPacket(copy, RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET, length)).apply {
                            this.receivedTime = receivedTime
                        }
                    transceiver.handleIncomingPacket(pktInfo)
                }
            }
        }
        iceTransport.eventHandler = object : IceTransport.EventHandler {
            override fun connected() {
                logger.info("ICE connected")
                eventEmitter.fireEvent { iceSucceeded() }
                transceiver.setOutgoingPacketHandler(object : PacketHandler {
                    override fun processPacket(packetInfo: PacketInfo) {
                        packetInfo.addEvent(SRTP_QUEUE_ENTRY_EVENT)
                        outgoingSrtpPacketQueue.add(packetInfo)
                    }
                })
                TaskPools.IO_POOL.execute(iceTransport::startReadingData)
                TaskPools.IO_POOL.execute(dtlsTransport::startDtlsHandshake)
            }

            override fun failed() {
                eventEmitter.fireEvent { iceFailed() }
            }

            override fun consentUpdated(time: Instant) {
                transceiver.packetIOActivity.lastIceActivityInstant = time
            }
        }
    }

    /**
     * Whether this endpoint has any endpoints "on-stage".
     */
    fun isInStageView() = bitrateController.allocationSettings.onStageEndpoints.isNotEmpty()

    private fun setupDtlsTransport() {
        dtlsTransport.incomingDataHandler = object : DtlsTransport.IncomingDataHandler {
            override fun dtlsAppDataReceived(buf: ByteArray, off: Int, len: Int) {
                this@Endpoint.dtlsAppPacketReceived(buf, off, len)
            }
        }
        dtlsTransport.outgoingDataHandler = object : DtlsTransport.OutgoingDataHandler {
            override fun sendData(buf: ByteArray, off: Int, len: Int) {
                iceTransport.send(buf, off, len)
            }
        }
        dtlsTransport.eventHandler = object : DtlsTransport.EventHandler {
            override fun handshakeComplete(
                chosenSrtpProtectionProfile: Int,
                tlsRole: TlsRole,
                keyingMaterial: ByteArray
            ) {
                logger.info("DTLS handshake complete")
                transceiver.setSrtpInformation(chosenSrtpProtectionProfile, tlsRole, keyingMaterial)
                // TODO(brian): the old code would work even if the sctp connection was created after
                //  the handshake had completed, but this won't (since this is a one-time event).  do
                //  we need to worry about that case?
                sctpSocket.ifPresent(::acceptSctpConnection)
                scheduleEndpointMessageTransportTimeout()
            }
        }
    }

    fun updateForceMute(audioForceMuted: Boolean, videoForceMuted: Boolean) {
        transceiver.forceMuteAudio(audioForceMuted)
        transceiver.forceMuteVideo(videoForceMuted)
    }

    override fun addPayloadType(payloadType: PayloadType) {
        transceiver.addPayloadType(payloadType)
        bitrateController.addPayloadType(payloadType)
    }

    override fun addRtpExtension(rtpExtension: RtpExtension) = transceiver.addRtpExtension(rtpExtension)

    fun addReceiveSsrc(ssrc: Long, mediaType: MediaType) {
        logger.cdebug { "Adding receive ssrc $ssrc of type $mediaType" }
        transceiver.addReceiveSsrc(ssrc, mediaType)
        conference.addEndpointSsrc(this, ssrc)
    }

    override fun onNewSsrcAssociation(
        endpointId: String,
        primarySsrc: Long,
        secondarySsrc: Long,
        type: SsrcAssociationType
    ) {
        if (endpointId.equals(id, ignoreCase = true)) {
            transceiver.addSsrcAssociation(LocalSsrcAssociation(primarySsrc, secondarySsrc, type))
        } else {
            transceiver.addSsrcAssociation(RemoteSsrcAssociation(primarySsrc, secondarySsrc, type))
        }
    }

    fun setFeature(feature: EndpointDebugFeatures, enabled: Boolean) {
        when (feature) {
            EndpointDebugFeatures.PCAP_DUMP -> transceiver.setFeature(Features.TRANSCEIVER_PCAP_DUMP, enabled)
        }
    }

    fun isFeatureEnabled(feature: EndpointDebugFeatures): Boolean {
        return when (feature) {
            EndpointDebugFeatures.PCAP_DUMP -> transceiver.isFeatureEnabled(Features.TRANSCEIVER_PCAP_DUMP)
        }
    }

    override fun isSendingAudio(): Boolean {
        // The endpoint is sending audio if we (the transceiver) are receiving audio.
        return transceiver.isReceivingAudio()
    }
    override fun isSendingVideo(): Boolean {
        // The endpoint is sending video if we (the transceiver) are receiving video.
        return transceiver.isReceivingVideo()
    }

    private fun doSendSrtp(packetInfo: PacketInfo): Boolean {
        packetInfo.addEvent(SRTP_QUEUE_EXIT_EVENT)
        PacketTransitStats.packetSent(packetInfo)

        packetInfo.sent()
        if (timelineLogger.isTraceEnabled && logTimeline()) {
            timelineLogger.trace { packetInfo.timeline.toString() }
        }
        iceTransport.send(packetInfo.packet.buffer, packetInfo.packet.offset, packetInfo.packet.length)
        ByteBufferPool.returnBuffer(packetInfo.packet.buffer)
        return true
    }

    /**
     * Notifies this endpoint that the ordered list of last-n endpoints has changed
     */
    fun lastNEndpointsChanged() = bitrateController.endpointOrderingChanged()

    /**
     * {@inheritDoc}
     */
    override fun sendMessage(msg: BridgeChannelMessage) = messageTransport.sendMessage(msg)

    // TODO: this should be part of an EndpointMessageTransport.EventHandler interface
    fun endpointMessageTransportConnected() {
        sendAllVideoConstraints()
        videoSsrcs.sendAllMappings()
        audioSsrcs.sendAllMappings()
    }

    private fun sendAllVideoConstraints() {
        maxReceiverVideoConstraintsMap.forEach { (sourceName, constraints) ->
            sendVideoConstraintsV2(sourceName, constraints)
        }
    }

    /**
     * Handle a DTLS app packet (that is, a packet of some other protocol sent
     * over DTLS) which has just been received.
     */
    // TODO(brian): change sctp handler to take buf, off, len
    fun dtlsAppPacketReceived(data: ByteArray, off: Int, len: Int) =
        sctpHandler.processPacket(PacketInfo(UnparsedPacket(data, off, len)))

    private fun effectiveVideoConstraintsChanged(
        oldEffectiveConstraints: EffectiveConstraintsMap,
        newEffectiveConstraints: EffectiveConstraintsMap
    ) {
        val removedSources = oldEffectiveConstraints.keys.filterNot { it in newEffectiveConstraints.keys }

        // Sources that "this" endpoint no longer receives.
        removedSources.map { it.sourceName }.forEach { removedSourceName ->
            // Remove ourself as a receiver from that endpoint
            conference.findSourceOwner(removedSourceName)?.removeSourceReceiver(removedSourceName, id)
        }

        // Added or updated
        newEffectiveConstraints.forEach { (source, effectiveConstraints) ->
            conference.findSourceOwner(source.sourceName)?.addReceiver(id, source.sourceName, effectiveConstraints)
        }

        if (doSsrcRewriting) {
            val newActiveSources =
                newEffectiveConstraints.entries.filter { !it.value.isDisabled() }.map { it.key }.toList()
            val newActiveSourceNames = newActiveSources.mapNotNull { it.sourceName }.toSet()
            /* safe unlocked access of activeSources. BitrateController will not overlap calls to this method. */
            if (activeSources != newActiveSourceNames) {
                activeSources = newActiveSourceNames
                videoSsrcs.activate(newActiveSources)
            }
        }
    }

    override fun sendVideoConstraints(maxVideoConstraints: VideoConstraints) {
        // Note that it's up to the client to respect these constraints.
        if (mediaSources.isEmpty()) {
            logger.cdebug { "Suppressing sending a SenderVideoConstraints message, endpoint has no streams." }
        } else {
            val senderVideoConstraintsMessage = SenderVideoConstraintsMessage(maxVideoConstraints.maxHeight)
            logger.cdebug { "Sender constraints changed: ${senderVideoConstraintsMessage.toJson()}" }
            sendMessage(senderVideoConstraintsMessage)
        }
    }

    override fun sendVideoConstraintsV2(sourceName: String, maxVideoConstraints: VideoConstraints) {
        // Note that it's up to the client to respect these constraints.
        if (findMediaSourceDesc(sourceName) == null) {
            logger.warn {
                "Suppressing sending a SenderVideoConstraints message, endpoint has no such source: $sourceName"
            }
        } else {
            if (isUsingSourceNames) {
                val senderSourceConstraintsMessage =
                    SenderSourceConstraintsMessage(sourceName, maxVideoConstraints.maxHeight)
                logger.cdebug { "Sender constraints changed: ${senderSourceConstraintsMessage.toJson()}" }
                sendMessage(senderSourceConstraintsMessage)
            } else {
            // holding a lock inside the SctpSocket which can cause a deadlock
            // if two endpoints are trying to send datachannel messages to one
            // another (with stats broadcasting it can happen often)
            TaskPools.IO_POOL.execute { dataChannelHandler.processPacket(PacketInfo(dataChannelPacket)) }
        }
        socket.listen()
        sctpSocket = Optional.of(socket)

    /**
     * Create an SCTP connection for this Endpoint.  If [openDataChannelLocally] is true,
     * we will create the data channel locally, otherwise we will wait for the remote side
     * to open it.
     */
    fun createSctpConnection() {
        logger.cdebug { "Creating SCTP manager" }
        // Create the SctpManager and provide it a method for sending SCTP data
        sctpManager = SctpManager(
            { data, offset, length ->
                dtlsTransport.sendDtlsData(data, offset, length)
                0
            },
            logger
        )
        sctpHandler.setSctpManager(sctpManager!!)
        // NOTE(brian): as far as I know we always act as the 'server' for sctp
        // connections, but if not we can make which type we use dynamic
        val socket = sctpManager!!.createServerSocket()
        socket.eventHandler = object : SctpSocket.SctpSocketEventHandler {
            override fun onReady() {
                logger.info("SCTP connection is ready, creating the Data channel stack")
                dataChannelStack = DataChannelStack(
                    { data, sid, ppid -> socket.send(data, true, sid, ppid) },
                    logger
                )
                // This handles if the remote side will be opening the data channel
                dataChannelStack!!.onDataChannelStackEvents { dataChannel ->
                    logger.info("Remote side opened a data channel.")
                    _messageTransport.setDataChannel(dataChannel)
                }
                dataChannelHandler.setDataChannelStack(dataChannelStack!!)
                if (openDataChannelLocally) {
                    // This logic is for opening the data channel locally
                    logger.info("Will open the data channel.")
                    val dataChannel = dataChannelStack!!.createDataChannel(
                        DataChannelProtocolConstants.RELIABLE,
                        0,
                        0,
                        0,
                        "default"
                    )
                    _messageTransport.setDataChannel(dataChannel)
                    dataChannel.open()
                } else {
                    logger.info("Will wait for the remote side to open the data channel.")
                }
            }

            override fun onDisconnected() {
                logger.info("SCTP connection is disconnected")
            }
        }
        socket.dataCallback = SctpDataCallback { data, sid, ssn, tsn, ppid, context, flags ->
            // We assume all data coming over SCTP will be datachannel data
            val dataChannelPacket = DataChannelPacket(data, 0, data.size, sid, ppid.toInt())
            // Post the rest of the task here because the current context is
            // holding a lock inside the SctpSocket which can cause a deadlock
            // if two endpoints are trying to send datachannel messages to one
            // another (with stats broadcasting it can happen often)
            incomingDataChannelMessagesQueue.add(PacketInfo(dataChannelPacket))
        }
        socket.listen()
        sctpSocket = Optional.of(socket)
    }

    fun acceptSctpConnection(sctpServerSocket: SctpServerSocket) {
        TaskPools.IO_POOL.execute {
            // We don't want to block the thread calling
            // onDtlsHandshakeComplete so run the socket acceptance in an IO
            // pool thread
            // FIXME: This runs forever once the socket is closed (
            // accept never returns true).
            logger.info("Attempting to establish SCTP socket connection")
            var attempts = 0
            while (!sctpServerSocket.accept()) {
                attempts++
                try {
                    Thread.sleep(100)
                } catch (e: InterruptedException) {
                    break
                }
                if (attempts > 100) {
                    logger.error("Timed out waiting for SCTP connection from remote side")
                    break
                }
            }
            logger.cdebug { "SCTP socket ${sctpServerSocket.hashCode()} accepted connection" }
        }
    }

    /**
     * Schedule a timeout to fire log a message and track a stat if we don't
     * have an endpoint message transport connected within the timeout.
     */
    fun scheduleEndpointMessageTransportTimeout() {
        TaskPools.SCHEDULED_POOL.schedule(
            {
                if (!isExpired) {
                    if (!messageTransport.isConnected) {
                        logger.error("EndpointMessageTransport still not connected.")
                        conference.videobridge.statistics.numEndpointsNoMessageTransportAfterDelay.inc()
                    }
                }
            },
            30,
            TimeUnit.SECONDS
        )
    }

    /**
     * Checks whether a WebSocket connection with a specific password string
     * should be accepted for this endpoint.
     * @param password the
     * @return {@code true} iff the password matches.
     */
    fun acceptWebSocket(password: String): Boolean {
        if (iceTransport.icePassword != password) {
            logger.warn(
                "Incoming web socket request with an invalid password. " +
                    "Expected: ${iceTransport.icePassword} received $password"
            )
            return false
        }
        return true
    }

    /**
     * Sends a message to this endpoint in order to notify it that the set of endpoints for which the bridge
     * is sending video has changed.
     *
     * @param forwardedEndpoints the collection of forwarded endpoints.
     */
    @Deprecated("", ReplaceWith("sendForwardedSourcesMessage"), DeprecationLevel.WARNING)
    fun sendForwardedEndpointsMessage(forwardedEndpoints: Collection<String>) {
        if (isUsingSourceNames) {
            return
        }

        val msg = ForwardedEndpointsMessage(forwardedEndpoints)
        TaskPools.IO_POOL.execute {
            try {
                sendMessage(msg)
            } catch (t: Throwable) {
                logger.warn("Failed to send message:", t)
            }
        }
    }

    /**
     * Sends a message to this endpoint in order to notify it that the set of media sources for which the bridge
     * is sending video has changed.
     *
     * @param forwardedSources the collection of forwarded media sources (by name).
     */
    fun sendForwardedSourcesMessage(forwardedSources: Collection<String>) {
        if (!isUsingSourceNames) {
            return
        }

        val msg = ForwardedSourcesMessage(forwardedSources)
        TaskPools.IO_POOL.execute {
            try {
                sendMessage(msg)
            } catch (t: Throwable) {
                logger.warn("Failed to send message:", t)
            }
        }
    }

    /**
     * Sets the remote transport information (ICE candidates, DTLS fingerprints).
     *
     * @param transportInfo the XML extension which contains the remote
     * transport information.
     */
    fun setTransportInfo(transportInfo: IceUdpTransportPacketExtension) {
        val remoteFingerprints = mutableMapOf<String, String>()
        val fingerprintExtensions = transportInfo.getChildExtensionsOfType(DtlsFingerprintPacketExtension::class.java)
        fingerprintExtensions.forEach { fingerprintExtension ->
            if (fingerprintExtension.hash != null && fingerprintExtension.fingerprint != null) {
                remoteFingerprints[fingerprintExtension.hash] = fingerprintExtension.fingerprint
            } else {
                logger.info("Ignoring empty DtlsFingerprint extension: ${transportInfo.toXML()}")
            }
        }
        dtlsTransport.setRemoteFingerprints(remoteFingerprints)
        if (fingerprintExtensions.isNotEmpty()) {
            val setup = fingerprintExtensions.first().setup
            dtlsTransport.setSetupAttribute(setup)
        }
        iceTransport.startConnectivityEstablishment(transportInfo)
    }

    fun describeTransport(): IceUdpTransportPacketExtension {
        val iceUdpTransportPacketExtension = IceUdpTransportPacketExtension()
        iceTransport.describe(iceUdpTransportPacketExtension)
        dtlsTransport.describe(iceUdpTransportPacketExtension)
        colibriWebSocketServiceSupplier.get()?.let { colibriWebsocketService ->
            colibriWebsocketService.getColibriWebSocketUrl(
                conference.id,
                id,
                iceTransport.icePassword
            )?.let { wsUrl ->
                val wsPacketExtension = WebSocketPacketExtension(wsUrl)
                iceUdpTransportPacketExtension.addChildExtension(wsPacketExtension)
            }
        }

        logger.cdebug { "Transport description:\n${iceUdpTransportPacketExtension.toXML()}" }

        return iceUdpTransportPacketExtension
    }

    /**
     * Handle incoming RTP packets which have been fully processed by the
     * transceiver's incoming pipeline.
     */
    fun handleIncomingPacket(packetInfo: PacketInfo) {
        if (visitor) {
            /* Never forward RTP/RTCP from a visitor. */
            ByteBufferPool.returnBuffer(packetInfo.packet.buffer)
            return
        }

        packetInfo.endpointId = id
        conference.handleIncomingPacket(packetInfo)
    }

    override fun receivesSsrc(ssrc: Long): Boolean = transceiver.receivesSsrc(ssrc)

    fun doesSsrcRewriting(): Boolean = doSsrcRewriting

    fun unmapRtcpFbSsrc(packet: RtcpFbPacket) = videoSsrcs.unmapRtcpFbSsrc(packet)

    override fun getSsrcs() = HashSet(transceiver.receiveSsrcs)

    override fun getLastIncomingActivity(): Instant = transceiver.packetIOActivity.lastIncomingActivityInstant

    override fun requestKeyframe() = transceiver.requestKeyFrame()

    override fun requestKeyframe(mediaSsrc: Long) = transceiver.requestKeyFrame(mediaSsrc)

    /** Whether we are currently oversending to this endpoint. */
    fun isOversending(): Boolean = bitrateController.isOversending()

    /**
     * Returns how many endpoints this Endpoint is currently forwarding video for
     */
    fun numForwardedEndpoints(): Int = bitrateController.numForwardedEndpoints()

    fun setBandwidthAllocationSettings(message: ReceiverVideoConstraintsMessage) {
        bitrateController.setBandwidthAllocationSettings(message)
    }

    /**
     * {@inheritDoc}
     */
    override fun findVideoSourceProps(ssrc: Long): MediaSourceDesc? {
        conference.getEndpointBySsrc(ssrc)?.let { ep ->
            ep.mediaSources.forEach { s ->
                if (s.findRtpEncodingDesc(ssrc) != null) {
                    return s
                }
            }
        }
        logger.error { "No properties found for SSRC $ssrc." }
        return null
    }

    /**
     * {@inheritDoc}
     */
    override fun findAudioSourceProps(ssrc: Long): AudioSourceDesc? {
        conference.getEndpointBySsrc(ssrc)?.let { ep ->
            if (ep !is Endpoint)
                return null
            return ep.audioSources.find { s -> s.ssrc == ssrc }
        }
        logger.error { "No properties found for SSRC $ssrc." }
        return null
    }

    /**
     * {@inheritDoc}
     */
    override fun getNextSendSsrc(): Long {
        synchronized(sendSsrcs) {
            while (true) {
                val ssrc = if (useRandomSendSsrcs)
                    random.nextLong().and(0xFFFF_FFFFL)
                else
                    nextSendSsrc++
                if (sendSsrcs.add(ssrc))
                    return ssrc
            }
        }
    }

    override fun send(packetInfo: PacketInfo) {
        when (val packet = packetInfo.packet) {
            is VideoRtpPacket -> {
                if (bitrateController.transformRtp(packetInfo)) {
                    // The original packet was transformed in place.
                    if (doSsrcRewriting) {
                        val start = packet !is ParsedVideoPacket || (packet.isKeyframe && packet.isStartOfFrame)
                        if (!videoSsrcs.rewriteRtp(packet, start)) {
                            return
                        }
                    }
                    transceiver.sendPacket(packetInfo)
                } else {
                    logger.warn("Dropping a packet which was supposed to be accepted:$packet")
                }
                return
            }
            is AudioRtpPacket -> if (doSsrcRewriting) audioSsrcs.rewriteRtp(packet)
            is RtcpSrPacket -> {
                // Allow the BC to update the timestamp (in place).
                bitrateController.transformRtcp(packet)
                if (doSsrcRewriting) {
                    // Just check both tables instead of looking up the type first.
                    if (!videoSsrcs.rewriteRtcp(packet) && !audioSsrcs.rewriteRtcp(packet))
                        return
                }
                logger.trace {
                    "relaying an sr from ssrc=${packet.senderSsrc}, timestamp=${packet.senderInfo.rtpTimestamp}"
                }
            }
        }
        transceiver.sendPacket(packetInfo)
    }

    /**
     * To find out whether the endpoint should be expired, we check the activity timestamps from the transceiver.
     */
    override fun shouldExpire(): Boolean {
        if (iceTransport.hasFailed()) {
            logger.warn("Allowing to expire because ICE failed.")
            return true
        }

        val lastActivity = lastIncomingActivity
        val now = clock.instant()

        if (lastActivity == NEVER) {
            val timeSinceCreation = Duration.between(creationTime, now)
            if (timeSinceCreation > epTimeout) {
                logger.info(
                    "Endpoint's ICE connection has neither failed nor connected " +
                        "after $timeSinceCreation expiring"
                )
                return true
            }
            // We haven't seen any activity yet. If this continues ICE will
            // eventually fail (which is handled above).
            return false
        }

        val expireTimeout = VideobridgeExpireThreadConfig.config.inactivityTimeout
        if (Duration.between(lastActivity, now) > expireTimeout) {
            logger.info("Allowing to expire because of no activity in over $expireTimeout")
            return true
        }
        return false
    }

    fun setLastN(lastN: Int) {
        bitrateController.lastN = lastN
    }

    /**
     * Returns true if this endpoint's transport is 'fully' connected (both ICE and DTLS), false otherwise
     */
    private fun isTransportConnected(): Boolean = iceTransport.isConnected() && dtlsTransport.isConnected

    fun getRtt(): Double = transceiver.getTransceiverStats().endpointConnectionStats.rtt

    override fun wants(packetInfo: PacketInfo): Boolean {
        if (!isTransportConnected()) {
            return false
        }

        return when (val packet = packetInfo.packet) {
            is VideoRtpPacket -> acceptVideo && bitrateController.accept(packetInfo)
            is AudioRtpPacket -> acceptAudio
            is RtcpSrPacket -> {
                // TODO: For SRs we're only interested in the ntp/rtp timestamp
                //  association, so we could only accept srs from the main ssrc
                bitrateController.accept(packet)
            }
            is RtcpFbPliPacket, is RtcpFbFirPacket -> {
                // We assume that we are only given PLIs/FIRs destined for this
                // endpoint. This is because Conference has to find the target
                // endpoint (this endpoint) anyway, and we would essentially be
                // performing the same check twice.
                true
            }
            else -> {
                logger.warn("Ignoring an unknown packet type:" + packet.javaClass.simpleName)
                false
            }
        }
    }

    /**
     * Determine whether to forward endpoint stats from another endpoint to this one.
     */
    fun wantsStatsFrom(ep: AbstractEndpoint): Boolean {
        // Always forward stats in small conferences
        if (conference.endpointCount <= statsFilterThreshold) {
            return true
        }
        logger.debug {
            buildString {
                append("wantsStatsFrom(${ep.id}): isRecentSpeaker=${conference.speechActivity.isRecentSpeaker(ep)} ")
                append("isRankedSpeaker=${conference.isRankedSpeaker(ep)} ")
                if (ep.mediaSources.isEmpty()) {
                    append("(no media sources)")
                }
                ep.mediaSources.forEach { source ->
                    val name = source.sourceName
                    append("isOnStageOrSelected($name)=${bitrateController.isOnStageOrSelected(source)} ")
                    append("hasNonZeroEffectiveConstraints($name)=")
                    append("${bitrateController.hasNonZeroEffectiveConstraints(source)} ")
                }
            }
        }

        if (conference.speechActivity.isRecentSpeaker(ep) || conference.isRankedSpeaker(ep)) {
            return true
        }
        return ep.mediaSources.any { source ->
            bitrateController.isOnStageOrSelected(source) ||
                bitrateController.hasNonZeroEffectiveConstraints(source)
        }
    }

    /**
     * Updates the conference statistics with value from this endpoint. Since
     * the values are cumulative this should execute only once when the endpoint
     * expires.
     */
    private fun updateStatsOnExpire() {
        val conferenceStats = conference.statistics
        val transceiverStats = transceiver.getTransceiverStats()

        conferenceStats.apply {
            val incomingStats = transceiverStats.rtpReceiverStats.packetStreamStats
            val outgoingStats = transceiverStats.outgoingPacketStreamStats
            totalBytesReceived.addAndGet(incomingStats.bytes)
            totalPacketsReceived.addAndGet(incomingStats.packets)
            totalBytesSent.addAndGet(outgoingStats.bytes)
            totalPacketsSent.addAndGet(outgoingStats.packets)
        }

        conference.videobridge.statistics.apply {
            val bweStats = transceiverStats.bandwidthEstimatorStats
            bweStats.getNumber("incomingEstimateExpirations")?.toLong()?.let {
                incomingBitrateExpirations.addAndGet(it)
            }
            keyframesReceived.addAndGet(transceiverStats.rtpReceiverStats.videoParserStats.numKeyframes.toLong())
            layeringChangesReceived.addAndGet(
                transceiverStats.rtpReceiverStats.videoParserStats.numLayeringChanges.toLong()
            )

            val durationActiveVideo = transceiverStats.rtpReceiverStats.incomingStats.ssrcStats.values.filter {
                it.mediaType == MediaType.VIDEO
            }.sumOf { it.durationActive }
            totalVideoStreamMillisecondsReceived.addAndGet(durationActiveVideo.toMillis())
        }

        run {
            val bweStats = transceiverStats.bandwidthEstimatorStats
            val lossLimitedMs = bweStats.getNumber("lossLimitedMs")?.toLong() ?: return@run
            val lossDegradedMs = bweStats.getNumber("lossDegradedMs")?.toLong() ?: return@run
            val lossFreeMs = bweStats.getNumber("lossFreeMs")?.toLong() ?: return@run

            val participantMs = lossFreeMs + lossDegradedMs + lossLimitedMs
            conference.videobridge.statistics.apply {
                totalLossControlledParticipantMs.addAndGet(participantMs)
                totalLossLimitedParticipantMs.addAndGet(lossLimitedMs)
                totalLossDegradedParticipantMs.addAndGet(lossDegradedMs)
            }
        }

        if (iceTransport.isConnected() && !dtlsTransport.isConnected) {
            logger.info("Expiring an endpoint with ICE connected, but not DTLS.")
            conferenceStats.dtlsFailedEndpoints.incrementAndGet()
        }
    }

    override fun getDebugState(): JSONObject {
        return super.getDebugState().apply {
            put("bitrateController", bitrateController.debugState)
            put("bandwidthProbing", bandwidthProbing.getDebugState())
            put("iceTransport", iceTransport.getDebugState())
            put("dtlsTransport", dtlsTransport.getDebugState())
            put("transceiver", transceiver.getNodeStats().toJson())
            put("acceptAudio", acceptAudio)
            put("acceptVideo", acceptVideo)
            put("visitor", visitor)
            put("messageTransport", messageTransport.debugState)
            if (doSsrcRewriting) {
                put("audioSsrcs", audioSsrcs.getDebugState())
                put("videoSsrcs", videoSsrcs.getDebugState())
            }
        }
    }

    override fun expire() {
        if (super.isExpired()) {
            return
        }
        super.expire()

        try {
            bitrateController.expire()
            updateStatsOnExpire()
            transceiver.stop()
            logger.cdebug { transceiver.getNodeStats().prettyPrint(0) }
            logger.cdebug { bitrateController.debugState.toJSONString() }
            logger.cdebug { iceTransport.getDebugState().toJSONString() }
            logger.cdebug { dtlsTransport.getDebugState().toJSONString() }

            logger.info("Spent ${bitrateController.getTotalOversendingTime().seconds} seconds oversending")
        private var dataChannelStack: DataChannelStack? = null
        private val cachedDataChannelPackets = LinkedBlockingQueue<PacketInfo>()

        override fun consume(packetInfo: PacketInfo) {
            synchronized(dataChannelStackLock) {
                when (val packet = packetInfo.packet) {
                    is DataChannelPacket -> {
        }

        bandwidthProbing.enabled = false
        recurringRunnableExecutor.deRegisterRecurringRunnable(bandwidthProbing)
        conference.encodingsManager.unsubscribe(this)

        dtlsTransport.stop()
        iceTransport.stop()
        outgoingSrtpPacketQueue.close()

        logger.info("Expired.")
    }

    companion object {
        /**
         * Whether or not the bridge should be the peer which opens the data channel
         * (as opposed to letting the far peer/client open it).
         */
        private const val openDataChannelLocally = false

        /**
         * Count the number of dropped packets and exceptions.
         */
        @JvmField
        val queueErrorCounter = CountingErrorHandler()

        /**
         * The executor which runs bandwidth probing.
         *
         * TODO (brian): align the recurringRunnable stuff with whatever we end up
         * doing with all the other executors.
         */
        private val recurringRunnableExecutor = RecurringRunnableExecutor(Endpoint::class.java.simpleName)

        /**
         * How long we'll give an endpoint to either successfully establish
         * an ICE connection or fail before we expire it.
         */
        // TODO: make this configurable
        private val epTimeout = 2.mins

        private val timelineCounter = AtomicLong()
        private val TIMELINE_FRACTION = 10000L

        fun logTimeline() = timelineCounter.getAndIncrement() % TIMELINE_FRACTION == 0L

        private const val SRTP_QUEUE_ENTRY_EVENT = "Entered Endpoint SRTP sender outgoing queue"
        private const val SRTP_QUEUE_EXIT_EVENT = "Exited Endpoint SRTP sender outgoing queue"

        private val statsFilterThreshold: Int by config {
            "videobridge.stats-filter-threshold".from(JitsiConfig.newConfig)
        }

        /**
         * This can be switched off to ease debugging.
         */
        private val useRandomSendSsrcs = true

        /**
         * Used for generating send SSRCs.
         */
        private val random = SecureRandom()
    }

    private inner class TransceiverEventHandlerImpl : TransceiverEventHandler {
        /**
         * Forward audio level events from the Transceiver to the conference. We use the same thread, because this fires
         * for every packet and we want to avoid the switch. The conference audio level code must not block.
         */
        override fun audioLevelReceived(sourceSsrc: Long, level: Long): Boolean =
            conference.levelChanged(this@Endpoint, level)

        /**
         * Forward bwe events from the Transceiver.
         */
        override fun bandwidthEstimationChanged(newValue: Bandwidth) {
            logger.cdebug { "Estimated bandwidth is now $newValue" }
            bitrateController.bandwidthChanged(newValue.bps.toLong())
            bandwidthProbing.bandwidthEstimationChanged(newValue)
        }
    }

    /**
     * A node which can be placed in the pipeline to cache Data channel packets
     * until the DataChannelStack is ready to handle them.
     */
    private class DataChannelHandler : ConsumerNode("Data channel handler") {
        private val dataChannelStackLock = Any()
        private var dataChannelStack: DataChannelStack? = null
        private val cachedDataChannelPackets = LinkedBlockingQueue<PacketInfo>()

        public override fun consume(packetInfo: PacketInfo) {
            synchronized(dataChannelStackLock) {
                when (val packet = packetInfo.packet) {
                    is DataChannelPacket -> {
                        dataChannelStack?.onIncomingDataChannelPacket(
                            ByteBuffer.wrap(packet.buffer), packet.sid, packet.ppid
                        ) ?: run {
                            cachedDataChannelPackets.add(packetInfo)
                        }
                    }
                    else -> Unit
                }
            }
        }

        fun setDataChannelStack(dataChannelStack: DataChannelStack) {
            // Submit this to the pool since we wait on the lock and process any
            // cached packets here as well

            // Submit this to the pool since we wait on the lock and process any
            // cached packets here as well
            TaskPools.IO_POOL.execute {
                // We grab the lock here so that we can set the SCTP manager and
                // process any previously-cached packets as an atomic operation.
                // It also prevents another thread from coming in via
                // #doProcessPackets and processing packets at the same time in
                // another thread, which would be a problem.
                synchronized(dataChannelStackLock) {
                    this.dataChannelStack = dataChannelStack
                    cachedDataChannelPackets.forEach {
                        val dcp = it.packet as DataChannelPacket
                        dataChannelStack.onIncomingDataChannelPacket(
                            ByteBuffer.wrap(dcp.buffer), dcp.sid, dcp.ppid
                        )
                    }
                }
            }
        }

        override fun trace(f: () -> Unit) = f.invoke()
    }

    /**
     * A node which can be placed in the pipeline to cache SCTP packets until
     * the SCTPManager is ready to handle them.
     */
    private class SctpHandler : ConsumerNode("SCTP handler") {
        private val sctpManagerLock = Any()
        private var sctpManager: SctpManager? = null
        private val numCachedSctpPackets = AtomicLong(0)
        private val cachedSctpPackets = LinkedBlockingQueue<PacketInfo>(100)

        override fun consume(packetInfo: PacketInfo) {
            synchronized(sctpManagerLock) {
                if (SctpConfig.config.enabled) {
                    sctpManager?.handleIncomingSctp(packetInfo) ?: run {
                        numCachedSctpPackets.incrementAndGet()
                        cachedSctpPackets.add(packetInfo)
                    }
                }
            }
        }

        override fun getNodeStats(): NodeStatsBlock = super.getNodeStats().apply {
            addNumber("num_cached_packets", numCachedSctpPackets.get())
        }

        fun setSctpManager(sctpManager: SctpManager) {
            // Submit this to the pool since we wait on the lock and process any
            // cached packets here as well
            TaskPools.IO_POOL.execute {
                // We grab the lock here so that we can set the SCTP manager and
                // process any previously-cached packets as an atomic operation.
                // It also prevents another thread from coming in via
                // #doProcessPackets and processing packets at the same time in
                // another thread, which would be a problem.
                synchronized(sctpManagerLock) {
                    this.sctpManager = sctpManager
                    cachedSctpPackets.forEach { sctpManager.handleIncomingSctp(it) }
                    cachedSctpPackets.clear()
                }
            }
        }

        override fun trace(f: () -> Unit) = f.invoke()
    }
}

<MSG> Process data channel messages sequentially (#1665)

This prevents SCTP DataChannel messages from being processed out of order, causing warnings on channel initialization and potential issues with Last-N selection.
<DFF> @@ -180,6 +180,20 @@ class Endpoint @JvmOverloads constructor(
         setErrorHandler(queueErrorCounter)
     }
 
+    /**
+     * The queue which enforces sequential processing of incoming data channel messages
+     * to maintain processing order.
+     */
+    private val incomingDataChannelMessagesQueue = PacketInfoQueue(
+        "${javaClass.simpleName}-incoming-data-channel-queue",
+        TaskPools.IO_POOL,
+        { packetInfo ->
+            dataChannelHandler.consume(packetInfo)
+            true
+        },
+        TransportConfig.queueSize
+    )
+
     private val bitrateController = BitrateController(
         object : BitrateController.EventHandler {
             override fun allocationChanged(allocation: BandwidthAllocation) {
@@ -565,7 +579,7 @@ class Endpoint @JvmOverloads constructor(
             // holding a lock inside the SctpSocket which can cause a deadlock
             // if two endpoints are trying to send datachannel messages to one
             // another (with stats broadcasting it can happen often)
-            TaskPools.IO_POOL.execute { dataChannelHandler.processPacket(PacketInfo(dataChannelPacket)) }
+            incomingDataChannelMessagesQueue.add(PacketInfo(dataChannelPacket))
         }
         socket.listen()
         sctpSocket = Optional.of(socket)
@@ -1099,7 +1113,7 @@ class Endpoint @JvmOverloads constructor(
         private var dataChannelStack: DataChannelStack? = null
         private val cachedDataChannelPackets = LinkedBlockingQueue<PacketInfo>()
 
-        override fun consume(packetInfo: PacketInfo) {
+        public override fun consume(packetInfo: PacketInfo) {
             synchronized(dataChannelStackLock) {
                 when (val packet = packetInfo.packet) {
                     is DataChannelPacket -> {

{'commit': 'diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt\nindex 2da72053..c2e51d23 100644\n--- a/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt\n+++ b/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt\n@@ -180,6 +180,20 @@ class Endpoint @JvmOverloads constructor(\n         setErrorHandler(queueErrorCounter)\n     }\n \n+    /**\n+     * The queue which enforces sequential processing of incoming data channel messages\n+     * to maintain processing order.\n+     */\n+    private val incomingDataChannelMessagesQueue = PacketInfoQueue(\n+        "${javaClass.simpleName}-incoming-data-channel-queue",\n+        TaskPools.IO_POOL,\n+        { packetInfo ->\n+            dataChannelHandler.consume(packetInfo)\n+            true\n+        },\n+        TransportConfig.queueSize\n+    )\n+\n     private val bitrateController = BitrateController(\n         object : BitrateController.EventHandler {\n             override fun allocationChanged(allocation: BandwidthAllocation) {\n@@ -565,7 +579,7 @@ class Endpoint @JvmOverloads constructor(\n             // holding a lock inside the SctpSocket which can cause a deadlock\n             // if two endpoints are trying to send datachannel messages to one\n             // another (with stats broadcasting it can happen often)\n-            TaskPools.IO_POOL.execute { dataChannelHandler.processPacket(PacketInfo(dataChannelPacket)) }\n+            incomingDataChannelMessagesQueue.add(PacketInfo(dataChannelPacket))\n         }\n         socket.listen()\n         sctpSocket = Optional.of(socket)\n@@ -1099,7 +1113,7 @@ class Endpoint @JvmOverloads constructor(\n         private var dataChannelStack: DataChannelStack? = null\n         private val cachedDataChannelPackets = LinkedBlockingQueue<PacketInfo>()\n \n-        override fun consume(packetInfo: PacketInfo) {\n+        public override fun consume(packetInfo: PacketInfo) {\n             synchronized(dataChannelStackLock) {\n                 when (val packet = packetInfo.packet) {\n                     is DataChannelPacket -> {\n', 'commit_subject': 'Process data channel messages sequentially (#1665)', 'repo_name': 'jitsi/jitsi-videobridge', 'committer': {'name': 'GitHub', 'email': '1505422b2465e9a84f6fdfaa161078890c593f06@github.com', 'time_sec': '1623051763', 'tz_offset': '180', 'date': {'seconds': '1623051763'}}, 'commit_tree': 'b1cfc1a8575a0447f19ef922fa7dd47b00ea6a60', 'commit_message': 'Process data channel messages sequentially (#1665)\n\nThis prevents SCTP DataChannel messages from being processed out of order, causing warnings on channel initialization and potential issues with Last-N selection.', 'language_name': 'HTML', 'license': 'apache-2.0', 'hunks': '@@ -180,6 +180,20 @@ class Endpoint @JvmOverloads constructor(\n         setErrorHandler(queueErrorCounter)\n     }\n \n+    /**\n+     * The queue which enforces sequential processing of incoming data channel messages\n+     * to maintain processing order.\n+     */\n+    private val incomingDataChannelMessagesQueue = PacketInfoQueue(\n+        "${javaClass.simpleName}-incoming-data-channel-queue",\n+        TaskPools.IO_POOL,\n+        { packetInfo ->\n+            dataChannelHandler.consume(packetInfo)\n+            true\n+        },\n+        TransportConfig.queueSize\n+    )\n+\n     private val bitrateController = BitrateController(\n         object : BitrateController.EventHandler {\n             override fun allocationChanged(allocation: BandwidthAllocation) {\n@@ -565,7 +579,7 @@ class Endpoint @JvmOverloads constructor(\n             // holding a lock inside the SctpSocket which can cause a deadlock\n             // if two endpoints are trying to send datachannel messages to one\n             // another (with stats broadcasting it can happen often)\n-            TaskPools.IO_POOL.execute { dataChannelHandler.processPacket(PacketInfo(dataChannelPacket)) }\n+            incomingDataChannelMessagesQueue.add(PacketInfo(dataChannelPacket))\n         }\n         socket.listen()\n         sctpSocket = Optional.of(socket)\n@@ -1099,7 +1113,7 @@ class Endpoint @JvmOverloads constructor(\n         private var dataChannelStack: DataChannelStack? = null\n         private val cachedDataChannelPackets = LinkedBlockingQueue<PacketInfo>()\n \n-        override fun consume(packetInfo: PacketInfo) {\n+        public override fun consume(packetInfo: PacketInfo) {\n             synchronized(dataChannelStackLock) {\n                 when (val packet = packetInfo.packet) {\n                     is DataChannelPacket -> {\n', 'addition_count': 16, 'deletion_count': 2, 'src_file': 'a/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt', 'tgt_file': 'b/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt', 'file_extension': '.kt', 'before_file': '/*\n * Copyright @ 2018 - present 8x8, Inc.\n *\n * Licensed under the Apache License, Version 2.0 (the "License");\n * you may not use this file except in compliance with the License.\n * You may obtain a copy of the License at\n *\n *     http://www.apache.org/licenses/LICENSE-2.0\n *\n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an "AS IS" BASIS,\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n * See the License for the specific language governing permissions and\n * limitations under the License.\n */\n\npackage org.jitsi.videobridge\n\nimport org.jitsi.config.JitsiConfig\nimport org.jitsi.metaconfig.config\nimport org.jitsi.nlj.Features\nimport org.jitsi.nlj.MediaSourceDesc\nimport org.jitsi.nlj.PacketHandler\nimport org.jitsi.nlj.PacketInfo\nimport org.jitsi.nlj.Transceiver\nimport org.jitsi.nlj.TransceiverEventHandler\nimport org.jitsi.nlj.format.PayloadType\nimport org.jitsi.nlj.rtp.AudioRtpPacket\nimport org.jitsi.nlj.rtp.ParsedVideoPacket\nimport org.jitsi.nlj.rtp.RtpExtension\nimport org.jitsi.nlj.rtp.SsrcAssociationType\nimport org.jitsi.nlj.rtp.VideoRtpPacket\nimport org.jitsi.nlj.srtp.TlsRole\nimport org.jitsi.nlj.stats.EndpointConnectionStats\nimport org.jitsi.nlj.stats.NodeStatsBlock\nimport org.jitsi.nlj.transform.node.ConsumerNode\nimport org.jitsi.nlj.util.Bandwidth\nimport org.jitsi.nlj.util.LocalSsrcAssociation\nimport org.jitsi.nlj.util.NEVER\nimport org.jitsi.nlj.util.PacketInfoQueue\nimport org.jitsi.nlj.util.RemoteSsrcAssociation\nimport org.jitsi.nlj.util.sumOf\nimport org.jitsi.rtp.Packet\nimport org.jitsi.rtp.UnparsedPacket\nimport org.jitsi.rtp.rtcp.RtcpSrPacket\nimport org.jitsi.rtp.rtcp.rtcpfb.RtcpFbPacket\nimport org.jitsi.rtp.rtcp.rtcpfb.payload_specific_fb.RtcpFbFirPacket\nimport org.jitsi.rtp.rtcp.rtcpfb.payload_specific_fb.RtcpFbPliPacket\nimport org.jitsi.rtp.rtp.RtpPacket\nimport org.jitsi.utils.MediaType\nimport org.jitsi.utils.concurrent.RecurringRunnableExecutor\nimport org.jitsi.utils.logging2.Logger\nimport org.jitsi.utils.logging2.cdebug\nimport org.jitsi.utils.mins\nimport org.jitsi.utils.queue.CountingErrorHandler\nimport org.jitsi.videobridge.cc.BandwidthProbing\nimport org.jitsi.videobridge.cc.allocation.BandwidthAllocation\nimport org.jitsi.videobridge.cc.allocation.BitrateController\nimport org.jitsi.videobridge.cc.allocation.EffectiveConstraintsMap\nimport org.jitsi.videobridge.cc.allocation.VideoConstraints\nimport org.jitsi.videobridge.datachannel.DataChannelStack\nimport org.jitsi.videobridge.datachannel.protocol.DataChannelPacket\nimport org.jitsi.videobridge.datachannel.protocol.DataChannelProtocolConstants\nimport org.jitsi.videobridge.message.BridgeChannelMessage\nimport org.jitsi.videobridge.message.ForwardedEndpointsMessage\nimport org.jitsi.videobridge.message.ForwardedSourcesMessage\nimport org.jitsi.videobridge.message.ReceiverVideoConstraintsMessage\nimport org.jitsi.videobridge.message.SenderSourceConstraintsMessage\nimport org.jitsi.videobridge.message.SenderVideoConstraintsMessage\nimport org.jitsi.videobridge.relay.AudioSourceDesc\nimport org.jitsi.videobridge.rest.root.debug.EndpointDebugFeatures\nimport org.jitsi.videobridge.sctp.SctpConfig\nimport org.jitsi.videobridge.sctp.SctpManager\nimport org.jitsi.videobridge.stats.PacketTransitStats\nimport org.jitsi.videobridge.transport.dtls.DtlsTransport\nimport org.jitsi.videobridge.transport.ice.IceTransport\nimport org.jitsi.videobridge.util.ByteBufferPool\nimport org.jitsi.videobridge.util.TaskPools\nimport org.jitsi.videobridge.util.looksLikeDtls\nimport org.jitsi.videobridge.websocket.colibriWebSocketServiceSupplier\nimport org.jitsi.xmpp.extensions.colibri.WebSocketPacketExtension\nimport org.jitsi.xmpp.extensions.jingle.DtlsFingerprintPacketExtension\nimport org.jitsi.xmpp.extensions.jingle.IceUdpTransportPacketExtension\nimport org.jitsi_modified.sctp4j.SctpDataCallback\nimport org.jitsi_modified.sctp4j.SctpServerSocket\nimport org.jitsi_modified.sctp4j.SctpSocket\nimport org.json.simple.JSONObject\nimport java.nio.ByteBuffer\nimport java.security.SecureRandom\nimport java.time.Clock\nimport java.time.Duration\nimport java.time.Instant\nimport java.util.Optional\nimport java.util.concurrent.LinkedBlockingQueue\nimport java.util.concurrent.TimeUnit\nimport java.util.concurrent.atomic.AtomicLong\nimport java.util.function.Supplier\n\n/**\n * Models a local endpoint (participant) in a [Conference]\n */\nclass Endpoint @JvmOverloads constructor(\n    id: String,\n    conference: Conference,\n    parentLogger: Logger,\n    /**\n     * True if the ICE agent for this Endpoint will be initialized to serve\n     * as a controlling ICE agent, false otherwise\n     */\n    iceControlling: Boolean,\n    private val isUsingSourceNames: Boolean,\n    private val doSsrcRewriting: Boolean,\n    /**\n     * Whether this endpoint is in "visitor" mode, i.e. should be invisible to other endpoints.\n     */\n    val visitor: Boolean,\n    private val clock: Clock = Clock.systemUTC()\n) : AbstractEndpoint(conference, id, parentLogger),\n    PotentialPacketHandler,\n    EncodingsManager.EncodingsUpdateListener,\n    SsrcRewriter {\n    /**\n     * The time at which this endpoint was created\n     */\n    val creationTime = clock.instant()\n\n    private val sctpHandler = SctpHandler()\n    private val dataChannelHandler = DataChannelHandler()\n\n    /* TODO: do we ever want to support useUniquePort for an Endpoint? */\n    private val iceTransport = IceTransport(id, iceControlling, false, logger)\n    private val dtlsTransport = DtlsTransport(logger)\n\n    private val diagnosticContext = conference.newDiagnosticContext().apply {\n        put("endpoint_id", id)\n    }\n\n    private val timelineLogger = logger.createChildLogger("timeline.${this.javaClass.name}")\n\n    /**\n     * The [SctpManager] instance we\'ll use to manage the SCTP connection\n     */\n    private var sctpManager: SctpManager? = null\n\n    private var dataChannelStack: DataChannelStack? = null\n\n    /**\n     * The [SctpSocket] for this endpoint, if an SCTP connection was\n     * negotiated.\n     */\n    private var sctpSocket: Optional<SctpServerSocket> = Optional.empty()\n\n    /**\n     * Whether this endpoint should accept audio packets. We set this according\n     * to whether the endpoint has an audio Colibri channel whose direction\n     * allows sending.\n     */\n    var acceptAudio = false\n\n    /**\n     * Whether this endpoint should accept video packets. We set this according\n     * to whether the endpoint has a video Colibri channel whose direction\n     * allows sending.\n     */\n    var acceptVideo = false\n\n    /**\n     * The queue we put outgoing SRTP packets onto so they can be sent\n     * out via the [IceTransport] on an IO thread.\n     */\n    private val outgoingSrtpPacketQueue = PacketInfoQueue(\n        "${javaClass.simpleName}-outgoing-packet-queue",\n        TaskPools.IO_POOL,\n        this::doSendSrtp,\n        TransportConfig.queueSize\n    ).apply {\n        setErrorHandler(queueErrorCounter)\n    }\n\n        setErrorHandler(queueErrorCounter)\n    }\n\n    private val bitrateController = BitrateController(\n        object : BitrateController.EventHandler {\n            override fun allocationChanged(allocation: BandwidthAllocation) {\n            override fun forwardedEndpointsChanged(forwardedEndpoints: Set<String>) =\n                sendForwardedEndpointsMessage(forwardedEndpoints)\n\n            override fun forwardedSourcesChanged(forwardedSources: Set<String>) {\n                sendForwardedSourcesMessage(forwardedSources)\n            }\n\n            override fun effectiveVideoConstraintsChanged(\n                oldEffectiveConstraints: EffectiveConstraintsMap,\n                newEffectiveConstraints: EffectiveConstraintsMap,\n            ) = this@Endpoint.effectiveVideoConstraintsChanged(oldEffectiveConstraints, newEffectiveConstraints)\n\n            override fun keyframeNeeded(endpointId: String?, ssrc: Long) =\n                conference.requestKeyframe(endpointId, ssrc)\n        },\n        Supplier { getOrderedEndpoints() },\n        diagnosticContext,\n        logger,\n        isUsingSourceNames,\n    )\n\n    /** Whether any sources are suspended from being sent to this endpoint because of BWE. */\n    fun hasSuspendedSources() = bitrateController.hasSuspendedSources()\n\n    /**\n     * The instance which manages the Colibri messaging (over a data channel\n     * or web sockets).\n     */\n    private val _messageTransport = EndpointMessageTransport(\n        this,\n        Supplier { conference.videobridge.statistics },\n        conference,\n        logger\n    )\n\n    override fun getMessageTransport(): EndpointMessageTransport = _messageTransport\n\n    /**\n     * Gets the endpoints in the conference in LastN order, with this {@link Endpoint} removed.\n     */\n    fun getOrderedEndpoints(): List<AbstractEndpoint> =\n        conference.orderedEndpoints.filterNot { it == this }\n\n    /**\n     * Listen for RTT updates from [transceiver] and update the ICE stats the first time an RTT is available. Note that\n     * the RTT is measured via RTCP, since we don\'t expose response time for STUN requests.\n     */\n    private val rttListener: EndpointConnectionStats.EndpointConnectionStatsListener =\n        object : EndpointConnectionStats.EndpointConnectionStatsListener {\n            override fun onRttUpdate(newRttMs: Double) {\n                if (newRttMs > 0) {\n                    transceiver.removeEndpointConnectionStatsListener(this)\n                    iceTransport.updateStatsOnInitialRtt(newRttMs)\n                }\n            }\n        }\n\n    val transceiver = Transceiver(\n        id,\n        TaskPools.CPU_POOL,\n        TaskPools.CPU_POOL,\n        TaskPools.SCHEDULED_POOL,\n        diagnosticContext,\n        logger,\n        TransceiverEventHandlerImpl(),\n        clock\n    ).apply {\n        setIncomingPacketHandler(object : ConsumerNode("receiver chain handler") {\n            override fun consume(packetInfo: PacketInfo) {\n                this@Endpoint.handleIncomingPacket(packetInfo)\n            }\n\n            override fun trace(f: () -> Unit) = f.invoke()\n        })\n        addEndpointConnectionStatsListener(rttListener)\n        setLocalSsrc(MediaType.AUDIO, conference.localAudioSsrc)\n        setLocalSsrc(MediaType.VIDEO, conference.localVideoSsrc)\n    }\n\n    private val bandwidthProbing = BandwidthProbing(\n        object : BandwidthProbing.ProbingDataSender {\n            override fun sendProbing(mediaSsrcs: Collection<Long>, numBytes: Int): Int {\n                return transceiver.sendProbing(mediaSsrcs, numBytes)\n            }\n        },\n        Supplier { bitrateController.getStatusSnapshot() }\n    ).apply {\n        diagnosticsContext = this@Endpoint.diagnosticContext\n        enabled = true\n    }.also {\n        recurringRunnableExecutor.registerRecurringRunnable(it)\n    }\n\n    /**\n     * Manages remapping of video SSRCs when enabled.\n     */\n    private val videoSsrcs = VideoSsrcCache(SsrcLimitConfig.config.maxVideoSsrcs, this, logger)\n\n    /**\n     * Manages remapping of audio SSRCs when enabled.\n     */\n    private val audioSsrcs = AudioSsrcCache(SsrcLimitConfig.config.maxAudioSsrcs, this, logger)\n\n    /**\n     * Last advertised forwarded-sources in remapping mode.\n     */\n    private var activeSources: Set<String> = emptySet()\n\n    /**\n     * Track allocated send SSRCs to avoid collisions.\n     */\n    private val sendSsrcs = mutableSetOf<Long>()\n\n    /**\n     * Next allocated send SSRC, when allocating serially (off by default).\n     */\n    private var nextSendSsrc = 777000001L\n\n    init {\n        conference.encodingsManager.subscribe(this)\n        setupIceTransport()\n        setupDtlsTransport()\n\n        conference.videobridge.statistics.totalEndpoints.inc()\n        if (visitor) {\n            conference.videobridge.statistics.totalVisitors.inc()\n        }\n\n        logger.info("Created new endpoint isUsingSourceNames=$isUsingSourceNames, iceControlling=$iceControlling")\n    }\n\n    override var mediaSources: Array<MediaSourceDesc>\n        get() = transceiver.getMediaSources()\n        set(value) {\n            applyVideoTypeCache(value)\n            val wasEmpty = transceiver.getMediaSources().isEmpty()\n            if (transceiver.setMediaSources(value)) {\n                eventEmitter.fireEvent { sourcesChanged() }\n            }\n            if (wasEmpty) {\n                sendAllVideoConstraints()\n            }\n        }\n\n    /**\n     *  Keep track of this endpoint\'s audio sources.\n     */\n    var audioSources: ArrayList<AudioSourceDesc> = ArrayList()\n\n    private fun setupIceTransport() {\n        iceTransport.incomingDataHandler = object : IceTransport.IncomingDataHandler {\n            override fun dataReceived(data: ByteArray, offset: Int, length: Int, receivedTime: Instant) {\n                // DTLS data will be handled by the DtlsTransport, but SRTP data can go\n                // straight to the transceiver\n                if (looksLikeDtls(data, offset, length)) {\n                    // DTLS transport is responsible for making its own copy, because it will manage its own\n                    // buffers\n                    dtlsTransport.dtlsDataReceived(data, offset, length)\n                } else {\n                    val copy = ByteBufferPool.getBuffer(\n                        length +\n                            RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET +\n                            Packet.BYTES_TO_LEAVE_AT_END_OF_PACKET\n                    )\n                    System.arraycopy(data, offset, copy, RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET, length)\n                    val pktInfo =\n                        PacketInfo(UnparsedPacket(copy, RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET, length)).apply {\n                            this.receivedTime = receivedTime\n                        }\n                    transceiver.handleIncomingPacket(pktInfo)\n                }\n            }\n        }\n        iceTransport.eventHandler = object : IceTransport.EventHandler {\n            override fun connected() {\n                logger.info("ICE connected")\n                eventEmitter.fireEvent { iceSucceeded() }\n                transceiver.setOutgoingPacketHandler(object : PacketHandler {\n                    override fun processPacket(packetInfo: PacketInfo) {\n                        packetInfo.addEvent(SRTP_QUEUE_ENTRY_EVENT)\n                        outgoingSrtpPacketQueue.add(packetInfo)\n                    }\n                })\n                TaskPools.IO_POOL.execute(iceTransport::startReadingData)\n                TaskPools.IO_POOL.execute(dtlsTransport::startDtlsHandshake)\n            }\n\n            override fun failed() {\n                eventEmitter.fireEvent { iceFailed() }\n            }\n\n            override fun consentUpdated(time: Instant) {\n                transceiver.packetIOActivity.lastIceActivityInstant = time\n            }\n        }\n    }\n\n    /**\n     * Whether this endpoint has any endpoints "on-stage".\n     */\n    fun isInStageView() = bitrateController.allocationSettings.onStageEndpoints.isNotEmpty()\n\n    private fun setupDtlsTransport() {\n        dtlsTransport.incomingDataHandler = object : DtlsTransport.IncomingDataHandler {\n            override fun dtlsAppDataReceived(buf: ByteArray, off: Int, len: Int) {\n                this@Endpoint.dtlsAppPacketReceived(buf, off, len)\n            }\n        }\n        dtlsTransport.outgoingDataHandler = object : DtlsTransport.OutgoingDataHandler {\n            override fun sendData(buf: ByteArray, off: Int, len: Int) {\n                iceTransport.send(buf, off, len)\n            }\n        }\n        dtlsTransport.eventHandler = object : DtlsTransport.EventHandler {\n            override fun handshakeComplete(\n                chosenSrtpProtectionProfile: Int,\n                tlsRole: TlsRole,\n                keyingMaterial: ByteArray\n            ) {\n                logger.info("DTLS handshake complete")\n                transceiver.setSrtpInformation(chosenSrtpProtectionProfile, tlsRole, keyingMaterial)\n                // TODO(brian): the old code would work even if the sctp connection was created after\n                //  the handshake had completed, but this won\'t (since this is a one-time event).  do\n                //  we need to worry about that case?\n                sctpSocket.ifPresent(::acceptSctpConnection)\n                scheduleEndpointMessageTransportTimeout()\n            }\n        }\n    }\n\n    fun updateForceMute(audioForceMuted: Boolean, videoForceMuted: Boolean) {\n        transceiver.forceMuteAudio(audioForceMuted)\n        transceiver.forceMuteVideo(videoForceMuted)\n    }\n\n    override fun addPayloadType(payloadType: PayloadType) {\n        transceiver.addPayloadType(payloadType)\n        bitrateController.addPayloadType(payloadType)\n    }\n\n    override fun addRtpExtension(rtpExtension: RtpExtension) = transceiver.addRtpExtension(rtpExtension)\n\n    fun addReceiveSsrc(ssrc: Long, mediaType: MediaType) {\n        logger.cdebug { "Adding receive ssrc $ssrc of type $mediaType" }\n        transceiver.addReceiveSsrc(ssrc, mediaType)\n        conference.addEndpointSsrc(this, ssrc)\n    }\n\n    override fun onNewSsrcAssociation(\n        endpointId: String,\n        primarySsrc: Long,\n        secondarySsrc: Long,\n        type: SsrcAssociationType\n    ) {\n        if (endpointId.equals(id, ignoreCase = true)) {\n            transceiver.addSsrcAssociation(LocalSsrcAssociation(primarySsrc, secondarySsrc, type))\n        } else {\n            transceiver.addSsrcAssociation(RemoteSsrcAssociation(primarySsrc, secondarySsrc, type))\n        }\n    }\n\n    fun setFeature(feature: EndpointDebugFeatures, enabled: Boolean) {\n        when (feature) {\n            EndpointDebugFeatures.PCAP_DUMP -> transceiver.setFeature(Features.TRANSCEIVER_PCAP_DUMP, enabled)\n        }\n    }\n\n    fun isFeatureEnabled(feature: EndpointDebugFeatures): Boolean {\n        return when (feature) {\n            EndpointDebugFeatures.PCAP_DUMP -> transceiver.isFeatureEnabled(Features.TRANSCEIVER_PCAP_DUMP)\n        }\n    }\n\n    override fun isSendingAudio(): Boolean {\n        // The endpoint is sending audio if we (the transceiver) are receiving audio.\n        return transceiver.isReceivingAudio()\n    }\n    override fun isSendingVideo(): Boolean {\n        // The endpoint is sending video if we (the transceiver) are receiving video.\n        return transceiver.isReceivingVideo()\n    }\n\n    private fun doSendSrtp(packetInfo: PacketInfo): Boolean {\n        packetInfo.addEvent(SRTP_QUEUE_EXIT_EVENT)\n        PacketTransitStats.packetSent(packetInfo)\n\n        packetInfo.sent()\n        if (timelineLogger.isTraceEnabled && logTimeline()) {\n            timelineLogger.trace { packetInfo.timeline.toString() }\n        }\n        iceTransport.send(packetInfo.packet.buffer, packetInfo.packet.offset, packetInfo.packet.length)\n        ByteBufferPool.returnBuffer(packetInfo.packet.buffer)\n        return true\n    }\n\n    /**\n     * Notifies this endpoint that the ordered list of last-n endpoints has changed\n     */\n    fun lastNEndpointsChanged() = bitrateController.endpointOrderingChanged()\n\n    /**\n     * {@inheritDoc}\n     */\n    override fun sendMessage(msg: BridgeChannelMessage) = messageTransport.sendMessage(msg)\n\n    // TODO: this should be part of an EndpointMessageTransport.EventHandler interface\n    fun endpointMessageTransportConnected() {\n        sendAllVideoConstraints()\n        videoSsrcs.sendAllMappings()\n        audioSsrcs.sendAllMappings()\n    }\n\n    private fun sendAllVideoConstraints() {\n        maxReceiverVideoConstraintsMap.forEach { (sourceName, constraints) ->\n            sendVideoConstraintsV2(sourceName, constraints)\n        }\n    }\n\n    /**\n     * Handle a DTLS app packet (that is, a packet of some other protocol sent\n     * over DTLS) which has just been received.\n     */\n    // TODO(brian): change sctp handler to take buf, off, len\n    fun dtlsAppPacketReceived(data: ByteArray, off: Int, len: Int) =\n        sctpHandler.processPacket(PacketInfo(UnparsedPacket(data, off, len)))\n\n    private fun effectiveVideoConstraintsChanged(\n        oldEffectiveConstraints: EffectiveConstraintsMap,\n        newEffectiveConstraints: EffectiveConstraintsMap\n    ) {\n        val removedSources = oldEffectiveConstraints.keys.filterNot { it in newEffectiveConstraints.keys }\n\n        // Sources that "this" endpoint no longer receives.\n        removedSources.map { it.sourceName }.forEach { removedSourceName ->\n            // Remove ourself as a receiver from that endpoint\n            conference.findSourceOwner(removedSourceName)?.removeSourceReceiver(removedSourceName, id)\n        }\n\n        // Added or updated\n        newEffectiveConstraints.forEach { (source, effectiveConstraints) ->\n            conference.findSourceOwner(source.sourceName)?.addReceiver(id, source.sourceName, effectiveConstraints)\n        }\n\n        if (doSsrcRewriting) {\n            val newActiveSources =\n                newEffectiveConstraints.entries.filter { !it.value.isDisabled() }.map { it.key }.toList()\n            val newActiveSourceNames = newActiveSources.mapNotNull { it.sourceName }.toSet()\n            /* safe unlocked access of activeSources. BitrateController will not overlap calls to this method. */\n            if (activeSources != newActiveSourceNames) {\n                activeSources = newActiveSourceNames\n                videoSsrcs.activate(newActiveSources)\n            }\n        }\n    }\n\n    override fun sendVideoConstraints(maxVideoConstraints: VideoConstraints) {\n        // Note that it\'s up to the client to respect these constraints.\n        if (mediaSources.isEmpty()) {\n            logger.cdebug { "Suppressing sending a SenderVideoConstraints message, endpoint has no streams." }\n        } else {\n            val senderVideoConstraintsMessage = SenderVideoConstraintsMessage(maxVideoConstraints.maxHeight)\n            logger.cdebug { "Sender constraints changed: ${senderVideoConstraintsMessage.toJson()}" }\n            sendMessage(senderVideoConstraintsMessage)\n        }\n    }\n\n    override fun sendVideoConstraintsV2(sourceName: String, maxVideoConstraints: VideoConstraints) {\n        // Note that it\'s up to the client to respect these constraints.\n        if (findMediaSourceDesc(sourceName) == null) {\n            logger.warn {\n                "Suppressing sending a SenderVideoConstraints message, endpoint has no such source: $sourceName"\n            }\n        } else {\n            if (isUsingSourceNames) {\n                val senderSourceConstraintsMessage =\n                    SenderSourceConstraintsMessage(sourceName, maxVideoConstraints.maxHeight)\n                logger.cdebug { "Sender constraints changed: ${senderSourceConstraintsMessage.toJson()}" }\n                sendMessage(senderSourceConstraintsMessage)\n            } else {\n            // holding a lock inside the SctpSocket which can cause a deadlock\n            // if two endpoints are trying to send datachannel messages to one\n            // another (with stats broadcasting it can happen often)\n            TaskPools.IO_POOL.execute { dataChannelHandler.processPacket(PacketInfo(dataChannelPacket)) }\n        }\n        socket.listen()\n        sctpSocket = Optional.of(socket)\n\n    /**\n     * Create an SCTP connection for this Endpoint.  If [openDataChannelLocally] is true,\n     * we will create the data channel locally, otherwise we will wait for the remote side\n     * to open it.\n     */\n    fun createSctpConnection() {\n        logger.cdebug { "Creating SCTP manager" }\n        // Create the SctpManager and provide it a method for sending SCTP data\n        sctpManager = SctpManager(\n            { data, offset, length ->\n                dtlsTransport.sendDtlsData(data, offset, length)\n                0\n            },\n            logger\n        )\n        sctpHandler.setSctpManager(sctpManager!!)\n        // NOTE(brian): as far as I know we always act as the \'server\' for sctp\n        // connections, but if not we can make which type we use dynamic\n        val socket = sctpManager!!.createServerSocket()\n        socket.eventHandler = object : SctpSocket.SctpSocketEventHandler {\n            override fun onReady() {\n                logger.info("SCTP connection is ready, creating the Data channel stack")\n                dataChannelStack = DataChannelStack(\n                    { data, sid, ppid -> socket.send(data, true, sid, ppid) },\n                    logger\n                )\n                // This handles if the remote side will be opening the data channel\n                dataChannelStack!!.onDataChannelStackEvents { dataChannel ->\n                    logger.info("Remote side opened a data channel.")\n                    _messageTransport.setDataChannel(dataChannel)\n                }\n                dataChannelHandler.setDataChannelStack(dataChannelStack!!)\n                if (openDataChannelLocally) {\n                    // This logic is for opening the data channel locally\n                    logger.info("Will open the data channel.")\n                    val dataChannel = dataChannelStack!!.createDataChannel(\n                        DataChannelProtocolConstants.RELIABLE,\n                        0,\n                        0,\n                        0,\n                        "default"\n                    )\n                    _messageTransport.setDataChannel(dataChannel)\n                    dataChannel.open()\n                } else {\n                    logger.info("Will wait for the remote side to open the data channel.")\n                }\n            }\n\n            override fun onDisconnected() {\n                logger.info("SCTP connection is disconnected")\n            }\n        }\n        socket.dataCallback = SctpDataCallback { data, sid, ssn, tsn, ppid, context, flags ->\n            // We assume all data coming over SCTP will be datachannel data\n            val dataChannelPacket = DataChannelPacket(data, 0, data.size, sid, ppid.toInt())\n            // Post the rest of the task here because the current context is\n            // holding a lock inside the SctpSocket which can cause a deadlock\n            // if two endpoints are trying to send datachannel messages to one\n            // another (with stats broadcasting it can happen often)\n            incomingDataChannelMessagesQueue.add(PacketInfo(dataChannelPacket))\n        }\n        socket.listen()\n        sctpSocket = Optional.of(socket)\n    }\n\n    fun acceptSctpConnection(sctpServerSocket: SctpServerSocket) {\n        TaskPools.IO_POOL.execute {\n            // We don\'t want to block the thread calling\n            // onDtlsHandshakeComplete so run the socket acceptance in an IO\n            // pool thread\n            // FIXME: This runs forever once the socket is closed (\n            // accept never returns true).\n            logger.info("Attempting to establish SCTP socket connection")\n            var attempts = 0\n            while (!sctpServerSocket.accept()) {\n                attempts++\n                try {\n                    Thread.sleep(100)\n                } catch (e: InterruptedException) {\n                    break\n                }\n                if (attempts > 100) {\n                    logger.error("Timed out waiting for SCTP connection from remote side")\n                    break\n                }\n            }\n            logger.cdebug { "SCTP socket ${sctpServerSocket.hashCode()} accepted connection" }\n        }\n    }\n\n    /**\n     * Schedule a timeout to fire log a message and track a stat if we don\'t\n     * have an endpoint message transport connected within the timeout.\n     */\n    fun scheduleEndpointMessageTransportTimeout() {\n        TaskPools.SCHEDULED_POOL.schedule(\n            {\n                if (!isExpired) {\n                    if (!messageTransport.isConnected) {\n                        logger.error("EndpointMessageTransport still not connected.")\n                        conference.videobridge.statistics.numEndpointsNoMessageTransportAfterDelay.inc()\n                    }\n                }\n            },\n            30,\n            TimeUnit.SECONDS\n        )\n    }\n\n    /**\n     * Checks whether a WebSocket connection with a specific password string\n     * should be accepted for this endpoint.\n     * @param password the\n     * @return {@code true} iff the password matches.\n     */\n    fun acceptWebSocket(password: String): Boolean {\n        if (iceTransport.icePassword != password) {\n            logger.warn(\n                "Incoming web socket request with an invalid password. " +\n                    "Expected: ${iceTransport.icePassword} received $password"\n            )\n            return false\n        }\n        return true\n    }\n\n    /**\n     * Sends a message to this endpoint in order to notify it that the set of endpoints for which the bridge\n     * is sending video has changed.\n     *\n     * @param forwardedEndpoints the collection of forwarded endpoints.\n     */\n    @Deprecated("", ReplaceWith("sendForwardedSourcesMessage"), DeprecationLevel.WARNING)\n    fun sendForwardedEndpointsMessage(forwardedEndpoints: Collection<String>) {\n        if (isUsingSourceNames) {\n            return\n        }\n\n        val msg = ForwardedEndpointsMessage(forwardedEndpoints)\n        TaskPools.IO_POOL.execute {\n            try {\n                sendMessage(msg)\n            } catch (t: Throwable) {\n                logger.warn("Failed to send message:", t)\n            }\n        }\n    }\n\n    /**\n     * Sends a message to this endpoint in order to notify it that the set of media sources for which the bridge\n     * is sending video has changed.\n     *\n     * @param forwardedSources the collection of forwarded media sources (by name).\n     */\n    fun sendForwardedSourcesMessage(forwardedSources: Collection<String>) {\n        if (!isUsingSourceNames) {\n            return\n        }\n\n        val msg = ForwardedSourcesMessage(forwardedSources)\n        TaskPools.IO_POOL.execute {\n            try {\n                sendMessage(msg)\n            } catch (t: Throwable) {\n                logger.warn("Failed to send message:", t)\n            }\n        }\n    }\n\n    /**\n     * Sets the remote transport information (ICE candidates, DTLS fingerprints).\n     *\n     * @param transportInfo the XML extension which contains the remote\n     * transport information.\n     */\n    fun setTransportInfo(transportInfo: IceUdpTransportPacketExtension) {\n        val remoteFingerprints = mutableMapOf<String, String>()\n        val fingerprintExtensions = transportInfo.getChildExtensionsOfType(DtlsFingerprintPacketExtension::class.java)\n        fingerprintExtensions.forEach { fingerprintExtension ->\n            if (fingerprintExtension.hash != null && fingerprintExtension.fingerprint != null) {\n                remoteFingerprints[fingerprintExtension.hash] = fingerprintExtension.fingerprint\n            } else {\n                logger.info("Ignoring empty DtlsFingerprint extension: ${transportInfo.toXML()}")\n            }\n        }\n        dtlsTransport.setRemoteFingerprints(remoteFingerprints)\n        if (fingerprintExtensions.isNotEmpty()) {\n            val setup = fingerprintExtensions.first().setup\n            dtlsTransport.setSetupAttribute(setup)\n        }\n        iceTransport.startConnectivityEstablishment(transportInfo)\n    }\n\n    fun describeTransport(): IceUdpTransportPacketExtension {\n        val iceUdpTransportPacketExtension = IceUdpTransportPacketExtension()\n        iceTransport.describe(iceUdpTransportPacketExtension)\n        dtlsTransport.describe(iceUdpTransportPacketExtension)\n        colibriWebSocketServiceSupplier.get()?.let { colibriWebsocketService ->\n            colibriWebsocketService.getColibriWebSocketUrl(\n                conference.id,\n                id,\n                iceTransport.icePassword\n            )?.let { wsUrl ->\n                val wsPacketExtension = WebSocketPacketExtension(wsUrl)\n                iceUdpTransportPacketExtension.addChildExtension(wsPacketExtension)\n            }\n        }\n\n        logger.cdebug { "Transport description:\\n${iceUdpTransportPacketExtension.toXML()}" }\n\n        return iceUdpTransportPacketExtension\n    }\n\n    /**\n     * Handle incoming RTP packets which have been fully processed by the\n     * transceiver\'s incoming pipeline.\n     */\n    fun handleIncomingPacket(packetInfo: PacketInfo) {\n        if (visitor) {\n            /* Never forward RTP/RTCP from a visitor. */\n            ByteBufferPool.returnBuffer(packetInfo.packet.buffer)\n            return\n        }\n\n        packetInfo.endpointId = id\n        conference.handleIncomingPacket(packetInfo)\n    }\n\n    override fun receivesSsrc(ssrc: Long): Boolean = transceiver.receivesSsrc(ssrc)\n\n    fun doesSsrcRewriting(): Boolean = doSsrcRewriting\n\n    fun unmapRtcpFbSsrc(packet: RtcpFbPacket) = videoSsrcs.unmapRtcpFbSsrc(packet)\n\n    override fun getSsrcs() = HashSet(transceiver.receiveSsrcs)\n\n    override fun getLastIncomingActivity(): Instant = transceiver.packetIOActivity.lastIncomingActivityInstant\n\n    override fun requestKeyframe() = transceiver.requestKeyFrame()\n\n    override fun requestKeyframe(mediaSsrc: Long) = transceiver.requestKeyFrame(mediaSsrc)\n\n    /** Whether we are currently oversending to this endpoint. */\n    fun isOversending(): Boolean = bitrateController.isOversending()\n\n    /**\n     * Returns how many endpoints this Endpoint is currently forwarding video for\n     */\n    fun numForwardedEndpoints(): Int = bitrateController.numForwardedEndpoints()\n\n    fun setBandwidthAllocationSettings(message: ReceiverVideoConstraintsMessage) {\n        bitrateController.setBandwidthAllocationSettings(message)\n    }\n\n    /**\n     * {@inheritDoc}\n     */\n    override fun findVideoSourceProps(ssrc: Long): MediaSourceDesc? {\n        conference.getEndpointBySsrc(ssrc)?.let { ep ->\n            ep.mediaSources.forEach { s ->\n                if (s.findRtpEncodingDesc(ssrc) != null) {\n                    return s\n                }\n            }\n        }\n        logger.error { "No properties found for SSRC $ssrc." }\n        return null\n    }\n\n    /**\n     * {@inheritDoc}\n     */\n    override fun findAudioSourceProps(ssrc: Long): AudioSourceDesc? {\n        conference.getEndpointBySsrc(ssrc)?.let { ep ->\n            if (ep !is Endpoint)\n                return null\n            return ep.audioSources.find { s -> s.ssrc == ssrc }\n        }\n        logger.error { "No properties found for SSRC $ssrc." }\n        return null\n    }\n\n    /**\n     * {@inheritDoc}\n     */\n    override fun getNextSendSsrc(): Long {\n        synchronized(sendSsrcs) {\n            while (true) {\n                val ssrc = if (useRandomSendSsrcs)\n                    random.nextLong().and(0xFFFF_FFFFL)\n                else\n                    nextSendSsrc++\n                if (sendSsrcs.add(ssrc))\n                    return ssrc\n            }\n        }\n    }\n\n    override fun send(packetInfo: PacketInfo) {\n        when (val packet = packetInfo.packet) {\n            is VideoRtpPacket -> {\n                if (bitrateController.transformRtp(packetInfo)) {\n                    // The original packet was transformed in place.\n                    if (doSsrcRewriting) {\n                        val start = packet !is ParsedVideoPacket || (packet.isKeyframe && packet.isStartOfFrame)\n                        if (!videoSsrcs.rewriteRtp(packet, start)) {\n                            return\n                        }\n                    }\n                    transceiver.sendPacket(packetInfo)\n                } else {\n                    logger.warn("Dropping a packet which was supposed to be accepted:$packet")\n                }\n                return\n            }\n            is AudioRtpPacket -> if (doSsrcRewriting) audioSsrcs.rewriteRtp(packet)\n            is RtcpSrPacket -> {\n                // Allow the BC to update the timestamp (in place).\n                bitrateController.transformRtcp(packet)\n                if (doSsrcRewriting) {\n                    // Just check both tables instead of looking up the type first.\n                    if (!videoSsrcs.rewriteRtcp(packet) && !audioSsrcs.rewriteRtcp(packet))\n                        return\n                }\n                logger.trace {\n                    "relaying an sr from ssrc=${packet.senderSsrc}, timestamp=${packet.senderInfo.rtpTimestamp}"\n                }\n            }\n        }\n        transceiver.sendPacket(packetInfo)\n    }\n\n    /**\n     * To find out whether the endpoint should be expired, we check the activity timestamps from the transceiver.\n     */\n    override fun shouldExpire(): Boolean {\n        if (iceTransport.hasFailed()) {\n            logger.warn("Allowing to expire because ICE failed.")\n            return true\n        }\n\n        val lastActivity = lastIncomingActivity\n        val now = clock.instant()\n\n        if (lastActivity == NEVER) {\n            val timeSinceCreation = Duration.between(creationTime, now)\n            if (timeSinceCreation > epTimeout) {\n                logger.info(\n                    "Endpoint\'s ICE connection has neither failed nor connected " +\n                        "after $timeSinceCreation expiring"\n                )\n                return true\n            }\n            // We haven\'t seen any activity yet. If this continues ICE will\n            // eventually fail (which is handled above).\n            return false\n        }\n\n        val expireTimeout = VideobridgeExpireThreadConfig.config.inactivityTimeout\n        if (Duration.between(lastActivity, now) > expireTimeout) {\n            logger.info("Allowing to expire because of no activity in over $expireTimeout")\n            return true\n        }\n        return false\n    }\n\n    fun setLastN(lastN: Int) {\n        bitrateController.lastN = lastN\n    }\n\n    /**\n     * Returns true if this endpoint\'s transport is \'fully\' connected (both ICE and DTLS), false otherwise\n     */\n    private fun isTransportConnected(): Boolean = iceTransport.isConnected() && dtlsTransport.isConnected\n\n    fun getRtt(): Double = transceiver.getTransceiverStats().endpointConnectionStats.rtt\n\n    override fun wants(packetInfo: PacketInfo): Boolean {\n        if (!isTransportConnected()) {\n            return false\n        }\n\n        return when (val packet = packetInfo.packet) {\n            is VideoRtpPacket -> acceptVideo && bitrateController.accept(packetInfo)\n            is AudioRtpPacket -> acceptAudio\n            is RtcpSrPacket -> {\n                // TODO: For SRs we\'re only interested in the ntp/rtp timestamp\n                //  association, so we could only accept srs from the main ssrc\n                bitrateController.accept(packet)\n            }\n            is RtcpFbPliPacket, is RtcpFbFirPacket -> {\n                // We assume that we are only given PLIs/FIRs destined for this\n                // endpoint. This is because Conference has to find the target\n                // endpoint (this endpoint) anyway, and we would essentially be\n                // performing the same check twice.\n                true\n            }\n            else -> {\n                logger.warn("Ignoring an unknown packet type:" + packet.javaClass.simpleName)\n                false\n            }\n        }\n    }\n\n    /**\n     * Determine whether to forward endpoint stats from another endpoint to this one.\n     */\n    fun wantsStatsFrom(ep: AbstractEndpoint): Boolean {\n        // Always forward stats in small conferences\n        if (conference.endpointCount <= statsFilterThreshold) {\n            return true\n        }\n        logger.debug {\n            buildString {\n                append("wantsStatsFrom(${ep.id}): isRecentSpeaker=${conference.speechActivity.isRecentSpeaker(ep)} ")\n                append("isRankedSpeaker=${conference.isRankedSpeaker(ep)} ")\n                if (ep.mediaSources.isEmpty()) {\n                    append("(no media sources)")\n                }\n                ep.mediaSources.forEach { source ->\n                    val name = source.sourceName\n                    append("isOnStageOrSelected($name)=${bitrateController.isOnStageOrSelected(source)} ")\n                    append("hasNonZeroEffectiveConstraints($name)=")\n                    append("${bitrateController.hasNonZeroEffectiveConstraints(source)} ")\n                }\n            }\n        }\n\n        if (conference.speechActivity.isRecentSpeaker(ep) || conference.isRankedSpeaker(ep)) {\n            return true\n        }\n        return ep.mediaSources.any { source ->\n            bitrateController.isOnStageOrSelected(source) ||\n                bitrateController.hasNonZeroEffectiveConstraints(source)\n        }\n    }\n\n    /**\n     * Updates the conference statistics with value from this endpoint. Since\n     * the values are cumulative this should execute only once when the endpoint\n     * expires.\n     */\n    private fun updateStatsOnExpire() {\n        val conferenceStats = conference.statistics\n        val transceiverStats = transceiver.getTransceiverStats()\n\n        conferenceStats.apply {\n            val incomingStats = transceiverStats.rtpReceiverStats.packetStreamStats\n            val outgoingStats = transceiverStats.outgoingPacketStreamStats\n            totalBytesReceived.addAndGet(incomingStats.bytes)\n            totalPacketsReceived.addAndGet(incomingStats.packets)\n            totalBytesSent.addAndGet(outgoingStats.bytes)\n            totalPacketsSent.addAndGet(outgoingStats.packets)\n        }\n\n        conference.videobridge.statistics.apply {\n            val bweStats = transceiverStats.bandwidthEstimatorStats\n            bweStats.getNumber("incomingEstimateExpirations")?.toLong()?.let {\n                incomingBitrateExpirations.addAndGet(it)\n            }\n            keyframesReceived.addAndGet(transceiverStats.rtpReceiverStats.videoParserStats.numKeyframes.toLong())\n            layeringChangesReceived.addAndGet(\n                transceiverStats.rtpReceiverStats.videoParserStats.numLayeringChanges.toLong()\n            )\n\n            val durationActiveVideo = transceiverStats.rtpReceiverStats.incomingStats.ssrcStats.values.filter {\n                it.mediaType == MediaType.VIDEO\n            }.sumOf { it.durationActive }\n            totalVideoStreamMillisecondsReceived.addAndGet(durationActiveVideo.toMillis())\n        }\n\n        run {\n            val bweStats = transceiverStats.bandwidthEstimatorStats\n            val lossLimitedMs = bweStats.getNumber("lossLimitedMs")?.toLong() ?: return@run\n            val lossDegradedMs = bweStats.getNumber("lossDegradedMs")?.toLong() ?: return@run\n            val lossFreeMs = bweStats.getNumber("lossFreeMs")?.toLong() ?: return@run\n\n            val participantMs = lossFreeMs + lossDegradedMs + lossLimitedMs\n            conference.videobridge.statistics.apply {\n                totalLossControlledParticipantMs.addAndGet(participantMs)\n                totalLossLimitedParticipantMs.addAndGet(lossLimitedMs)\n                totalLossDegradedParticipantMs.addAndGet(lossDegradedMs)\n            }\n        }\n\n        if (iceTransport.isConnected() && !dtlsTransport.isConnected) {\n            logger.info("Expiring an endpoint with ICE connected, but not DTLS.")\n            conferenceStats.dtlsFailedEndpoints.incrementAndGet()\n        }\n    }\n\n    override fun getDebugState(): JSONObject {\n        return super.getDebugState().apply {\n            put("bitrateController", bitrateController.debugState)\n            put("bandwidthProbing", bandwidthProbing.getDebugState())\n            put("iceTransport", iceTransport.getDebugState())\n            put("dtlsTransport", dtlsTransport.getDebugState())\n            put("transceiver", transceiver.getNodeStats().toJson())\n            put("acceptAudio", acceptAudio)\n            put("acceptVideo", acceptVideo)\n            put("visitor", visitor)\n            put("messageTransport", messageTransport.debugState)\n            if (doSsrcRewriting) {\n                put("audioSsrcs", audioSsrcs.getDebugState())\n                put("videoSsrcs", videoSsrcs.getDebugState())\n            }\n        }\n    }\n\n    override fun expire() {\n        if (super.isExpired()) {\n            return\n        }\n        super.expire()\n\n        try {\n            bitrateController.expire()\n            updateStatsOnExpire()\n            transceiver.stop()\n            logger.cdebug { transceiver.getNodeStats().prettyPrint(0) }\n            logger.cdebug { bitrateController.debugState.toJSONString() }\n            logger.cdebug { iceTransport.getDebugState().toJSONString() }\n            logger.cdebug { dtlsTransport.getDebugState().toJSONString() }\n\n            logger.info("Spent ${bitrateController.getTotalOversendingTime().seconds} seconds oversending")\n        private var dataChannelStack: DataChannelStack? = null\n        private val cachedDataChannelPackets = LinkedBlockingQueue<PacketInfo>()\n\n        override fun consume(packetInfo: PacketInfo) {\n            synchronized(dataChannelStackLock) {\n                when (val packet = packetInfo.packet) {\n                    is DataChannelPacket -> {\n        }\n\n        bandwidthProbing.enabled = false\n        recurringRunnableExecutor.deRegisterRecurringRunnable(bandwidthProbing)\n        conference.encodingsManager.unsubscribe(this)\n\n        dtlsTransport.stop()\n        iceTransport.stop()\n        outgoingSrtpPacketQueue.close()\n\n        logger.info("Expired.")\n    }\n\n    companion object {\n        /**\n         * Whether or not the bridge should be the peer which opens the data channel\n         * (as opposed to letting the far peer/client open it).\n         */\n        private const val openDataChannelLocally = false\n\n        /**\n         * Count the number of dropped packets and exceptions.\n         */\n        @JvmField\n        val queueErrorCounter = CountingErrorHandler()\n\n        /**\n         * The executor which runs bandwidth probing.\n         *\n         * TODO (brian): align the recurringRunnable stuff with whatever we end up\n         * doing with all the other executors.\n         */\n        private val recurringRunnableExecutor = RecurringRunnableExecutor(Endpoint::class.java.simpleName)\n\n        /**\n         * How long we\'ll give an endpoint to either successfully establish\n         * an ICE connection or fail before we expire it.\n         */\n        // TODO: make this configurable\n        private val epTimeout = 2.mins\n\n        private val timelineCounter = AtomicLong()\n        private val TIMELINE_FRACTION = 10000L\n\n        fun logTimeline() = timelineCounter.getAndIncrement() % TIMELINE_FRACTION == 0L\n\n        private const val SRTP_QUEUE_ENTRY_EVENT = "Entered Endpoint SRTP sender outgoing queue"\n        private const val SRTP_QUEUE_EXIT_EVENT = "Exited Endpoint SRTP sender outgoing queue"\n\n        private val statsFilterThreshold: Int by config {\n            "videobridge.stats-filter-threshold".from(JitsiConfig.newConfig)\n        }\n\n        /**\n         * This can be switched off to ease debugging.\n         */\n        private val useRandomSendSsrcs = true\n\n        /**\n         * Used for generating send SSRCs.\n         */\n        private val random = SecureRandom()\n    }\n\n    private inner class TransceiverEventHandlerImpl : TransceiverEventHandler {\n        /**\n         * Forward audio level events from the Transceiver to the conference. We use the same thread, because this fires\n         * for every packet and we want to avoid the switch. The conference audio level code must not block.\n         */\n        override fun audioLevelReceived(sourceSsrc: Long, level: Long): Boolean =\n            conference.levelChanged(this@Endpoint, level)\n\n        /**\n         * Forward bwe events from the Transceiver.\n         */\n        override fun bandwidthEstimationChanged(newValue: Bandwidth) {\n            logger.cdebug { "Estimated bandwidth is now $newValue" }\n            bitrateController.bandwidthChanged(newValue.bps.toLong())\n            bandwidthProbing.bandwidthEstimationChanged(newValue)\n        }\n    }\n\n    /**\n     * A node which can be placed in the pipeline to cache Data channel packets\n     * until the DataChannelStack is ready to handle them.\n     */\n    private class DataChannelHandler : ConsumerNode("Data channel handler") {\n        private val dataChannelStackLock = Any()\n        private var dataChannelStack: DataChannelStack? = null\n        private val cachedDataChannelPackets = LinkedBlockingQueue<PacketInfo>()\n\n        public override fun consume(packetInfo: PacketInfo) {\n            synchronized(dataChannelStackLock) {\n                when (val packet = packetInfo.packet) {\n                    is DataChannelPacket -> {\n                        dataChannelStack?.onIncomingDataChannelPacket(\n                            ByteBuffer.wrap(packet.buffer), packet.sid, packet.ppid\n                        ) ?: run {\n                            cachedDataChannelPackets.add(packetInfo)\n                        }\n                    }\n                    else -> Unit\n                }\n            }\n        }\n\n        fun setDataChannelStack(dataChannelStack: DataChannelStack) {\n            // Submit this to the pool since we wait on the lock and process any\n            // cached packets here as well\n\n            // Submit this to the pool since we wait on the lock and process any\n            // cached packets here as well\n            TaskPools.IO_POOL.execute {\n                // We grab the lock here so that we can set the SCTP manager and\n                // process any previously-cached packets as an atomic operation.\n                // It also prevents another thread from coming in via\n                // #doProcessPackets and processing packets at the same time in\n                // another thread, which would be a problem.\n                synchronized(dataChannelStackLock) {\n                    this.dataChannelStack = dataChannelStack\n                    cachedDataChannelPackets.forEach {\n                        val dcp = it.packet as DataChannelPacket\n                        dataChannelStack.onIncomingDataChannelPacket(\n                            ByteBuffer.wrap(dcp.buffer), dcp.sid, dcp.ppid\n                        )\n                    }\n                }\n            }\n        }\n\n        override fun trace(f: () -> Unit) = f.invoke()\n    }\n\n    /**\n     * A node which can be placed in the pipeline to cache SCTP packets until\n     * the SCTPManager is ready to handle them.\n     */\n    private class SctpHandler : ConsumerNode("SCTP handler") {\n        private val sctpManagerLock = Any()\n        private var sctpManager: SctpManager? = null\n        private val numCachedSctpPackets = AtomicLong(0)\n        private val cachedSctpPackets = LinkedBlockingQueue<PacketInfo>(100)\n\n        override fun consume(packetInfo: PacketInfo) {\n            synchronized(sctpManagerLock) {\n                if (SctpConfig.config.enabled) {\n                    sctpManager?.handleIncomingSctp(packetInfo) ?: run {\n                        numCachedSctpPackets.incrementAndGet()\n                        cachedSctpPackets.add(packetInfo)\n                    }\n                }\n            }\n        }\n\n        override fun getNodeStats(): NodeStatsBlock = super.getNodeStats().apply {\n            addNumber("num_cached_packets", numCachedSctpPackets.get())\n        }\n\n        fun setSctpManager(sctpManager: SctpManager) {\n            // Submit this to the pool since we wait on the lock and process any\n            // cached packets here as well\n            TaskPools.IO_POOL.execute {\n                // We grab the lock here so that we can set the SCTP manager and\n                // process any previously-cached packets as an atomic operation.\n                // It also prevents another thread from coming in via\n                // #doProcessPackets and processing packets at the same time in\n                // another thread, which would be a problem.\n                synchronized(sctpManagerLock) {\n                    this.sctpManager = sctpManager\n                    cachedSctpPackets.forEach { sctpManager.handleIncomingSctp(it) }\n                    cachedSctpPackets.clear()\n                }\n            }\n        }\n\n        override fun trace(f: () -> Unit) = f.invoke()\n    }\n}\n', 'message': 'Process data channel messages sequentially (#1665)\n\nThis prevents SCTP DataChannel messages from being processed out of order, causing warnings on channel initialization and potential issues with Last-N selection.', 'author_name': 'GitHub', 'lang': 'kt'}