<commit_before>const { EventEmitter } = require('events')

const debug = require('debug')

const RoundRobinList = require('./roundrobinlist')
const lookup = require('./lookupd')
const { NSQDConnection } = require('./nsqdconnection')
const { ReaderConfig } = require('./config')
const { ReaderRdy } = require('./readerrdy')

/**
 * Reader provides high-level functionality for building robust NSQ
 * consumers. Reader is built upon the EventEmitter and thus supports various
 * hooks when different events occur.
 * @type {Reader}
 */
class Reader extends EventEmitter {
  static get ERROR () {
    return 'error'
  }
  static get MESSAGE () {
    return 'message'
  }
  static get DISCARD () {
    return 'discard'
  }
  static get NSQD_CONNECTED () {
    return 'nsqd_connected'
  }
  static get NSQD_CLOSED () {
    return 'nsqd_closed'
  }

  /**
   * @constructor
   * @param  {String} topic
   * @param  {String} channel
   * @param  {Object} options
   */
  constructor (topic, channel, options, ...args) {
    super(topic, channel, options, ...args)
    this.topic = topic
    this.channel = channel
    this.debug = debug(`nsqjs:reader:${this.topic}/${this.channel}`)
    this.config = new ReaderConfig(options)
    this.config.validate()

    this.debug('Configuration')
    this.debug(this.config)

    this.roundrobinLookupd = new RoundRobinList(
      this.config.lookupdHTTPAddresses
    )

    this.readerRdy = new ReaderRdy(
      this.config.maxInFlight,
      this.config.maxBackoffDuration,
      `${this.topic}/${this.channel}`
    )

    this.connectIntervalId = null
    this.connectionIds = []
  }

  /**
   * Adds a connection to nsqd at the configured address.
   *
   * @return {undefined}
   */
  connect () {
    let delayedStart
    const interval = this.config.lookupdPollInterval * 1000
    const delay = Math.random() * this.config.lookupdPollJitter * interval

    // Connect to provided nsqds.
    if (this.config.nsqdTCPAddresses.length) {
      const directConnect = () => {
        // Don't establish new connections while the Reader is paused.
        if (this.isPaused()) return

        if (this.connectionIds.length < this.config.nsqdTCPAddresses.length) {
          return this.config.nsqdTCPAddresses.forEach(addr => {
            const [address, port] = addr.split(':')
            this.connectToNSQD(address, Number(port))
          })
        }
      }

      delayedStart = () => {
        this.connectIntervalId = setInterval(directConnect.bind(this), interval)
      }

      // Connect immediately.
      directConnect()

      // Start interval for connecting after delay.
      setTimeout(delayedStart, delay).unref()
    }

    delayedStart = () => {
      this.connectIntervalId = setInterval(
        this.queryLookupd.bind(this),
        interval
      )
    }

    // Connect immediately.
    this.queryLookupd()

    // Start interval for querying lookupd after delay.
    setTimeout(delayedStart, delay)
  }

  /**
   * Close all connections and prevent any periodic callbacks.
   * @return {Array} The closed connections.
   */
  close () {
    clearInterval(this.connectIntervalId)
    return this.readerRdy.close()
  }

  /**
   * Pause all connections
   * @return {Array} The paused connections.
   */
  pause () {
    this.debug('pause')
    return this.readerRdy.pause()
  }

  /**
   * Unpause all connections
   * @return {Array} The unpaused connections.
   */
  unpause () {
    this.debug('unpause')
    return this.readerRdy.unpause()
  }

  /**
   * @return {Boolean}
   */
  isPaused () {
    return this.readerRdy.isPaused()
  }<commit_msg>Simplify Reader.connect. Ensure exit on close. (#193)<commit_after>const { EventEmitter } = require('events')

const debug = require('debug')

const RoundRobinList = require('./roundrobinlist')
const lookup = require('./lookupd')
const { NSQDConnection } = require('./nsqdconnection')
const { ReaderConfig } = require('./config')
const { ReaderRdy } = require('./readerrdy')

/**
 * Reader provides high-level functionality for building robust NSQ
 * consumers. Reader is built upon the EventEmitter and thus supports various
 * hooks when different events occur.
 * @type {Reader}
 */
class Reader extends EventEmitter {
  static get ERROR () {
    return 'error'
  }
  static get MESSAGE () {
    return 'message'
  }
  static get DISCARD () {
    return 'discard'
  }
  static get NSQD_CONNECTED () {
    return 'nsqd_connected'
  }
  static get NSQD_CLOSED () {
    return 'nsqd_closed'
  }

  /**
   * @constructor
   * @param  {String} topic
   * @param  {String} channel
   * @param  {Object} options
   */
  constructor (topic, channel, options, ...args) {
    super(topic, channel, options, ...args)
    this.topic = topic
    this.channel = channel
    this.debug = debug(`nsqjs:reader:${this.topic}/${this.channel}`)
    this.config = new ReaderConfig(options)
    this.config.validate()

    this.debug('Configuration')
    this.debug(this.config)

    this.roundrobinLookupd = new RoundRobinList(
      this.config.lookupdHTTPAddresses
    )

    this.readerRdy = new ReaderRdy(
      this.config.maxInFlight,
      this.config.maxBackoffDuration,
      `${this.topic}/${this.channel}`
    )

    this.lookupdIntervalId = null
    this.directIntervalId = null
    this.connectionIds = []
  }

  /**
   * Adds a connection to nsqd at the configured address.
   *
   * @return {undefined}
   */
  connect () {
    this._connectTCPAddresses()
    this._connectLookupd()
  }

  _connectInterval () {
    return this.config.lookupdPollInterval * 1000
  }

  _connectTCPAddresses () {
    const directConnect = () => {
      // Don't establish new connections while the Reader is paused.
      if (this.isPaused()) return

      if (this.connectionIds.length < this.config.nsqdTCPAddresses.length) {
        return this.config.nsqdTCPAddresses.forEach(addr => {
          const [address, port] = addr.split(':')
          this.connectToNSQD(address, Number(port))
        })
      }
    }

    this.lookupdIntervalId = setInterval(() => { directConnect() },
      this._connectInterval())

    // Connect immediately.
    directConnect()
  }

  _connectLookupd () {
    this.directIntervalId = setInterval(
      () => { this.queryLookupd() },
      this._connectInterval()
    )

    // Connect immediately.
    this.queryLookupd()
  }

  /**
   * Close all connections and prevent any periodic callbacks.
   * @return {Array} The closed connections.
   */
  close () {
    clearInterval(this.directIntervalId)
    clearInterval(this.lookupdIntervalId)
    return this.readerRdy.close()
  }

  /**
   * Pause all connections
   * @return {Array} The paused connections.
   */
  pause () {
    this.debug('pause')
    return this.readerRdy.pause()
  }

  /**
   * Unpause all connections
   * @return {Array} The unpaused connections.
   */
  unpause () {
    this.debug('unpause')
    return this.readerRdy.unpause()
  }

  /**
   * @return {Boolean}
   */
  isPaused () {
    return this.readerRdy.isPaused()
  }
{'fraction_non_alphanumeric': 0.1257995735607676, 'fraction_numerical': 0.0015636105188343995, 'mean_word_length': 2.768612747723621, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, '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': '5672988', 'n_tokens_mistral': 2195, 'n_tokens_neox': 2136, 'n_words': 486}