<commit_before>import Emitter from './emitter'

class Peer extends Emitter {
    constructor(config) {
        super();
        this._pendingIncomingCandidates = [];
        this._pendingOutgoingCandidates = [];

        this._peerConnection = new RTCPeerConnection({
            optional: [{ RtpDataChannels: false }],
            iceServers: [
                { 'urls': ['stun:stun.l.google.com:19302'] },
                { 'urls': ['stun:stun1.l.google.com:19302'] },
                { 'urls': ['stun:stun2.l.google.com:19302'] },
                { 'urls': ['stun:stun3.l.google.com:19302'] },
                { 'urls': ['stun:stun4.l.google.com:19302'] },
            ].concat(config.iceServers),
        });

        this._peerConnection.onaddstream = e => {
            console.log('TRIBUTARY:PEER:onaddstream');
            if (e.stream) {
                this.stream = e.stream;
                this.emit('stream', e.stream);
            }
        };

        this._peerConnection.onicecandidate = e => {
            console.log('TRIBUTARY:PEER:onicecandidate', e.candidate);
            if (e.candidate) {
                if (this._pendingOutgoingCandidates) {
                    this._pendingOutgoingCandidates.push(e.candidate);
                    return;
                }
                this.emit('icecandidates', [e.candidate]);
            }
        };

        this._peerConnection.oniceconnectionstatechange = e => {
            console.log('TRIBUTARY:PEER:oniceconnectionstatechange', e);
        };
    }

    getOffer() {
        return this._peerConnection.createOffer({
            offerToReceiveVideo: true,
            offerToReceiveAudio: true,
        }).then(offer => {
            console.log('TRIBUTARY:PEER:offer', offer);
            return this._peerConnection.setLocalDescription(offer).then(() => offer);
        });
    }

    setAnswer(answer) {
        return this._peerConnection.setRemoteDescription(new RTCSessionDescription(answer))
        .then(() => {
            const candidates = this._pendingIncomingCandidates;
            this._pendingIncomingCandidates = null;
            this.addIceCandidates(candidates);
        });
    }

    setOffer(offer) {
        return this._peerConnection.setRemoteDescription(new RTCSessionDescription(offer))
        .then(() => {
            return this._peerConnection.createAnswer();
        })
        .then(answer => {
            this._peerConnection.setLocalDescription(answer);
            const candidates = this._pendingIncomingCandidates;
            this._pendingIncomingCandidates = null;
            this.addIceCandidates(candidates);
            return answer;
        });
    }

    addIceCandidates(candidates) {
        if (this._pendingIncomingCandidates) {
            this._pendingIncomingCandidates = this._pendingIncomingCandidates.concat(candidates);
            return;
        }

        candidates.forEach(candidate => {
            console.log('TRIBUTARY:PEER:addIceCandidate', candidate);
            this._peerConnection.addIceCandidate(new RTCIceCandidate(candidate));
        });
    }

    addStream(stream) {
        this._peerConnection.addStream(stream);
    }

    removeStream(stream) {
        this._peerConnection.removeStream(stream);
    }

    unblockOutgoingCandidates() {
        this.emit('icecandidates', this._pendingOutgoingCandidates);
        this._pendingOutgoingCandidates = null;
    }

    close() {
        this._peerConnection.close();
    }
}

const TributaryState = {
    READY: 'READY',
    BROADCASTING: 'BROADCASTING',
    LISTENING: 'LISTENING',
    BROADCAST_ENDED: 'BROADCAST_ENDED',
    RECONNECTING: 'RECONNECTING',
};

class Tributary extends Emitter {

    constructor(options) {
        super();
        this.state = TributaryState.READY;
        this._downstreamPeers = {};

        this._responseHandlers = {};

        this._socket = new WebSocket(options.url);
        this._socket.onerror = e => console.error('TRIBUTARY:WebSocket error', e);
        this._socket.onmessage = m => {
            try {
                m = JSON.parse(m.data);
            } catch(e) {
                console.error('TRIBUTARY:invalid JSON', m, e);
                return;
            }

            console.log('TRIBUTARY:message', m);

            // FIXME: not super useful, we need to instead have specific error responses for the various
            // commands.
            if (m.command === 'ERROR') {
                console.error('TRIBUTARY:error', m.message);
                return;
            }

            if (this._responseHandlers[m.command]) {
                this._responseHandlers[m.command].resolve(m);
                delete this._responseHandlers[m.command];
                return;
            }

            const commandName = m.command.toLowerCase().split('_').reduce((result, component) => {
                return result + component.substr(0, 1).toUpperCase() + component.substr(1)
            }, '');

            if (this['on' + commandName]) {
                this['on' + commandName](m);
                return;
            }

            console.warn('TRIBUTARY:unhandled message', m);
        };
    }

    sendAndWait(message) {
        return new Promise((resolve, reject) => {
            this._responseHandlers[message.command + '_RECEIVED'] = {
                resolve,
                reject
            };

            this.send(message);
        });
    }

    send(message) {
        this._socket.send(JSON.stringify(message));
    }

    startCamera(constraints) {
        return navigator.mediaDevices.getUserMedia(constraints);
    }

    stopCamera() {
        if (!this._stream) {
            return;
        }

        if (typeof this._stream.removeTrack === 'function') { // MediaStream
            let tracks = this.stream.getTracks();
            for (let i = tracks.length - 1; i >= 0; i--) {
                tracks[i].stop();
                this._stream.removeTrack(tracks[i]);
            };
        } else if (typeof this._stream.stop === 'function') { // LocalMediaStream
            this._stream.stop();
        }
        this._stream = null;<commit_msg>Fix a bug with stopping the camera<commit_after>import Emitter from './emitter'

class Peer extends Emitter {
    constructor(config) {
        super();
        this._pendingIncomingCandidates = [];
        this._pendingOutgoingCandidates = [];

        this._peerConnection = new RTCPeerConnection({
            optional: [{ RtpDataChannels: false }],
            iceServers: [
                { 'urls': ['stun:stun.l.google.com:19302'] },
                { 'urls': ['stun:stun1.l.google.com:19302'] },
                { 'urls': ['stun:stun2.l.google.com:19302'] },
                { 'urls': ['stun:stun3.l.google.com:19302'] },
                { 'urls': ['stun:stun4.l.google.com:19302'] },
            ].concat(config.iceServers),
        });

        this._peerConnection.onaddstream = e => {
            console.log('TRIBUTARY:PEER:onaddstream');
            if (e.stream) {
                this.stream = e.stream;
                this.emit('stream', e.stream);
            }
        };

        this._peerConnection.onicecandidate = e => {
            console.log('TRIBUTARY:PEER:onicecandidate', e.candidate);
            if (e.candidate) {
                if (this._pendingOutgoingCandidates) {
                    this._pendingOutgoingCandidates.push(e.candidate);
                    return;
                }
                this.emit('icecandidates', [e.candidate]);
            }
        };

        this._peerConnection.oniceconnectionstatechange = e => {
            console.log('TRIBUTARY:PEER:oniceconnectionstatechange', e);
        };
    }

    getOffer() {
        return this._peerConnection.createOffer({
            offerToReceiveVideo: true,
            offerToReceiveAudio: true,
        }).then(offer => {
            console.log('TRIBUTARY:PEER:offer', offer);
            return this._peerConnection.setLocalDescription(offer).then(() => offer);
        });
    }

    setAnswer(answer) {
        return this._peerConnection.setRemoteDescription(new RTCSessionDescription(answer))
        .then(() => {
            const candidates = this._pendingIncomingCandidates;
            this._pendingIncomingCandidates = null;
            this.addIceCandidates(candidates);
        });
    }

    setOffer(offer) {
        return this._peerConnection.setRemoteDescription(new RTCSessionDescription(offer))
        .then(() => {
            return this._peerConnection.createAnswer();
        })
        .then(answer => {
            this._peerConnection.setLocalDescription(answer);
            const candidates = this._pendingIncomingCandidates;
            this._pendingIncomingCandidates = null;
            this.addIceCandidates(candidates);
            return answer;
        });
    }

    addIceCandidates(candidates) {
        if (this._pendingIncomingCandidates) {
            this._pendingIncomingCandidates = this._pendingIncomingCandidates.concat(candidates);
            return;
        }

        candidates.forEach(candidate => {
            console.log('TRIBUTARY:PEER:addIceCandidate', candidate);
            this._peerConnection.addIceCandidate(new RTCIceCandidate(candidate));
        });
    }

    addStream(stream) {
        this._peerConnection.addStream(stream);
    }

    removeStream(stream) {
        this._peerConnection.removeStream(stream);
    }

    unblockOutgoingCandidates() {
        this.emit('icecandidates', this._pendingOutgoingCandidates);
        this._pendingOutgoingCandidates = null;
    }

    close() {
        this._peerConnection.close();
    }
}

const TributaryState = {
    READY: 'READY',
    BROADCASTING: 'BROADCASTING',
    LISTENING: 'LISTENING',
    BROADCAST_ENDED: 'BROADCAST_ENDED',
    RECONNECTING: 'RECONNECTING',
};

class Tributary extends Emitter {

    constructor(options) {
        super();
        this.state = TributaryState.READY;
        this._downstreamPeers = {};

        this._responseHandlers = {};

        this._socket = new WebSocket(options.url);
        this._socket.onerror = e => console.error('TRIBUTARY:WebSocket error', e);
        this._socket.onmessage = m => {
            try {
                m = JSON.parse(m.data);
            } catch(e) {
                console.error('TRIBUTARY:invalid JSON', m, e);
                return;
            }

            console.log('TRIBUTARY:message', m);

            // FIXME: not super useful, we need to instead have specific error responses for the various
            // commands.
            if (m.command === 'ERROR') {
                console.error('TRIBUTARY:error', m.message);
                return;
            }

            if (this._responseHandlers[m.command]) {
                this._responseHandlers[m.command].resolve(m);
                delete this._responseHandlers[m.command];
                return;
            }

            const commandName = m.command.toLowerCase().split('_').reduce((result, component) => {
                return result + component.substr(0, 1).toUpperCase() + component.substr(1)
            }, '');

            if (this['on' + commandName]) {
                this['on' + commandName](m);
                return;
            }

            console.warn('TRIBUTARY:unhandled message', m);
        };
    }

    sendAndWait(message) {
        return new Promise((resolve, reject) => {
            this._responseHandlers[message.command + '_RECEIVED'] = {
                resolve,
                reject
            };

            this.send(message);
        });
    }

    send(message) {
        this._socket.send(JSON.stringify(message));
    }

    startCamera(constraints) {
        return navigator.mediaDevices.getUserMedia(constraints);
    }

    stopCamera() {
        if (!this._stream) {
            return;
        }

        if (typeof this._stream.removeTrack === 'function') { // MediaStream
            let tracks = this._stream.getTracks();
            for (let i = tracks.length - 1; i >= 0; i--) {
                tracks[i].stop();
                this._stream.removeTrack(tracks[i]);
            };
        } else if (typeof this._stream.stop === 'function') { // LocalMediaStream
            this._stream.stop();
        }
        this._stream = null;
{'fraction_non_alphanumeric': 0.12392538523925385, 'fraction_numerical': 0.00551500405515004, 'mean_word_length': 1.8643437862950059, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6508028', 'n_tokens_mistral': 3569, 'n_tokens_neox': 3353, 'n_words': 511}