<commit_before>///base on node built-in http module
const EventEmitter = require("events");
const util = require('util');
const request = require("request");
const config = require("./config.js").getAll();
const Readable = require('stream').Readable;
const http2 = require("http2");
const fs = require('fs');
const Dicer = require('dicer');
const BufferManager = require("./wakeup/buffermanager").BufferManager;

function DownStream() {
    EventEmitter.call(this);
    this.init();
}



//this.state: connecting/closed/connected

DownStream.prototype.isConnected = function() {
    return this.state == "connected";
};

DownStream.prototype.init = async function() {
    var self = this;
    if (this.state == "connecting") {
        return;
    }

    if (this.req) {
        try {
            this.req.rstWithCancel();
        } catch (e) {}
        this.req = null;
    }

    if (this.pingInterval) {
        clearInterval(this.pingInterval);
        if (this.pingReq) {
            try {
                this.pingReq.rstWithCancel();
            } catch (e) {}
            this.pingReq = null;
        }
    }

    if (this.http2session) {
        try {
            this.http2session.shutdown({
                graceful: true
            });
        } catch (e) {}
        this.http2session = null;
        await new Promise((resolve, reject) => {
            setTimeout(resolve, 1000);
        });
    }
    console.log(config.oauth_token);
    console.log("https://" + config.ip);
    this.http2session = http2.connect("https://" + config.ip,{rejectUnauthorized:false});
    this.http2session.socket.setNoDelay(true);
    
    this.http2session.on("error", () => {
        this.state = "closed";
        console.log('downstream session error!!!!!!!!');
    });
    this.http2session.on("close", () => {
        this.state = "closed";
        this.emit("sessionClosed");
        console.log('downstream session closed!!!!!!!!');
        this.init();
    });
    var logid = config.device_id + "_" + new Date().getTime() + "_monitor";
    console.log("downstream logid:" + logid);
    this.req = this.http2session.request({
        ":path": config.directive_uri,
        "SAIYALOGID": logid,
        "authorization": "Bearer " + config.oauth_token,
        "Dueros-Device-Id": config.device_id
    });
    this.req.on("error", (e) => {
        console.log('downstream error!!!!!!!!' + e.toString());
    });
    this.req.on("streamClosed", () => {
        this.emit("streamClosed");
        console.log('downstream closed');
        this.init();
    });
    this.http2session.setTimeout(0, () => {
        console.log("downstream session timeout");
    });
    this.req.setTimeout(0, () => {
        console.log("downstream stream timeout");
    });
    this.state = "connecting";
    if (this.pingInterval) {
        clearInterval(this.pingInterval);
    }
    this.pingInterval = setInterval(() => {
        if (!this.http2session || this.http2session.aborted || this.http2session.destroyed) {
            console.log('downstream ping error, stream closed');
            if (this.pingInterval) {
                clearInterval(this.pingInterval);
            }
            return;
        }
        var req = this.pingReq = this.http2session.request({
            ":path": config.ping_uri,
            "authorization": "Bearer " + config.oauth_token,
            "Dueros-Device-Id": config.device_id
        });
        req.setTimeout(5000, () => {<commit_msg>fix downstream<commit_after>///base on node built-in http module
const EventEmitter = require("events");
const util = require('util');
const request = require("request");
const config = require("./config.js").getAll();
const Readable = require('stream').Readable;
const http2 = require("http2");
const fs = require('fs');
const Dicer = require('dicer');
const BufferManager = require("./wakeup/buffermanager").BufferManager;

function DownStream() {
    EventEmitter.call(this);
    this.init();
}



//this.state: connecting/closed/connected

DownStream.prototype.isConnected = function() {
    return this.state == "connected";
};
function sleep(sec){
    return new Promise((resolve,reject)=>{
        setTimeout(()=>{
            resolve();
        },sec*1000);
    });
}

DownStream.prototype.init = async function() {
    var self = this;
    if (this.state == "connecting") {
        return;
    }

    if (this.req) {
        try {
            this.req.rstWithCancel();
        } catch (e) {}
        this.req = null;
    }

    if (this.pingInterval) {
        clearInterval(this.pingInterval);
        if (this.pingReq) {
            try {
                this.pingReq.rstWithCancel();
            } catch (e) {}
            this.pingReq = null;
        }
    }

    if (this.http2session) {
        try {
            this.http2session.shutdown({
                graceful: true
            });
        } catch (e) {}
        this.http2session = null;
        await new Promise((resolve, reject) => {
            setTimeout(resolve, 1000);
        });
    }
    console.log(config.oauth_token);
    console.log("https://" + config.ip);
    this.http2session = http2.connect("https://" + config.ip,{rejectUnauthorized:false});
    this.http2session.socket.setNoDelay(true);
    
    this.http2session.on("error", () => {
        this.state = "closed";
        console.log('downstream session error!!!!!!!!');
    });
    this.http2session.on("close", async () => {
        this.state = "closed";
        this.emit("sessionClosed");
        console.log('downstream session closed!!!!!!!!');
        await sleep(1);
        this.init();
    });
    var logid = config.device_id + "_" + new Date().getTime()+(parseInt(Math.random()*100,10)) + "_monitor";
    console.log("downstream logid:" + logid);
    this.req = this.http2session.request({
        ":path": config.directive_uri,
        "SAIYALOGID": logid,
        "authorization": "Bearer " + config.oauth_token,
        "Dueros-Device-Id": config.device_id
    });
    this.req.on("error", (e) => {
        console.log('downstream error!!!!!!!!' + e.toString());
    });
    this.req.on("streamClosed", () => {
        this.emit("streamClosed");
        console.log('downstream closed');
        try {
            this.http2session.shutdown({
                graceful: true
            });
        } catch (e) {}
    });
    this.http2session.setTimeout(0, () => {
        console.log("downstream session timeout");
    });
    this.req.setTimeout(0, () => {
        console.log("downstream stream timeout");
    });
    this.state = "connecting";
    if (this.pingInterval) {
        clearInterval(this.pingInterval);
    }
    this.pingInterval = setInterval(() => {
        if (!this.http2session || this.http2session.aborted || this.http2session.destroyed) {
            console.log('downstream ping error, stream closed');
            if (this.pingInterval) {
                clearInterval(this.pingInterval);
            }
            return;
        }
        var req = this.pingReq = this.http2session.request({
            ":path": config.ping_uri,
            "authorization": "Bearer " + config.oauth_token,
            "Dueros-Device-Id": config.device_id
        });
        req.setTimeout(5000, () => {
{'fraction_non_alphanumeric': 0.1630089717046239, 'fraction_numerical': 0.008695652173913044, 'mean_word_length': 2.29663330300273, 'pattern_counts': {'":': 18, '<': 3, '<?xml version=': 0, '>': 23, 'https://': 4, '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': '6561125', 'n_tokens_mistral': 2182, 'n_tokens_neox': 2134, 'n_words': 377}