<commit_before>err = require('./errors')
util = require('./util')

protoResponseType = require('./proto-def').Response.ResponseType
Promise = require('bluebird')
EventEmitter = require('events').EventEmitter

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

# setImmediate is not defined in some browsers (including Chrome)
if not setImmediate?
    setImmediate = (cb) ->
        setTimeout cb, 0

class IterableResult
    stackSize: 100

    constructor: (conn, token, opts, root) ->
        @_conn = conn
        @_token = token
        @_opts = opts
        @_root = root # current query

        @_responses = []
        @_responseIndex = 0
        @_outstandingRequests = 1 # Because we haven't add the response yet
        @_iterations = 0
        @_endFlag = false
        @_contFlag = false
        @_closeAsap = false
        @_cont = null
        @_cbQueue = []

        @next = @_next
        @each = @_each

    _addResponse: (response) ->
        if response.t is @_type or response.t is protoResponseType.SUCCESS_SEQUENCE
            # We push a "ok" response only if it's not empty
            if response.r.length > 0
                @_responses.push response
        else
            @_responses.push response


        @_outstandingRequests -= 1
        if response.t isnt @_type
            # We got an error or a SUCCESS_SEQUENCE
            @_endFlag = true

            if @_closeCb?
                switch response.t
                    when protoResponseType.COMPILE_ERROR
                        @_closeCb mkErr(err.RqlRuntimeError, response, @_root)
                    when protoResponseType.CLIENT_ERROR
                        @_closeCb mkErr(err.RqlRuntimeError, response, @_root)
                    when protoResponseType.RUNTIME_ERROR
                        @_closeCb mkErr(err.RqlRuntimeError, response, @_root)
                    else
                        @_closeCb()

        @_contFlag = false
        if @_closeAsap is false
            @_promptNext()
        else
            @close @_closeCb
        @

    _getCallback: ->
        @_iterations += 1
        cb = @_cbQueue.shift()

        if @_iterations % @stackSize is @stackSize - 1
            immediateCb = ((err, row) -> setImmediate -> cb(err, row))
            return immediateCb
        else
            return cb

    _handleRow: ->
        response = @_responses[0]
        row = util.recursivelyConvertPseudotype(response.r[@_responseIndex], @_opts)
        cb = @_getCallback()

        @_responseIndex += 1

        # If we're done with this response, discard it
        if @_responseIndex is response.r.length
            @_responses.shift()
            @_responseIndex = 0

        cb null, row

    bufferEmpty: ->
        @_responses.length is 0 or @_responses[0].r.length <= @_responseIndex

    _promptNext: ->
        # If there are no more waiting callbacks, just wait until the next event
        while @_cbQueue[0]?
            if @bufferEmpty() is true
                # We prefetch things here, set `is 0` to avoid prefectch
                if @_endFlag is true
                    cb = @_getCallback()
                    cb new err.RqlDriverError "No more rows in the cursor."
                else if @_responses.length <= 1
                    @_promptCont()

                return
            else

                # Try to get a row out of the responses
                response = @_responses[0]

                if @_responses.length is 1
                    # We're low on data, prebuffer
                    @_promptCont()

                # Error responses are not discarded, and the error will be sent to all future callbacks
                switch response.t
                    when protoResponseType.SUCCESS_PARTIAL
                        @_handleRow()
                    when protoResponseType.SUCCESS_SEQUENCE
                        if response.r.length is 0
                            @_responses.shift()
                        else
                            @_handleRow()
                    when protoResponseType.COMPILE_ERROR
                        @_responses.shift()
                        cb = @_getCallback()
                        cb mkErr(err.RqlCompileError, response, @_root)
                    when protoResponseType.CLIENT_ERROR
                        @_responses.shift()
                        cb = @_getCallback()
                        cb mkErr(err.RqlClientError, response, @_root)
                    when protoResponseType.RUNTIME_ERROR
                        @_responses.shift()
                        cb = @_getCallback()
                        cb mkErr(err.RqlRuntimeError, response, @_root)
                    else
                        @_responses.shift()
                        cb = @_getCallback()
                        cb new err.RqlDriverError "Unknown response type for cursor"

    _promptCont: ->
        # Let's ask the server for more data if we haven't already
        if (not @_contFlag) and (not @_endFlag) and @_conn.isOpen()
            @_contFlag = true
            @_outstandingRequests += 1
            @_conn._continueQuery(@_token)


    ## Implement IterableResult
    hasNext: ->
        throw new err.RqlDriverError "The `hasNext` command has been removed since 1.13. Use `next` instead."

    _next: varar 0, 1, (cb) ->
        fn = (cb) =>
            @_cbQueue.push cb
            @_promptNext()

        if typeof cb is "function"
            fn(cb)
        else if cb is undefined
            p = new Promise (resolve, reject) ->
                cb = (err, result) ->
                    if (err)
                        reject(err)
                    else
                        resolve(result)
                fn(cb)
            return p
        else
            throw new err.RqlDriverError "First argument to `next` must be a function or undefined."


    close: varar 0, 1, (cb) ->
        new Promise( (resolve, reject) =>
            if @_endFlag is true
                resolve()
            else if not @_closeCb?
                @_closeCb = (err) ->
                    # Clear all callbacks for outstanding requests
                    while @_cbQueue.length > 0
                        @_cbQueue.shift()
                    # The connection uses _outstandingRequests to see
                    # if it should remove the token for this
                    # cursor. This states unambiguously that we don't
                    # care whatever responses return now.
                    @_outstandingRequests = 0
                    if (err)
                        reject(err)
                    else
                        resolve()
                @_closeAsap = true
                @_outstandingRequests += 1
                @_conn._endQuery(@_token)
            else
                @emit 'error', new err.RqlDriverError "This shouldn't happen"
        ).nodeify cb

    _each: varar(1, 2, (cb, onFinished) -><commit_msg>Fix scoping issue in cursor.coffee<commit_after>err = require('./errors')
util = require('./util')

protoResponseType = require('./proto-def').Response.ResponseType
Promise = require('bluebird')
EventEmitter = require('events').EventEmitter

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

# setImmediate is not defined in some browsers (including Chrome)
if not setImmediate?
    setImmediate = (cb) ->
        setTimeout cb, 0

class IterableResult
    stackSize: 100

    constructor: (conn, token, opts, root) ->
        @_conn = conn
        @_token = token
        @_opts = opts
        @_root = root # current query

        @_responses = []
        @_responseIndex = 0
        @_outstandingRequests = 1 # Because we haven't add the response yet
        @_iterations = 0
        @_endFlag = false
        @_contFlag = false
        @_closeAsap = false
        @_cont = null
        @_cbQueue = []

        @next = @_next
        @each = @_each

    _addResponse: (response) ->
        if response.t is @_type or response.t is protoResponseType.SUCCESS_SEQUENCE
            # We push a "ok" response only if it's not empty
            if response.r.length > 0
                @_responses.push response
        else
            @_responses.push response


        @_outstandingRequests -= 1
        if response.t isnt @_type
            # We got an error or a SUCCESS_SEQUENCE
            @_endFlag = true

            if @_closeCb?
                switch response.t
                    when protoResponseType.COMPILE_ERROR
                        @_closeCb mkErr(err.RqlRuntimeError, response, @_root)
                    when protoResponseType.CLIENT_ERROR
                        @_closeCb mkErr(err.RqlRuntimeError, response, @_root)
                    when protoResponseType.RUNTIME_ERROR
                        @_closeCb mkErr(err.RqlRuntimeError, response, @_root)
                    else
                        @_closeCb()

        @_contFlag = false
        if @_closeAsap is false
            @_promptNext()
        else
            @close @_closeCb
        @

    _getCallback: ->
        @_iterations += 1
        cb = @_cbQueue.shift()

        if @_iterations % @stackSize is @stackSize - 1
            immediateCb = ((err, row) -> setImmediate -> cb(err, row))
            return immediateCb
        else
            return cb

    _handleRow: ->
        response = @_responses[0]
        row = util.recursivelyConvertPseudotype(response.r[@_responseIndex], @_opts)
        cb = @_getCallback()

        @_responseIndex += 1

        # If we're done with this response, discard it
        if @_responseIndex is response.r.length
            @_responses.shift()
            @_responseIndex = 0

        cb null, row

    bufferEmpty: ->
        @_responses.length is 0 or @_responses[0].r.length <= @_responseIndex

    _promptNext: ->
        # If there are no more waiting callbacks, just wait until the next event
        while @_cbQueue[0]?
            if @bufferEmpty() is true
                # We prefetch things here, set `is 0` to avoid prefectch
                if @_endFlag is true
                    cb = @_getCallback()
                    cb new err.RqlDriverError "No more rows in the cursor."
                else if @_responses.length <= 1
                    @_promptCont()

                return
            else

                # Try to get a row out of the responses
                response = @_responses[0]

                if @_responses.length is 1
                    # We're low on data, prebuffer
                    @_promptCont()

                # Error responses are not discarded, and the error will be sent to all future callbacks
                switch response.t
                    when protoResponseType.SUCCESS_PARTIAL
                        @_handleRow()
                    when protoResponseType.SUCCESS_SEQUENCE
                        if response.r.length is 0
                            @_responses.shift()
                        else
                            @_handleRow()
                    when protoResponseType.COMPILE_ERROR
                        @_responses.shift()
                        cb = @_getCallback()
                        cb mkErr(err.RqlCompileError, response, @_root)
                    when protoResponseType.CLIENT_ERROR
                        @_responses.shift()
                        cb = @_getCallback()
                        cb mkErr(err.RqlClientError, response, @_root)
                    when protoResponseType.RUNTIME_ERROR
                        @_responses.shift()
                        cb = @_getCallback()
                        cb mkErr(err.RqlRuntimeError, response, @_root)
                    else
                        @_responses.shift()
                        cb = @_getCallback()
                        cb new err.RqlDriverError "Unknown response type for cursor"

    _promptCont: ->
        # Let's ask the server for more data if we haven't already
        if (not @_contFlag) and (not @_endFlag) and @_conn.isOpen()
            @_contFlag = true
            @_outstandingRequests += 1
            @_conn._continueQuery(@_token)


    ## Implement IterableResult
    hasNext: ->
        throw new err.RqlDriverError "The `hasNext` command has been removed since 1.13. Use `next` instead."

    _next: varar 0, 1, (cb) ->
        fn = (cb) =>
            @_cbQueue.push cb
            @_promptNext()

        if typeof cb is "function"
            fn(cb)
        else if cb is undefined
            p = new Promise (resolve, reject) ->
                cb = (err, result) ->
                    if (err)
                        reject(err)
                    else
                        resolve(result)
                fn(cb)
            return p
        else
            throw new err.RqlDriverError "First argument to `next` must be a function or undefined."


    close: varar 0, 1, (cb) ->
        new Promise( (resolve, reject) =>
            if @_endFlag is true
                resolve()
            else if not @_closeCb?
                @_closeCb = (err) =>
                    # Clear all callbacks for outstanding requests
                    while @_cbQueue.length > 0
                        @_cbQueue.shift()
                    # The connection uses _outstandingRequests to see
                    # if it should remove the token for this
                    # cursor. This states unambiguously that we don't
                    # care whatever responses return now.
                    @_outstandingRequests = 0
                    if (err)
                        reject(err)
                    else
                        resolve()
                @_closeAsap = true
                @_outstandingRequests += 1
                @_conn._endQuery(@_token)
            else
                @emit 'error', new err.RqlDriverError "This shouldn't happen"
        ).nodeify cb

    _each: varar(1, 2, (cb, onFinished) ->
{'fraction_non_alphanumeric': 0.07579740001420757, 'fraction_numerical': 0.004972650422675286, 'mean_word_length': 1.364063811922754, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 45, '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': '5503406', 'n_tokens_mistral': 3800, 'n_tokens_neox': 3532, 'n_words': 1084}