<commit_before>"use strict"

const EventEmitter = require( 'events' ).EventEmitter
const util = require( 'util' )
const rethinkdb = require( 'rethinkdb' )
const Connection = require( './connection' )
const TableManager = require( './table-manager' )
const dataTransform = require( './transform-data' )
const pckg = require( '../package.json' )
const PRIMARY_KEY = require( './primary-key')

class Connector extends EventEmitter {

  /**
   * Connects deepstream to a rethinkdb. RethinksDB is a great fit for deepstream due to its realtime capabilities.
   *
   * Similar to other storage connectors (e.g. MongoDB), this connector supports saving records to multiple tables.
   * In order to do this, specify a splitChar, e.g. '/' and use it in your record names. Naming your record
   *
   * user/i4vcg5j1-16n1qrnziuog
   *
   * for instance will create a user table and store it in it. This will allow for more sophisticated queries as
   * well as speed up read operations since there are less entries to look through
   *
   * @param {Object} options rethinkdb driver options. See rethinkdb.com/api/javascript/#connect
   *
   * e.g.
   *
   * {
   *     host: 'localhost',
   *     port: 28015,
   *     authKey: 'someString'
   *     database: 'deepstream',
   *     defaultTable: 'deepstream_records',
   *     splitChar: '/'
   * }
   *
   * Please note the three additional, optional keys:
   *
   * database   specifies which database to use. Defaults to 'deepstream'
   * defaultTable specifies which table records will be stored in that don't specify a table. Defaults to deepstream_records
   * splitChar   specifies a character that separates the record's id from the table it should be stored in. defaults to null
   *
   * @constructor
   */
  constructor( options ) {
    super()
    this.isReady = false
    this.name = pckg.name
    this.version = pckg.version
    this._checkOptions( options )
    this._options = options
    this._connection = new Connection( options, this._onConnection.bind( this ) )
    this._tableManager = new TableManager( this._connection )
    this._defaultTable = options.defaultTable || 'deepstream_records'
    this._splitChar = options.splitChar || null
    this._primaryKey = options.primaryKey || PRIMARY_KEY
  }

  /**
   * Writes a value to the database. If the specified table doesn't exist yet, it will be created
   * before the write is excecuted. If a table creation is already in progress, create table will
   * only add the method to its array of callbacks
   *
   * @param {String}   key
   * @param {Object}   value
   * @param {Function} callback Will be called with null for successful set operations or with an error message string
   *
   * @public
   * @returns {void}
   */
  set( key, value, callback ) {
    const params = this._getParams( key )
    const entry  = dataTransform.transformValueForStorage( value )
    const insert = this._insert.bind( this, params, entry, callback )

    if( this._tableManager.hasTable( params.table ) ) {
      insert()
    } else {
      this._tableManager.createTable( params.table, this._primaryKey, insert )
    }
  }

  /**
   * Retrieves a value from the cache
   *
   * @param {String}   key
   * @param {Function} callback Will be called with null and the stored object
   *                            for successful operations or with an error message string
   *
   * @public
   * @returns {void}
   */
  get( key, callback ) {
    const params = this._getParams( key )

    if( this._tableManager.hasTable( params.table ) ) {
      rethinkdb.table( params.table ).get( params.id ).run( this._connection.get(), ( error, entry ) => {
        if( entry ) {
          delete entry[ this._primaryKey ]
          entry = dataTransform.transformValueFromStorage( entry )
        }
        callback( error, entry )
      } )
    } else {
      callback( null, null )
    }
  }

  /**
   * Deletes an entry from the cache.
   *
   * @param   {String}   key
   * @param   {Function} callback Will be called with null for successful deletions or with
   *                     an error message string
   *
   * @public
   * @returns {void}
   */
  delete( key, callback ) {
    const params = this._getParams( key )

    if( this._tableManager.hasTable( params.table ) ) {
      rethinkdb.table( params.table ).get( params.id ).delete().run( this._connection.get(), callback )
    } else {
      callback( new Error( 'Table \'' + params.table + '\' does not exist' ) )
    }
  }

  /**
   * Callback for established connections
   *
   * @param   {Error} error
   *
   * @private
   * @returns {void}
   */
  _onConnection( error ) {
    if( error ) {
      this.emit( 'error', error )
    } else {
      this._tableManager.refreshTables( () => {
        this.isReady = true
        this.emit( 'ready' )
      } )
    }
  }

  /**
   * Parses the provided record name and returns an object
   * containing a table name and a record name
   *
   * @param   {String} key the name of the record
   *
   * @private
   * @returns {Object} params
   */
  _getParams( key ) {
    const parts = key.split( this._splitChar )

    if( parts.length === 2 ) {
      return { table: parts[ 0 ], id: parts[ 1 ] }
    } else {
      return { table: this._defaultTable, id: key }
    }
  }

  /**
   * Augments a value with a primary key and writes it to the database
   *
   * @param   {Object} params Map in the format { table: String, id: String }
   * @param   {Object} value The value that will be written
   * @param   {Function} callback called with error or null
   *
   * @private
   * @returns {void}
   */
  _insert( params, value, callback ) {
    value[ this._primaryKey ] = params.id

    rethinkdb
      .table( params.table )
      .insert( value, { returnChanges: false, conflict: 'replace' } )
      .run( this._connection.get(), callback )
  }

  /**
   * Makes sure that the options object contains all mandatory
   * settings
   *
   * @param   {Object} options
   *
   * @private
   * @returns {void}
   */
  _checkOptions( options ) {
    if( typeof options.host !== 'string' ) {
      throw new Error( 'Missing option host' )
    }
    if( typeof options.port !== 'number' ) {
      throw new Error( 'Missing option port' )
    }
  }
 }
module.exports = Connector<commit_msg>Use isNaN instead of typeof<commit_after>"use strict"

const EventEmitter = require( 'events' ).EventEmitter
const util = require( 'util' )
const rethinkdb = require( 'rethinkdb' )
const Connection = require( './connection' )
const TableManager = require( './table-manager' )
const dataTransform = require( './transform-data' )
const pckg = require( '../package.json' )
const PRIMARY_KEY = require( './primary-key')

class Connector extends EventEmitter {

  /**
   * Connects deepstream to a rethinkdb. RethinksDB is a great fit for deepstream due to its realtime capabilities.
   *
   * Similar to other storage connectors (e.g. MongoDB), this connector supports saving records to multiple tables.
   * In order to do this, specify a splitChar, e.g. '/' and use it in your record names. Naming your record
   *
   * user/i4vcg5j1-16n1qrnziuog
   *
   * for instance will create a user table and store it in it. This will allow for more sophisticated queries as
   * well as speed up read operations since there are less entries to look through
   *
   * @param {Object} options rethinkdb driver options. See rethinkdb.com/api/javascript/#connect
   *
   * e.g.
   *
   * {
   *     host: 'localhost',
   *     port: 28015,
   *     authKey: 'someString'
   *     database: 'deepstream',
   *     defaultTable: 'deepstream_records',
   *     splitChar: '/'
   * }
   *
   * Please note the three additional, optional keys:
   *
   * database   specifies which database to use. Defaults to 'deepstream'
   * defaultTable specifies which table records will be stored in that don't specify a table. Defaults to deepstream_records
   * splitChar   specifies a character that separates the record's id from the table it should be stored in. defaults to null
   *
   * @constructor
   */
  constructor( options ) {
    super()
    this.isReady = false
    this.name = pckg.name
    this.version = pckg.version
    this._checkOptions( options )
    this._options = options
    this._connection = new Connection( options, this._onConnection.bind( this ) )
    this._tableManager = new TableManager( this._connection )
    this._defaultTable = options.defaultTable || 'deepstream_records'
    this._splitChar = options.splitChar || null
    this._primaryKey = options.primaryKey || PRIMARY_KEY
  }

  /**
   * Writes a value to the database. If the specified table doesn't exist yet, it will be created
   * before the write is excecuted. If a table creation is already in progress, create table will
   * only add the method to its array of callbacks
   *
   * @param {String}   key
   * @param {Object}   value
   * @param {Function} callback Will be called with null for successful set operations or with an error message string
   *
   * @public
   * @returns {void}
   */
  set( key, value, callback ) {
    const params = this._getParams( key )
    const entry  = dataTransform.transformValueForStorage( value )
    const insert = this._insert.bind( this, params, entry, callback )

    if( this._tableManager.hasTable( params.table ) ) {
      insert()
    } else {
      this._tableManager.createTable( params.table, this._primaryKey, insert )
    }
  }

  /**
   * Retrieves a value from the cache
   *
   * @param {String}   key
   * @param {Function} callback Will be called with null and the stored object
   *                            for successful operations or with an error message string
   *
   * @public
   * @returns {void}
   */
  get( key, callback ) {
    const params = this._getParams( key )

    if( this._tableManager.hasTable( params.table ) ) {
      rethinkdb.table( params.table ).get( params.id ).run( this._connection.get(), ( error, entry ) => {
        if( entry ) {
          delete entry[ this._primaryKey ]
          entry = dataTransform.transformValueFromStorage( entry )
        }
        callback( error, entry )
      } )
    } else {
      callback( null, null )
    }
  }

  /**
   * Deletes an entry from the cache.
   *
   * @param   {String}   key
   * @param   {Function} callback Will be called with null for successful deletions or with
   *                     an error message string
   *
   * @public
   * @returns {void}
   */
  delete( key, callback ) {
    const params = this._getParams( key )

    if( this._tableManager.hasTable( params.table ) ) {
      rethinkdb.table( params.table ).get( params.id ).delete().run( this._connection.get(), callback )
    } else {
      callback( new Error( 'Table \'' + params.table + '\' does not exist' ) )
    }
  }

  /**
   * Callback for established connections
   *
   * @param   {Error} error
   *
   * @private
   * @returns {void}
   */
  _onConnection( error ) {
    if( error ) {
      this.emit( 'error', error )
    } else {
      this._tableManager.refreshTables( () => {
        this.isReady = true
        this.emit( 'ready' )
      } )
    }
  }

  /**
   * Parses the provided record name and returns an object
   * containing a table name and a record name
   *
   * @param   {String} key the name of the record
   *
   * @private
   * @returns {Object} params
   */
  _getParams( key ) {
    const parts = key.split( this._splitChar )

    if( parts.length === 2 ) {
      return { table: parts[ 0 ], id: parts[ 1 ] }
    } else {
      return { table: this._defaultTable, id: key }
    }
  }

  /**
   * Augments a value with a primary key and writes it to the database
   *
   * @param   {Object} params Map in the format { table: String, id: String }
   * @param   {Object} value The value that will be written
   * @param   {Function} callback called with error or null
   *
   * @private
   * @returns {void}
   */
  _insert( params, value, callback ) {
    value[ this._primaryKey ] = params.id

    rethinkdb
      .table( params.table )
      .insert( value, { returnChanges: false, conflict: 'replace' } )
      .run( this._connection.get(), callback )
  }

  /**
   * Makes sure that the options object contains all mandatory
   * settings
   *
   * @param   {Object} options
   *
   * @private
   * @returns {void}
   */
  _checkOptions( options ) {
    if( typeof options.host !== 'string' ) {
      throw new Error( 'Missing option host' )
    }
    if( isNaN( options.port ) ) {
      throw new Error( 'Missing option port' )
    }
  }
 }
module.exports = Connector
{'fraction_non_alphanumeric': 0.10628825820812465, 'fraction_numerical': 0.0022259321090706734, 'mean_word_length': 2.5476593344613647, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '2832645', 'n_tokens_mistral': 3749, 'n_tokens_neox': 3684, 'n_words': 1254}