<commit_before>net = require('net')
events = require('events')

util = require('./util')
err = require('./errors')
cursors = require('./cursor')

protodef = require('./proto-def')
protoVersion = protodef.VersionDummy.Version.V0_3
protoProtocol = protodef.VersionDummy.Protocol.JSON
protoQueryType = protodef.Query.QueryType
protoResponseType = protodef.Response.ResponseType

r = require('./ast')
Promise = require('bluebird')

# Import some names to this namespace for convienience
ar = util.ar
varar = util.varar
aropt = util.aropt
mkAtom = util.mkAtom
mkErr = util.mkErr

class Connection extends events.EventEmitter
    DEFAULT_HOST: 'localhost'
    DEFAULT_PORT: 28015
    DEFAULT_AUTH_KEY: ''
    DEFAULT_TIMEOUT: 20 # In seconds

    constructor: (host, callback) ->
        if typeof host is 'undefined'
            host = {}
        else if typeof host is 'string'
            host = {host: host}

        @host = host.host || @DEFAULT_HOST
        @port = host.port || @DEFAULT_PORT
        @db = host.db # left undefined if this is not set
        @authKey = host.authKey || @DEFAULT_AUTH_KEY
        @timeout = host.timeout || @DEFAULT_TIMEOUT

        @outstandingCallbacks = {}
        @nextToken = 1
        @open = false

        @buffer = new Buffer(0)

        @_events = @_events || {}

        errCallback = (e) =>
            @removeListener 'connect', conCallback
            if e instanceof err.RqlDriverError
                callback e
            else
                callback new err.RqlDriverError "Could not connect to #{@host}:#{@port}.\n#{e.message}"
        @once 'error', errCallback

        conCallback = =>
            @removeListener 'error', errCallback
            @open = true
            callback null, @
        @once 'connect', conCallback


    _data: (buf) ->
        # Buffer data, execute return results if need be
        @buffer = Buffer.concat([@buffer, buf])

        while @buffer.length >= 12 
            token = @buffer.readUInt32LE(0) + 0x100000000 * @buffer.readUInt32LE(4)

            responseLength = @buffer.readUInt32LE(8)
            unless @buffer.length >= (12 + responseLength)
                break

            responseBuffer = @buffer.slice(12, responseLength + 12)
            response = JSON.parse(responseBuffer)

            @_processResponse response, token
            @buffer = @buffer.slice(12 + responseLength)

    _delQuery: (token) ->
        # This query is done, delete this cursor
        delete @outstandingCallbacks[token]

        if Object.keys(@outstandingCallbacks).length < 1 and not @open
            @cancel()

    _processResponse: (response, token) ->
        profile = response.p
        if @outstandingCallbacks[token]?
            {cb:cb, root:root, cursor: cursor, opts: opts, feed: feed} = @outstandingCallbacks[token]
            if cursor?
                cursor._addResponse(response)

                if cursor._endFlag && cursor._outstandingRequests is 0
                    @_delQuery(token)

            else if cb?
                # Behavior varies considerably based on response type
                switch response.t
                    when protoResponseType.COMPILE_ERROR
                        cb mkErr(err.RqlCompileError, response, root)
                        @_delQuery(token)
                    when protoResponseType.CLIENT_ERROR
                        cb mkErr(err.RqlClientError, response, root)
                        @_delQuery(token)
                    when protoResponseType.RUNTIME_ERROR
                        cb mkErr(err.RqlRuntimeError, response, root)
                        @_delQuery(token)
                    when protoResponseType.SUCCESS_ATOM
                        response = mkAtom response, opts
                        if Array.isArray response
                            response = cursors.makeIterable response<commit_msg>Callback should be called once<commit_after>net = require('net')
events = require('events')

util = require('./util')
err = require('./errors')
cursors = require('./cursor')

protodef = require('./proto-def')
protoVersion = protodef.VersionDummy.Version.V0_3
protoProtocol = protodef.VersionDummy.Protocol.JSON
protoQueryType = protodef.Query.QueryType
protoResponseType = protodef.Response.ResponseType

r = require('./ast')
Promise = require('bluebird')

# Import some names to this namespace for convienience
ar = util.ar
varar = util.varar
aropt = util.aropt
mkAtom = util.mkAtom
mkErr = util.mkErr

class Connection extends events.EventEmitter
    DEFAULT_HOST: 'localhost'
    DEFAULT_PORT: 28015
    DEFAULT_AUTH_KEY: ''
    DEFAULT_TIMEOUT: 20 # In seconds

    constructor: (host, callback) ->
        if typeof host is 'undefined'
            host = {}
        else if typeof host is 'string'
            host = {host: host}

        @host = host.host || @DEFAULT_HOST
        @port = host.port || @DEFAULT_PORT
        @db = host.db # left undefined if this is not set
        @authKey = host.authKey || @DEFAULT_AUTH_KEY
        @timeout = host.timeout || @DEFAULT_TIMEOUT

        @outstandingCallbacks = {}
        @nextToken = 1
        @open = false

        @buffer = new Buffer(0)

        @_events = @_events || {}

        errCallback = (e) =>
            @removeListener 'connect', conCallback
            if e instanceof err.RqlDriverError
                callback e
            else
                callback new err.RqlDriverError "Could not connect to #{@host}:#{@port}.\n#{e.message}"
        @once 'error', errCallback

        conCallback = =>
            @removeListener 'error', errCallback
            @open = true
            callback null, @
        @once 'connect', conCallback


    _data: (buf) ->
        # Buffer data, execute return results if need be
        @buffer = Buffer.concat([@buffer, buf])

        while @buffer.length >= 12 
            token = @buffer.readUInt32LE(0) + 0x100000000 * @buffer.readUInt32LE(4)

            responseLength = @buffer.readUInt32LE(8)
            unless @buffer.length >= (12 + responseLength)
                break

            responseBuffer = @buffer.slice(12, responseLength + 12)
            response = JSON.parse(responseBuffer)

            @_processResponse response, token
            @buffer = @buffer.slice(12 + responseLength)

    _delQuery: (token) ->
        # This query is done, delete this cursor
        delete @outstandingCallbacks[token]

        if Object.keys(@outstandingCallbacks).length < 1 and not @open
            @cancel()

    _processResponse: (response, token) ->
        profile = response.p
        if @outstandingCallbacks[token]?
            {cb:cb, root:root, cursor: cursor, opts: opts, feed: feed} = @outstandingCallbacks[token]
            if cursor?
                cursor._addResponse(response)

                if cursor._endFlag && cursor._outstandingRequests is 0
                    @_delQuery(token)
            else if feed?
                feed._addResponse(response)

                if feed._endFlag && feed._outstandingRequests is 0
                    @_delQuery(token)
            else if cb?
                # Behavior varies considerably based on response type
                switch response.t
                    when protoResponseType.COMPILE_ERROR
                        cb mkErr(err.RqlCompileError, response, root)
                        @_delQuery(token)
                    when protoResponseType.CLIENT_ERROR
                        cb mkErr(err.RqlClientError, response, root)
                        @_delQuery(token)
                    when protoResponseType.RUNTIME_ERROR
                        cb mkErr(err.RqlRuntimeError, response, root)
                        @_delQuery(token)
                    when protoResponseType.SUCCESS_ATOM
                        response = mkAtom response, opts
                        if Array.isArray response
                            response = cursors.makeIterable response
{'fraction_non_alphanumeric': 0.09130982367758186, 'fraction_numerical': 0.010705289672544081, 'mean_word_length': 1.9356746765249537, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, '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': '3840640', 'n_tokens_mistral': 2160, 'n_tokens_neox': 2032, 'n_words': 543}