<commit_before>'use strict';
const {EventEmitter} = require('events');
const WebSocket = require('ws');
// temporary: RangeSet will be moved to api/common soon
const RangeSet = require('./rangeset').RangeSet;
const {RippledError, DisconnectedError, NotConnectedError,
  TimeoutError, UnexpectedError} = require('./errors');

function isStreamMessageType(type) {
  return type === 'ledgerClosed' ||
         type === 'transaction' ||
         type === 'path_find';
}

class Connection extends EventEmitter {
  constructor(url, options = {}) {
    super();
    this._url = url;
    this._timeout = options.timeout || (20 * 1000);
    this._isReady = false;
    this._ws = null;
    this._ledgerVersion = null;
    this._availableLedgerVersions = new RangeSet();
    this._nextRequestID = 1;
  }

  // return value is array of arguments to Connection.emit
  _parseMessage(message) {
    const data = JSON.parse(message);
    if (data.type === 'response') {
      if (!(Number.isInteger(data.id) && data.id >= 0)) {
        throw new UnexpectedError('valid id not found in response');
      }
      return [data.id.toString(), data];
    } else if (isStreamMessageType(data.type)) {
      if (data.type === 'ledgerClosed') {
        this._ledgerVersion = Number(data.ledger_index);
        this._availableLedgerVersions.reset();
        this._availableLedgerVersions.parseAndAddRanges(
          data.validated_ledgers);
      }
      return [data.type, data];
    } else if (data.type === undefined && data.error) {
      return ['error', data.error, data.error_message];  // e.g. slowDown
    }
    throw new UnexpectedError('unrecognized message type: ' + data.type);
  }

  _onMessage(message) {
    let parameters;
    try {
      parameters = this._parseMessage(message);
    } catch (error) {
      this.emit('error', 'badMessage', message);
      return;
    }
    // we don't want this inside the try/catch or exceptions in listener
    // will be caught
    this.emit.apply(this, parameters);
  }

  get state() {
    return this._ws ? this._ws.readyState : WebSocket.CLOSED;
  }

  get _shouldBeConnected() {
    return this._ws !== null;
  }

  isConnected() {
    return this.state === WebSocket.OPEN && this._isReady;
  }

  _onUnexpectedClose() {
    this._isReady = false;
    this.connect().then();
  }

  _onOpen() {
    const request = {
      command: 'subscribe',
      streams: ['ledger']
    };
    return this.request(request).then(response => {
      this._ledgerVersion = Number(response.ledger_index);
      this._availableLedgerVersions.parseAndAddRanges(
        response.validated_ledgers);
      this._isReady = true;
      this.emit('connected');
    });
  }

  connect() {
    return new Promise((resolve, reject) => {
      if (this.state === WebSocket.OPEN) {
        resolve();
      } else if (this.state === WebSocket.CONNECTING) {
        this._ws.once('open', resolve);
      } else {
        this._ws = new WebSocket(this._url);
        this._ws.on('message', this._onMessage.bind(this));
        this._ws.once('close', () => this._onUnexpectedClose);
        this._ws.once('open', () => this._onOpen().then(resolve, reject));
      }
    });
  }

  disconnect() {
    return new Promise(resolve => {
      if (this.state === WebSocket.CLOSED) {
        resolve();
      } else if (this.state === WebSocket.CLOSING) {
        this._ws.once('close', resolve);
      } else {
        this._ws.removeListener('close', this._onUnexpectedClose);
        this._ws.once('close', () => {
          this._ws = null;
          this._isReady = false;
          resolve();
        });
        this._ws.close();
      }
    });
  }

  reconnect() {
    return this.disconnect().then(() => this.connect());
  }

  _whenReady(promise) {
    return new Promise((resolve, reject) => {
      if (!this._shouldBeConnected) {
        reject(new NotConnectedError());
      } else if (this.state === WebSocket.OPEN && this._isReady) {
        promise.then(resolve, reject);
      } else {
        this.once('connected', () => promise.then(resolve, reject));
      }
    });
  }

  getLedgerVersion() {
    return this._whenReady(Promise.resolve(this._ledgerVersion));
  }

  hasLedgerVersions(lowLedgerVersion, highLedgerVersion) {
    return this._whenReady(Promise.resolve(
      this._availableLedgerVersions.containsRange(
        lowLedgerVersion, highLedgerVersion || this._ledgerVersion)));<commit_msg>Use Promise.resolve<commit_after>'use strict';
const {EventEmitter} = require('events');
const WebSocket = require('ws');
// temporary: RangeSet will be moved to api/common soon
const RangeSet = require('./rangeset').RangeSet;
const {RippledError, DisconnectedError, NotConnectedError,
  TimeoutError, UnexpectedError} = require('./errors');

function isStreamMessageType(type) {
  return type === 'ledgerClosed' ||
         type === 'transaction' ||
         type === 'path_find';
}

class Connection extends EventEmitter {
  constructor(url, options = {}) {
    super();
    this._url = url;
    this._timeout = options.timeout || (20 * 1000);
    this._isReady = false;
    this._ws = null;
    this._ledgerVersion = null;
    this._availableLedgerVersions = new RangeSet();
    this._nextRequestID = 1;
  }

  // return value is array of arguments to Connection.emit
  _parseMessage(message) {
    const data = JSON.parse(message);
    if (data.type === 'response') {
      if (!(Number.isInteger(data.id) && data.id >= 0)) {
        throw new UnexpectedError('valid id not found in response');
      }
      return [data.id.toString(), data];
    } else if (isStreamMessageType(data.type)) {
      if (data.type === 'ledgerClosed') {
        this._ledgerVersion = Number(data.ledger_index);
        this._availableLedgerVersions.reset();
        this._availableLedgerVersions.parseAndAddRanges(
          data.validated_ledgers);
      }
      return [data.type, data];
    } else if (data.type === undefined && data.error) {
      return ['error', data.error, data.error_message];  // e.g. slowDown
    }
    throw new UnexpectedError('unrecognized message type: ' + data.type);
  }

  _onMessage(message) {
    let parameters;
    try {
      parameters = this._parseMessage(message);
    } catch (error) {
      this.emit('error', 'badMessage', message);
      return;
    }
    // we don't want this inside the try/catch or exceptions in listener
    // will be caught
    this.emit.apply(this, parameters);
  }

  get _state() {
    return this._ws ? this._ws.readyState : WebSocket.CLOSED;
  }

  get _shouldBeConnected() {
    return this._ws !== null;
  }

  isConnected() {
    return this._state === WebSocket.OPEN && this._isReady;
  }

  _onUnexpectedClose() {
    this._isReady = false;
    this.connect().then();
  }

  _onOpen() {
    const request = {
      command: 'subscribe',
      streams: ['ledger']
    };
    return this.request(request).then(response => {
      this._ledgerVersion = Number(response.ledger_index);
      this._availableLedgerVersions.parseAndAddRanges(
        response.validated_ledgers);
      this._isReady = true;
      this.emit('connected');
    });
  }

  connect() {
    return new Promise((resolve, reject) => {
      if (this._state === WebSocket.OPEN) {
        resolve();
      } else if (this._state === WebSocket.CONNECTING) {
        this._ws.once('open', resolve);
      } else {
        this._ws = new WebSocket(this._url);
        this._ws.on('message', this._onMessage.bind(this));
        this._ws.once('close', () => this._onUnexpectedClose);
        this._ws.once('open', () => this._onOpen().then(resolve, reject));
      }
    });
  }

  disconnect() {
    return new Promise(resolve => {
      if (this._state === WebSocket.CLOSED) {
        resolve();
      } else if (this._state === WebSocket.CLOSING) {
        this._ws.once('close', resolve);
      } else {
        this._ws.removeListener('close', this._onUnexpectedClose);
        this._ws.once('close', () => {
          this._ws = null;
          this._isReady = false;
          resolve();
        });
        this._ws.close();
      }
    });
  }

  reconnect() {
    return this.disconnect().then(() => this.connect());
  }

  _whenReady(promise) {
    return new Promise((resolve, reject) => {
      if (!this._shouldBeConnected) {
        reject(new NotConnectedError());
      } else if (this._state === WebSocket.OPEN && this._isReady) {
        promise.then(resolve, reject);
      } else {
        this.once('connected', () => promise.then(resolve, reject));
      }
    });
  }

  getLedgerVersion() {
    return this._whenReady(Promise.resolve(this._ledgerVersion));
  }

  hasLedgerVersions(lowLedgerVersion, highLedgerVersion) {
    return this._whenReady(Promise.resolve(
      this._availableLedgerVersions.containsRange(
        lowLedgerVersion, highLedgerVersion || this._ledgerVersion)));
{'fraction_non_alphanumeric': 0.14701568317725375, 'fraction_numerical': 0.0018052578133814735, 'mean_word_length': 2.7896536981616076, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, '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': '5974799', 'n_tokens_mistral': 2709, 'n_tokens_neox': 2676, 'n_words': 538}