<commit_before>/*
 * Copyright @ 2018 Atlassian Pty Ltd
 *
 * 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 kotlin.*;
import net.java.sip.communicator.impl.protocol.jabber.extensions.jingle.*;
import net.java.sip.communicator.impl.protocol.jabber.extensions.jingle.CandidateType;
import org.bouncycastle.crypto.tls.*;
import org.ice4j.*;
import org.ice4j.ice.*;
import org.ice4j.socket.*;
import org.jetbrains.annotations.*;
import org.jitsi.nlj.*;
import org.jitsi.nlj.dtls.*;
import org.jitsi.nlj.transform.*;
import org.jitsi.nlj.transform.node.*;
import org.jitsi.nlj.transform.node.incoming.*;
import org.jitsi.nlj.transform.node.outgoing.*;
import org.jitsi.nlj.util.*;
import org.jitsi.rtp.*;
import org.jitsi.service.neomedia.*;
import org.jitsi.util.*;
import org.jitsi.videobridge.transport.*;

import java.beans.*;
import java.io.*;
import java.net.*;
import java.nio.*;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author bbaldino
 */
public class IceDtlsTransportManager
    extends IceUdpTransportManager
{
    private static final Logger logger
            = Logger.getLogger(IceDtlsTransportManager.class);
    private final String ICE_STREAM_NAME;
    //TODO: we use this for a few different things (dtls connect, socket read, socket write).  do we need it?
    private ExecutorService executor;
    private DtlsClientStack dtlsStack = new DtlsClientStack();
    private DtlsReceiver dtlsReceiver = new DtlsReceiver(dtlsStack);
    private DtlsSender dtlsSender = new DtlsSender(dtlsStack);
    //TODO(brian): these 2 subscriber lists should be combined into a sort of 'transportmanagereventhandler' interface
    // but, what about thingds like dtlsConnected which only appliesd to IceDtlsTransportManager?
    private List<Runnable> transportConnectedSubscribers = new ArrayList<>();
    private List<Runnable> dtlsConnectedSubscribers = new ArrayList<>();
    //TODO: temp store dtls transport because newsctpconnection grabs it
    DTLSTransport dtlsTransport;
    LinkedBlockingQueue<PacketInfo> sctpAppPackets = new LinkedBlockingQueue<>();
    private Transceiver transceiver = null;
    class SocketSenderNode extends Node {
        public MultiplexingDatagramSocket socket = null;
        SocketSenderNode() {
            super("Socket sender");
        }
        @Override
        protected void doProcessPackets(@NotNull List<PacketInfo> pkts)
        {
            if (socket != null) {
                pkts.forEach(pktInfo -> {
                    try
                    {
                        socket.send(new DatagramPacket(pktInfo.getPacket().getBuffer().array(), 0, pktInfo.getPacket().getBuffer().limit()));
                    } catch (IOException e)
                    {
                        System.out.println("BRIAN: error sending outgoing dtls packet: " + e.toString());
                        throw new RuntimeException(e);
                    }
                });
            }

        }
    }
    private SocketSenderNode packetSender = new SocketSenderNode();

    private Node incomingPipelineRoot = createIncomingPipeline();
    private Node outgoingDtlsPipelineRoot = createOutgoingDtlsPipeline();
    private String id;
    protected boolean dtlsHandshakeComplete = false;

    public IceDtlsTransportManager(String id, Conference conference)
            throws IOException
    {
        super(conference, true, 1, "ice-stream-" + id, null);
        this.id = id;
        this.ICE_STREAM_NAME = "ice-stream-" + id;
        executor = Executors.newCachedThreadPool(new NameableThreadFactory("Transport manager threadpool-" + id));
        iceAgent.addStateChangeListener(this::iceAgentStateChange);
        logger.info("BRIAN: finished IceDtlsTransportManager ctor");
    }

    //TODO: need to take another look and make sure we're properly replicating all the behavior of this
    // method in IceUdpTransportManager
    @Override
    public void startConnectivityEstablishment(IceUdpTransportPacketExtension transport)
    {
        if (iceAgent.getState().isEstablished()) {
            logger.info(id + " with local ufrag " + iceAgent.getLocalUfrag() +
                    " is already established, not restarting");
            return;
        }
        logger.info(id + " with local ufrag " + iceAgent.getLocalUfrag() +
                " starting connectivity establishment with extension: " + transport.toXML());
        // Get the remote fingerprints and set them in the DTLS stack so we
        // have them to do the DTLS handshake later
        List<DtlsFingerprintPacketExtension> dfpes
                = transport.getChildExtensionsOfType(
                DtlsFingerprintPacketExtension.class);

        Map<String, String> remoteFingerprints = new HashMap<>();
        dfpes.forEach(dfpe -> {
            if (dfpe.getHash() != null && dfpe.getFingerprint() != null) {
                logger.info("Adding fingerprint " + dfpe.getHash() + " -> " + dfpe.getFingerprint());
                remoteFingerprints.put(dfpe.getHash(), dfpe.getFingerprint());
            } else {
                logger.info("Ignoring empty transport extension");
            }
        });
        if (remoteFingerprints.isEmpty()) {
            logger.info(id + " with local ufrag " + iceAgent.getLocalUfrag() + " empty transport extension, not starting connectivity yet");
            return;
        }

        // Set the remote ufrag/password
        if (transport.getUfrag() != null) {
            iceAgent.getStream(ICE_STREAM_NAME).setRemoteUfrag(transport.getUfrag());
        }
        if (transport.getPassword() != null) {
            iceAgent.getStream(ICE_STREAM_NAME).setRemotePassword(transport.getPassword());
        }

        // If ICE is running already, we try to update the checklists with the
        // candidates. Note that this is a best effort.
        boolean iceAgentStateIsRunning
                = IceProcessingState.RUNNING.equals(iceAgent.getState());

        List<CandidatePacketExtension> candidates
                = transport.getChildExtensionsOfType(
                CandidatePacketExtension.class);
        if (iceAgentStateIsRunning && candidates.isEmpty()) {
            logger.info("ICE agent is already running and this extension contained no new candidates, returning");
            return;
        }

        int remoteCandidateCount = addRemoteCandidates(candidates, iceAgentStateIsRunning);

        if (iceAgentStateIsRunning) {
            if (remoteCandidateCount == 0) {
                // XXX Effectively, the check above but realizing that all
                // candidates were ignored:
                // iceAgentStateIsRunning && candidates.isEmpty().
            } else {
                // update all components of all streams
                iceAgent.getStreams()
                        .forEach(stream -> stream.getComponents()
                                .forEach(Component::updateRemoteCandidates));
            }
        }
        else if (remoteCandidateCount != 0)
        {
            // Once again, because the ICE Agent does not support adding
            // candidates after the connectivity establishment has been started
            // and because multiple transport-info JingleIQs may be used to send
            // the whole set of transport candidates from the remote peer to the
            // local peer, do not really start the connectivity establishment
            // until we have at least one remote candidate per ICE Component.
            if (iceAgent.getStreams().stream().allMatch(
                    stream -> stream.getComponents().stream().allMatch(
                            component -> component.getRemoteCandidateCount() >= 1)))
            {
                logger.info(
                        "We have remote candidates for all ICE components. "
                                + "Starting the ICE agent.");
                iceAgent.startConnectivityEstablishment();
            }
        }
        else if (iceAgent.getStream(ICE_STREAM_NAME).getRemoteUfrag() != null
                && iceAgent.getStream(ICE_STREAM_NAME).getRemotePassword() != null)
        {
            // We don't have any remote candidates, but we already know the
            // remote ufrag and password, so we can start ICE.
            logger.info("Starting ICE agent without remote candidates.");
            iceAgent.startConnectivityEstablishment();
        }
    }

    private Transceiver getTransceiver() {
        return this.transceiver;
    }

    // Almost the same as the one in IceUdpTransportManager, but we don't use the iceStream member and
    // always assume rtcpmux
    private int addRemoteCandidates(
            List<CandidatePacketExtension> candidates,
            boolean iceAgentStateIsRunning)
    {
        // Sort the remote candidates (host < reflexive < relayed) in order to
        // create first the host, then the reflexive, the relayed candidates and
        // thus be able to set the relative-candidate matching the
        // rel-addr/rel-port attribute.
        Collections.sort(candidates);

        int generation = iceAgent.getGeneration();<commit_msg>rename some variables/misc cleanup<commit_after>/*
 * Copyright @ 2018 Atlassian Pty Ltd
 *
 * 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 kotlin.*;
import net.java.sip.communicator.impl.protocol.jabber.extensions.jingle.*;
import net.java.sip.communicator.impl.protocol.jabber.extensions.jingle.CandidateType;
import org.bouncycastle.crypto.tls.*;
import org.ice4j.*;
import org.ice4j.ice.*;
import org.ice4j.socket.*;
import org.jetbrains.annotations.*;
import org.jitsi.nlj.*;
import org.jitsi.nlj.dtls.*;
import org.jitsi.nlj.transform.*;
import org.jitsi.nlj.transform.node.*;
import org.jitsi.nlj.transform.node.incoming.*;
import org.jitsi.nlj.transform.node.outgoing.*;
import org.jitsi.nlj.util.*;
import org.jitsi.rtp.*;
import org.jitsi.service.neomedia.*;
import org.jitsi.util.*;
import org.jitsi.videobridge.transport.*;

import java.beans.*;
import java.io.*;
import java.net.*;
import java.nio.*;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author bbaldino
 */
public class IceDtlsTransportManager
    extends IceUdpTransportManager
{
    private static final Logger logger
            = Logger.getLogger(IceDtlsTransportManager.class);
    private final String ICE_STREAM_NAME;
    //TODO: we use this for a few different things (dtls connect, socket read, socket write).  do we need it?
    private ExecutorService executor;
    private DtlsClientStack dtlsStack = new DtlsClientStack();
    private DtlsReceiver dtlsReceiver = new DtlsReceiver(dtlsStack);
    private DtlsSender dtlsSender = new DtlsSender(dtlsStack);
    //TODO(brian): these 2 subscriber lists should be combined into a sort of 'transportmanagereventhandler' interface
    // but, what about thingds like dtlsConnected which only appliesd to IceDtlsTransportManager?
    private List<Runnable> transportConnectedSubscribers = new ArrayList<>();
    private List<Runnable> dtlsConnectedSubscribers = new ArrayList<>();
    //TODO: temp store dtls transport because newsctpconnection grabs it
    DTLSTransport dtlsTransport;
    LinkedBlockingQueue<PacketInfo> sctpAppPackets = new LinkedBlockingQueue<>();
    private Transceiver transceiver = null;
    class SocketSenderNode extends Node {
        public MultiplexingDatagramSocket socket = null;
        SocketSenderNode() {
            super("Socket sender");
        }
        @Override
        protected void doProcessPackets(@NotNull List<PacketInfo> pkts)
        {
            if (socket != null) {
                pkts.forEach(pktInfo -> {
                    try
                    {
                        socket.send(new DatagramPacket(pktInfo.getPacket().getBuffer().array(), 0, pktInfo.getPacket().getBuffer().limit()));
                    } catch (IOException e)
                    {
                        System.out.println("BRIAN: error sending outgoing dtls packet: " + e.toString());
                        throw new RuntimeException(e);
                    }
                });
            }

        }
    }
    private SocketSenderNode packetSender = new SocketSenderNode();

    private Node incomingPipelineRoot = createIncomingPipeline();
    private Node outgoingDtlsPipelineRoot = createOutgoingDtlsPipeline();
    private String id;
    protected boolean dtlsHandshakeComplete = false;

    public IceDtlsTransportManager(String id, Conference conference)
            throws IOException
    {
        super(conference, true, 1, "ice-stream-" + id, null);
        this.id = id;
        this.ICE_STREAM_NAME = "ice-stream-" + id;
        executor = Executors.newCachedThreadPool(new NameableThreadFactory("Transport manager threadpool-" + id));
        iceAgent.addStateChangeListener(this::iceAgentStateChange);
        logger.info("BRIAN: finished IceDtlsTransportManager ctor");
    }

    //TODO: need to take another look and make sure we're properly replicating all the behavior of this
    // method in IceUdpTransportManager
    @Override
    public void startConnectivityEstablishment(IceUdpTransportPacketExtension remoteTransportInformation)
    {
        if (iceAgent.getState().isEstablished()) {
            logger.info(id + " with local ufrag " + iceAgent.getLocalUfrag() +
                    " is already established, not restarting");
            return;
        }
        logger.info(id + " with local ufrag " + iceAgent.getLocalUfrag() +
                " starting connectivity establishment with extension: " + remoteTransportInformation.toXML());
        // Get the remote fingerprints and set them in the DTLS stack so we
        // have them to do the DTLS handshake later
        List<DtlsFingerprintPacketExtension> fingerprintExtensions
                = remoteTransportInformation.getChildExtensionsOfType(DtlsFingerprintPacketExtension.class);

        Map<String, String> remoteFingerprints = new HashMap<>();
        fingerprintExtensions.forEach(fingerprintExtension -> {
            if (fingerprintExtension.getHash() != null && fingerprintExtension.getFingerprint() != null) {
                logger.info("Adding fingerprint " + fingerprintExtension.getHash() +
                        " -> " + fingerprintExtension.getFingerprint());
                remoteFingerprints.put(fingerprintExtension.getHash(), fingerprintExtension.getFingerprint());
            } else {
                logger.info("Ignoring empty transport extension");
            }
        });
        if (remoteFingerprints.isEmpty()) {
            logger.info(id + " with local ufrag " + iceAgent.getLocalUfrag() +
                    " empty transport extension, not starting connectivity yet");
            return;
        }

        // Set the remote ufrag/password
        if (remoteTransportInformation.getUfrag() != null) {
            iceAgent.getStream(ICE_STREAM_NAME).setRemoteUfrag(remoteTransportInformation.getUfrag());
        }
        if (remoteTransportInformation.getPassword() != null) {
            iceAgent.getStream(ICE_STREAM_NAME).setRemotePassword(remoteTransportInformation.getPassword());
        }

        // If ICE is running already, we try to update the checklists with the
        // candidates. Note that this is a best effort.
        boolean iceAgentStateIsRunning
                = IceProcessingState.RUNNING.equals(iceAgent.getState());

        List<CandidatePacketExtension> remoteCandidates
                = remoteTransportInformation.getChildExtensionsOfType(CandidatePacketExtension.class);
        if (iceAgentStateIsRunning && remoteCandidates.isEmpty()) {
            logger.info("ICE agent is already running and this extension contained no new candidates, returning");
            return;
        }

        int remoteCandidateCount = addRemoteCandidates(remoteCandidates, iceAgentStateIsRunning);

        if (iceAgentStateIsRunning) {
            if (remoteCandidateCount == 0) {
                // XXX Effectively, the check above but realizing that all
                // candidates were ignored:
                // iceAgentStateIsRunning && candidates.isEmpty().
            } else {
                // update all components of all streams
                iceAgent.getStreams()
                        .forEach(stream -> stream.getComponents()
                                .forEach(Component::updateRemoteCandidates));
            }
        }
        else if (remoteCandidateCount != 0)
        {
            // Once again, because the ICE Agent does not support adding
            // candidates after the connectivity establishment has been started
            // and because multiple transport-info JingleIQs may be used to send
            // the whole set of transport candidates from the remote peer to the
            // local peer, do not really start the connectivity establishment
            // until we have at least one remote candidate per ICE Component.
            if (iceAgent.getStreams().stream().allMatch(
                    stream -> stream.getComponents().stream().allMatch(
                            component -> component.getRemoteCandidateCount() >= 1)))
            {
                logger.info(
                        "We have remote candidates for all ICE components. "
                                + "Starting the ICE agent.");
                iceAgent.startConnectivityEstablishment();
            }
        }
        else if (iceAgent.getStream(ICE_STREAM_NAME).getRemoteUfrag() != null
                && iceAgent.getStream(ICE_STREAM_NAME).getRemotePassword() != null)
        {
            // We don't have any remote candidates, but we already know the
            // remote ufrag and password, so we can start ICE.
            logger.info("Starting ICE agent without remote candidates.");
            iceAgent.startConnectivityEstablishment();
        }
    }

    private Transceiver getTransceiver() {
        return this.transceiver;
    }

    /**
     *
     * @param candidates
     * @param iceAgentStateIsRunning
     * @return the number of network reachable remote candidates contained in the given list of candidates
     *
     * NOTE(brian): Almost the same as the one that was in IceUdpTransportManager, but we don't use the iceStream
     * member (we get the stream dynamically) and we always assume rtcpmux
     */
    private int addRemoteCandidates(
            List<CandidatePacketExtension> candidates,
            boolean iceAgentStateIsRunning)
    {
        // Sort the remote candidates (host < reflexive < relayed) in order to
        // create first the host, then the reflexive, the relayed candidates and
        // thus be able to set the relative-candidate matching the
        // rel-addr/rel-port attribute.
        Collections.sort(candidates);

        int generation = iceAgent.getGeneration();
{'fraction_non_alphanumeric': 0.08772984618467859, 'fraction_numerical': 0.001703492158925798, 'mean_word_length': 2.4834205933682374, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5563088', 'n_tokens_mistral': 5127, 'n_tokens_neox': 4911, 'n_words': 1466}