<commit_before>'use strict';

const net = require('net'),
      buffer = require('buffer'),
      Rx = require('rxjs/Rx'),
      {Msg, Reader, deserialize, frame} = require('./proto');

class ForwardClient {
    constructor(logger, host, port = 5555, minFlushBufferSize = 1000, maxBufferSize = 2000, maxFlushInterval = 1000, reconnectTimeout = 1000) {
        this.logger = logger;
        this.host = host;
        this.port = port;
        this.client = null;
        this.reconnectTimeout = reconnectTimeout;
        this.minFlushBufferSize = minFlushBufferSize;
        this.maxBufferSize = maxBufferSize;
        this.maxFlushInterval = maxFlushInterval;
        this.buffer = new Buffer(0);
        this.bufferMessageCount = 0;
        this.reader = new Reader();
        this.messageLossCounter = new Rx.BehaviorSubject(0);
        this.flushHandler = null;
        this.flushTimer = setInterval(this.flush.bind(this), this.maxFlushInterval);
    }

    /** connect must be called after subscribing to state in order to receive first failure */
    connect() {
        const client = net.connect({
          host: this.host,
          port: this.port
        }, () => {
            this.logger.info('Riemann client connected to:', this.host + ':' + this.port);
            this.client = client;
        }).on('end', () => {
            this.logger.warn('Riemann client disconnected, attempting reconnect in ', this.reconnectTimeout, 'ms');
            this.handleRemoteError(new Error('Riemann client disconnected'));
            this.triggerReconnect();
        }).on('data', (data) => {
            const messages = this.reader.readMessagesFromBuffer(data);
            for (const raw of messages) {
                const ack = deserialize(raw);
                this.logger.debug('Riemann server responsed with', ack);
                if (ack.ok) {
                    this.flushHandler.resolve();
                }
                else {
                    const error = 'Riemann server returned error in response: ' + ack.error;
                    this.logger.error(error);
                    this.handleRemoteError(new Error(error));
                }
            }
        }).on('error', (error) => {
            this.logger.error('Riemann client error:', error, ' attempting reconnect in ', this.reconnectTimeout, 'ms');
            this.handleRemoteError(error);
            this.triggerReconnect();
        });
    }

    handleRemoteError(error) {
        if (this.flushHandler !== null) {
            this.flushHandler.reject(error);
        }
    }

    triggerReconnect() {
        this.client = null;
        setTimeout(() => { this.connect(); }, this.reconnectTimeout);
    }

    enqueue(messages) {
        if (messages.length === 0) {
            return;
        }

        const totalLength = messages.reduce((acc, curr) => acc + curr.length, this.buffer.length);
        if (totalLength > this.maxBufferSize) {
            this.messageLossCounter.next(this.messageLossCounter.getValue() + messages.length);
            this.logger.error('Riemann client buffer full, dropping ' + messages.length + ' message(s)');
        }
        this.bufferMessageCount += messages.length;
        const chunks = [this.buffer].concat(messages);
        this.buffer = Buffer.concat(chunks, totalLength);
        if (totalLength >= this.minFlushBufferSize) {
            this.flush();<commit_msg>bugfix: unbound forward client buffer size<commit_after>'use strict';

const net = require('net'),
      buffer = require('buffer'),
      Rx = require('rxjs/Rx'),
      {Msg, Reader, deserialize, frame} = require('./proto');

class ForwardClient {
    constructor(logger, host, port = 5555, minFlushBufferSize = 1000, maxBufferSize = 2000, maxFlushInterval = 1000, reconnectTimeout = 1000) {
        this.logger = logger;
        this.host = host;
        this.port = port;
        this.client = null;
        this.reconnectTimeout = reconnectTimeout;
        this.minFlushBufferSize = minFlushBufferSize;
        this.maxBufferSize = maxBufferSize;
        this.maxFlushInterval = maxFlushInterval;
        this.buffer = new Buffer(0);
        this.bufferMessageCount = 0;
        this.reader = new Reader();
        this.messageLossCounter = new Rx.BehaviorSubject(0);
        this.flushHandler = null;
        this.flushTimer = setInterval(this.flush.bind(this), this.maxFlushInterval);
    }

    /** connect must be called after subscribing to state in order to receive first failure */
    connect() {
        const client = net.connect({
          host: this.host,
          port: this.port
        }, () => {
            this.logger.info('Riemann client connected to:', this.host + ':' + this.port);
            this.client = client;
        }).on('end', () => {
            this.logger.warn('Riemann client disconnected, attempting reconnect in ', this.reconnectTimeout, 'ms');
            this.handleRemoteError(new Error('Riemann client disconnected'));
            this.triggerReconnect();
        }).on('data', (data) => {
            const messages = this.reader.readMessagesFromBuffer(data);
            for (const raw of messages) {
                const ack = deserialize(raw);
                this.logger.debug('Riemann server responsed with', ack);
                if (ack.ok) {
                    this.flushHandler.resolve();
                }
                else {
                    const error = 'Riemann server returned error in response: ' + ack.error;
                    this.logger.error(error);
                    this.handleRemoteError(new Error(error));
                }
            }
        }).on('error', (error) => {
            this.logger.error('Riemann client error:', error, ' attempting reconnect in ', this.reconnectTimeout, 'ms');
            this.handleRemoteError(error);
            this.triggerReconnect();
        });
    }

    handleRemoteError(error) {
        if (this.flushHandler !== null) {
            this.flushHandler.reject(error);
        }
    }

    triggerReconnect() {
        this.client = null;
        setTimeout(() => { this.connect(); }, this.reconnectTimeout);
    }

    enqueue(messages) {
        if (messages.length === 0) {
            return;
        }

        const totalLength = messages.reduce((acc, curr) => acc + curr.length, this.buffer.length);
        if (totalLength > this.maxBufferSize) {
            this.messageLossCounter.next(this.messageLossCounter.getValue() + messages.length);
            this.logger.error('Riemann client buffer full, dropping ' + messages.length + ' message(s)');
            return;
        }
        this.bufferMessageCount += messages.length;
        const chunks = [this.buffer].concat(messages);
        this.buffer = Buffer.concat(chunks, totalLength);
        if (totalLength >= this.minFlushBufferSize) {
            this.flush();
{'fraction_non_alphanumeric': 0.1237803990097568, 'fraction_numerical': 0.0069899519440803845, 'mean_word_length': 2.126081019572144, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, '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': '771532', 'n_tokens_mistral': 1895, 'n_tokens_neox': 1792, 'n_words': 401}