<commit_before>'use strict';

var Promise = require("bluebird");
var EventEmitter = require("events").EventEmitter;
var Utils = require("cyclon.p2p-common");

function RTC(signallingService, channelFactory) {

    Utils.checkArguments(arguments, 2);

    this._signallingService = signallingService;
    this._channelFactory = channelFactory;
    this._channelListeners = {};
    this._connected = false;
}

RTC.prototype = Object.create(EventEmitter.prototype);

RTC.prototype.connect = function (metadataProviders, rooms) {
    var self = this;
    if(!this._connected) {
        this._signallingService.connect(metadataProviders, rooms);
        this._signallingService.on("offer", function (message) {
            self.handleOffer(message);
        });
        this.connected = true;
    }
};

RTC.prototype.createNewPointer = function (metaData) {
    return this._signallingService.createNewPointer(metaData);
};

RTC.prototype.getLocalId = function () {
    return this._signallingService.getLocalId();
};

RTC.prototype.onChannel = function (type, callback) {
    this._channelListeners[type] = callback;
};

RTC.prototype.openChannel = function (type, remotePointer) {
    var channel = this._channelFactory.createChannel(remotePointer);

    return channel.createOffer(type)
        .then(channel.sendOffer)
        .then(channel.startListeningForRemoteIceCandidates)
        .then(channel.waitForAnswer)
        .then(channel.handleAnswer)
        .then(channel.startSendingIceCandidates)
        .then(channel.waitForChannelToOpen)
        .then(channel.stopSendingIceCandidates)
        .catch(function (error) {
            // If an error occurs here, cleanup our attempted channel
            // establishment resources before continuing
            channel.close();
            throw error;
        });
};

RTC.prototype.handleOffer = function (offerMessage) {
    var self = this;
    var channelType = offerMessage.channelType;
    var listener = this._channelListeners[channelType];
    var remotePointer = offerMessage.sourcePointer;
    var correlationId = offerMessage.correlationId;

    self.emit("offerReceived", channelType, offerMessage.sourcePointer);

    if (listener) {
        var channel = this._channelFactory.createChannel(remotePointer, correlationId);

        channel.createAnswer(offerMessage.sessionDescription)
            .then(channel.startListeningForRemoteIceCandidates)
            .then(channel.sendAnswer)
            .then(channel.startSendingIceCandidates)
            .then(channel.waitForChannelEstablishment)
            .then(channel.waitForChannelToOpen)
            .then(channel.stopSendingIceCandidates)
            .then(listener)
            .catch(Promise.TimeoutError, function () {
                self.emit("incomingTimeout", channelType, offerMessage.sourcePointer);
            })
            .catch(function (e) {
                self.emit("incomingError", channelType, offerMessage.sourcePointer, e);
            })
            .finally(channel.close);<commit_msg>Only automatically close connection if an error occurs<commit_after>'use strict';

var Promise = require("bluebird");
var EventEmitter = require("events").EventEmitter;
var Utils = require("cyclon.p2p-common");

function RTC(signallingService, channelFactory) {

    Utils.checkArguments(arguments, 2);

    this._signallingService = signallingService;
    this._channelFactory = channelFactory;
    this._channelListeners = {};
    this._connected = false;
}

RTC.prototype = Object.create(EventEmitter.prototype);

RTC.prototype.connect = function (metadataProviders, rooms) {
    var self = this;
    if (!this._connected) {
        this._signallingService.connect(metadataProviders, rooms);
        this._signallingService.on("offer", function (message) {
            self.handleOffer(message);
        });
        this.connected = true;
    }
};

RTC.prototype.createNewPointer = function (metaData) {
    return this._signallingService.createNewPointer(metaData);
};

RTC.prototype.getLocalId = function () {
    return this._signallingService.getLocalId();
};

RTC.prototype.onChannel = function (type, callback) {
    this._channelListeners[type] = callback;
};

RTC.prototype.openChannel = function (type, remotePointer) {
    var channel = this._channelFactory.createChannel(remotePointer);

    return channel.createOffer(type)
        .then(channel.sendOffer)
        .then(channel.startListeningForRemoteIceCandidates)
        .then(channel.waitForAnswer)
        .then(channel.handleAnswer)
        .then(channel.startSendingIceCandidates)
        .then(channel.waitForChannelToOpen)
        .then(channel.stopSendingIceCandidates)
        .catch(function (error) {
            // If an error occurs here, cleanup our attempted channel
            // establishment resources before continuing
            channel.close();
            throw error;
        });
};

RTC.prototype.handleOffer = function (offerMessage) {
    var self = this;
    var channelType = offerMessage.channelType;
    var listener = this._channelListeners[channelType];
    var remotePointer = offerMessage.sourcePointer;
    var correlationId = offerMessage.correlationId;

    self.emit("offerReceived", channelType, offerMessage.sourcePointer);

    if (listener) {
        var channel = this._channelFactory.createChannel(remotePointer, correlationId);

        channel.createAnswer(offerMessage.sessionDescription)
            .then(channel.startListeningForRemoteIceCandidates)
            .then(channel.sendAnswer)
            .then(channel.startSendingIceCandidates)
            .then(channel.waitForChannelEstablishment)
            .then(channel.waitForChannelToOpen)
            .then(channel.stopSendingIceCandidates)
            .then(listener)
            .catch(Promise.TimeoutError, function () {
                self.emit("incomingTimeout", channelType, offerMessage.sourcePointer);
                channel.close();
            })
            .catch(function (e) {
                self.emit("incomingError", channelType, offerMessage.sourcePointer, e);
                channel.close();
            });
{'fraction_non_alphanumeric': 0.10365058670143416, 'fraction_numerical': 0.000651890482398957, 'mean_word_length': 3.118791946308725, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '1003503', 'n_tokens_mistral': 1664, 'n_tokens_neox': 1593, 'n_words': 282}