<commit_before>/**
 * Copyright JS Foundation and other contributors, http://js.foundation
 *
 * 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.
 **/

module.exports = function(RED) {
    "use strict";
    var ws = require("ws");
    var inspect = require("util").inspect;
    var url = require("url");
    var HttpsProxyAgent = require('https-proxy-agent');


    var serverUpgradeAdded = false;
    function handleServerUpgrade(request, socket, head) {
        const pathname = url.parse(request.url).pathname;
        if (listenerNodes.hasOwnProperty(pathname)) {
            listenerNodes[pathname].server.handleUpgrade(request, socket, head, function done(ws) {
                listenerNodes[pathname].server.emit('connection', ws, request);
            });
        } else {
            // Don't destroy the socket as other listeners may want to handle the
            // event.
        }
    }
    var listenerNodes = {};
    var activeListenerNodes = 0;


    // A node red node that sets up a local websocket server
    function WebSocketListenerNode(n) {
        // Create a RED node
        RED.nodes.createNode(this,n);
        var node = this;

        // Store local copies of the node configuration (as defined in the .html)
        node.path = n.path;
        node.wholemsg = (n.wholemsg === "true");

        node._inputNodes = [];    // collection of nodes that want to receive events
        node._clients = {};
        // match absolute url
        node.isServer = !/^ws{1,2}:\/\//i.test(node.path);
        node.closing = false;
        node.tls = n.tls;

        if (n.hb) {
            var heartbeat = parseInt(n.hb);
            if (heartbeat > 0) {
                node.heartbeat = heartbeat * 1000;
            }
        }

        function startconn() {    // Connect to remote endpoint
            node.tout = null;
            var prox, noprox;
            if (process.env.http_proxy) { prox = process.env.http_proxy; }
            if (process.env.HTTP_PROXY) { prox = process.env.HTTP_PROXY; }
            if (process.env.no_proxy) { noprox = process.env.no_proxy.split(","); }
            if (process.env.NO_PROXY) { noprox = process.env.NO_PROXY.split(","); }

            var noproxy = false;
            if (noprox) {
                for (var i in noprox) {
                    if (node.path.indexOf(noprox[i].trim()) !== -1) { noproxy=true; }
                }
            }

            var agent = undefined;
            if (prox && !noproxy) {
                agent = new HttpsProxyAgent(prox);
            }

            var options = {};
            if (agent) {
                options.agent = agent;
            }
            if (node.tls) {
                var tlsNode = RED.nodes.getNode(node.tls);
                if (tlsNode) {
                    tlsNode.addTLSOptions(options);
                }
            }
            var socket = new ws(node.path,options);
            socket.setMaxListeners(0);
            node.server = socket; // keep for closing
            handleConnection(socket);
        }

        function handleConnection(/*socket*/socket) {
            var id = RED.util.generateId();
            socket.nrId = id;
            socket.nrPendingHeartbeat = false;
            if (node.isServer) {
                node._clients[id] = socket;
                node.emit('opened',{count:Object.keys(node._clients).length,id:id});
            } else {
                if (node.heartbeat) {
                    node.heartbeatInterval = setInterval(function() {
                        if (socket.nrPendingHeartbeat) {
                            // No pong received
                            socket.terminate();
                            socket.nrErrorHandler(new Error("timeout"));
                            return;
                        }
                        socket.nrPendingHeartbeat = true;
                        socket.ping();
                    },node.heartbeat);
                }
            }
            socket.on('open',function() {
                if (!node.isServer) {
                    node.emit('opened',{count:'',id:id});<commit_msg>Only setup ws client heartbeat once it is connected Fixes #3264<commit_after>/**
 * Copyright JS Foundation and other contributors, http://js.foundation
 *
 * 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.
 **/

module.exports = function(RED) {
    "use strict";
    var ws = require("ws");
    var inspect = require("util").inspect;
    var url = require("url");
    var HttpsProxyAgent = require('https-proxy-agent');


    var serverUpgradeAdded = false;
    function handleServerUpgrade(request, socket, head) {
        const pathname = url.parse(request.url).pathname;
        if (listenerNodes.hasOwnProperty(pathname)) {
            listenerNodes[pathname].server.handleUpgrade(request, socket, head, function done(ws) {
                listenerNodes[pathname].server.emit('connection', ws, request);
            });
        } else {
            // Don't destroy the socket as other listeners may want to handle the
            // event.
        }
    }
    var listenerNodes = {};
    var activeListenerNodes = 0;


    // A node red node that sets up a local websocket server
    function WebSocketListenerNode(n) {
        // Create a RED node
        RED.nodes.createNode(this,n);
        var node = this;

        // Store local copies of the node configuration (as defined in the .html)
        node.path = n.path;
        node.wholemsg = (n.wholemsg === "true");

        node._inputNodes = [];    // collection of nodes that want to receive events
        node._clients = {};
        // match absolute url
        node.isServer = !/^ws{1,2}:\/\//i.test(node.path);
        node.closing = false;
        node.tls = n.tls;

        if (n.hb) {
            var heartbeat = parseInt(n.hb);
            if (heartbeat > 0) {
                node.heartbeat = heartbeat * 1000;
            }
        }

        function startconn() {    // Connect to remote endpoint
            node.tout = null;
            var prox, noprox;
            if (process.env.http_proxy) { prox = process.env.http_proxy; }
            if (process.env.HTTP_PROXY) { prox = process.env.HTTP_PROXY; }
            if (process.env.no_proxy) { noprox = process.env.no_proxy.split(","); }
            if (process.env.NO_PROXY) { noprox = process.env.NO_PROXY.split(","); }

            var noproxy = false;
            if (noprox) {
                for (var i in noprox) {
                    if (node.path.indexOf(noprox[i].trim()) !== -1) { noproxy=true; }
                }
            }

            var agent = undefined;
            if (prox && !noproxy) {
                agent = new HttpsProxyAgent(prox);
            }

            var options = {};
            if (agent) {
                options.agent = agent;
            }
            if (node.tls) {
                var tlsNode = RED.nodes.getNode(node.tls);
                if (tlsNode) {
                    tlsNode.addTLSOptions(options);
                }
            }
            var socket = new ws(node.path,options);
            socket.setMaxListeners(0);
            node.server = socket; // keep for closing
            handleConnection(socket);
        }

        function handleConnection(/*socket*/socket) {
            var id = RED.util.generateId();
            socket.nrId = id;
            socket.nrPendingHeartbeat = false;
            if (node.isServer) {
                node._clients[id] = socket;
                node.emit('opened',{count:Object.keys(node._clients).length,id:id});
            }
            socket.on('open',function() {
                if (!node.isServer) {
                    if (node.heartbeat) {
                        clearInterval(node.heartbeatInterval);
                        node.heartbeatInterval = setInterval(function() {
                            if (socket.nrPendingHeartbeat) {
                                // No pong received
                                socket.terminate();
                                socket.nrErrorHandler(new Error("timeout"));
                                return;
                            }
                            socket.nrPendingHeartbeat = true;
                            try {
                                socket.ping();
                            } catch(err) {}
                        },node.heartbeat);
                    }
                    node.emit('opened',{count:'',id:id});
{'fraction_non_alphanumeric': 0.10933868582294527, 'fraction_numerical': 0.003380519754912318, 'mean_word_length': 1.5760544217687076, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6889376', 'n_tokens_mistral': 2588, 'n_tokens_neox': 2457, 'n_words': 651}