net.coffee

# # Network (net.coffee)
#
# This module handles network and protocol related functionality for
# the driver. The classes defined here are:
#
# - `Connection`, which is an EventEmitter and the base class for
# - `TcpConnection`, the standard driver connection
# - `HttpConnection`, the connection type used by the webui
#

# ### Imports
#
# The [net module](http://nodejs.org/api/net.html) is the low-level
# networking library Node.js provides. We need it for `TcpConnection`
# objects
net = require('net')

# The [events module](http://nodejs.org/api/events.html) is a core
# Node.js module that provides the ability for objects to emit events,
# and for callbacks to be attached to those events.
events = require('events')

# The [util module](util.html) contains utility methods used several
# places in the driver
util = require('./util')

# The [errors module](errors.html) contains exceptions thrown by the driver
err = require('./errors')

# The [cursors module](cursors.html) contains data structures used to
# iterate over large result sets or infinite streams of data from the
# database (changefeeds).
cursors = require('./cursor')

# The `proto-def` module is an autogenerated file full of protocol
# definitions used to communicate to the RethinkDB server (basically a
# ton of human readable names for magic numbers).
#
# It is created by the python script `convert_protofile` in the
# `drivers` directory (one level up from this file) by converting a
# protocol buffers definition file into a JavaScript file. To generate
# the `proto-def.js` file, see the instructions in the
# [README](./index.html#creating-the-proto-defjs)
#
# Note that it's a plain JavaScript file, not a CoffeeScript file.
protodef = require('./proto-def')

# Each version of the protocol has a magic number specified in
# `./proto-def.coffee`. The most recent version is 4. Generally the
# official driver will always be updated to the newest version of the
# protocol, though RethinkDB supports older versions for some time.
protoVersion = protodef.VersionDummy.Version.V0_4

# We are using the JSON protocol for RethinkDB, which is the most
# recent version. The older protocol is based on Protocol Buffers, and
# is deprecated.
protoProtocol = protodef.VersionDummy.Protocol.JSON

# The `QueryType` definitions are used to control at a high level how
# we interact with the server. So we can `START` a new query, `STOP`
# an executing query, `CONTINUE` receiving results from an existing
# cursor, or wait for all outstanding `noreply` queries to finish with
# `NOREPLY_WAIT`.
protoQueryType = protodef.Query.QueryType

# The server can respond to queries in several ways. These are the
# definitions for the response types.
protoResponseType = protodef.Response.ResponseType

# The [ast module](ast.html) contains the bulk of the api exposed by
# the driver. It defines how you can create ReQL queries, and handles
# serializing those queries into JSON to be transmitted over the wire
# to the database server.
r = require('./ast')

# We use the [bluebird](https://github.com/petkaantonov/bluebird)
# third-party module to provide a promise implementation for the
# driver.
Promise = require('bluebird')

# These are some functions we import directly from the `util` module
# for convenience.
ar = util.ar
varar = util.varar
aropt = util.aropt
mkAtom = util.mkAtom
mkErr = util.mkErr

# ### Connection
#
# Connection is the base class for both `TcpConnection` and
# `HttpConnection`. Applications using this driver will need to get a
# connection object to be able to query the server.
#
# Connection is a subclass of
# [EventEmitter](http://nodejs.org/api/events.html#events_class_events_eventemitter),
# and it will emit the following events:
#
# - `"connect"` is emitted by the subclasses `TcpConnection` and
#   `HttpConnection` when they connect to the server successfully.
# - `"error"` is emitted when protocol level errors occur (notably
#    not query errors! Those are returned as arguments to the
#    callback the user provides.) `"error"` will also be accompanied
#    by a message indicating what went wrong.
# - `"close"` is emitted when the connection is closed either through
#    an error or by the user calling `connection.close()` explicitly
# - `"timeout"` will be emitted by the `TcpConnection` subclass if the
#    underlying socket times out for any reason.
class Connection extends events.EventEmitter

    # These are the default hostname and port used by RethinkDB
    DEFAULT_HOST: 'localhost'
    DEFAULT_PORT: 28015
    # By default, RethinkDB doesn't use an authorization key.
    DEFAULT_AUTH_KEY: ''
    # Each connection has a timeout (in seconds) for the initial handshake with the
    # server. Note that this is not a timeout for queries to return
    # results.
    DEFAULT_TIMEOUT: 20 # In seconds

    # #### Connection constructor
    constructor: (host, callback) ->
        # We need to set the defaults if the user hasn't supplied anything.
        if typeof host is 'undefined'
            host = {}
        # It's really convenient to be able to pass just a hostname to
        # connect, since that's the thing that is most likely to vary
        # (default ports, no auth key, default connection timeout are
        # all common). We just detect that case and wrap it.
        else if typeof host is 'string'
            host = {host: host}

        # Here we set all of the connection parameters to their defaults.
        @host = host.host || @DEFAULT_HOST
        @port = host.port || @DEFAULT_PORT

        # One notable exception to defaulting is the db name. If the
        # user doesn't specify it, we leave it undefined. On the
        # server side, if no database is specified for a table, the
        # database will default to `"test"`.
        @db = host.db # left undefined if this is not set

        @authKey = host.authKey || @DEFAULT_AUTH_KEY
        @timeout = host.timeout || @DEFAULT_TIMEOUT

        # The protocol allows for responses to queries on the same
        # connection to be returned interleaved. When a query is run
        # with this connection, an entry is added to
        # `@outstandingCallbacks`. The key is the query token, and the
        # value is an object with the following fields:
        #
        # - **cb**: The callback to invoke when a query returns another
        #   batch of results
        # - **root**: a subclass of `TermBase` (defined in the
        #   [ast module](ast.html)) representing the query being run.
        # - **opts**: global options passed to `.run`.
        #
        # Once the server returns a response, one of two fields may be
        # added to the entry for that query depending on what comes
        # back (this is done in `_processResponse`):
        #
        # - **cursor**: is set to a `Cursor` object (defined in the
        #   [cursor module](cursor.html)) if the server
        #   replies with `SUCCESS_PARTIAL`. This happens when a
        #   result is too large to send back all in one batch. The
        #   `Cursor` allows fetching results lazily as they're needed.
        # - **feed**: is set to a `Feed` object (defined in the
        #   [cursor module](cursor.html)). This is very similar to a `Cursor`,
        #   except that it is potentially infinite. Changefeeds are a way
        #   for the server to notify the client when a change occurs to
        #   the results of a query the user wants to watch.
        #
        # Any other responses are considered "done" and don't have any
        # further results to fetch from the server. At that time the
        # query token is deleted from `@outstandingCallbacks` since we
        # don't expect any more results using that token.
        @outstandingCallbacks = {}

        # Each query to the server requires a unique token (per
        # connection). `nextToken` is incremented every time we issue
        # a new query
        @nextToken = 1

        # `@open` and `@closing` are used to record changes in the
        # connection state. A connection is open after the handshake
        # is successfully completed, and it becomes closed after the
        # server confirms the connection is closed.
        @open = false
        # `@closing` is true when the `conn.close()` method is called,
        # and becomes false again just before the callback provided to
        # `close` is called. This allows `noreplyWait` to be called
        # from within `.close`, but prevents any other queries from
        # being run on a closing connection. (`noreplyWait` checks
        # `@open` only to see if the connection is closed, but the
        # `.isOpen()` method checks both flags)
        @closing = false

        # We create a [Buffer](http://nodejs.org/api/buffer.html)
        # object to receive all bytes coming in on this
        # connection. The buffer is modified in the `_data` method
        # (which is called whenever data comes in over the network on
        # this connection), and it is also modified by the handshake
        # callback that's defined in the `TcpConnection` constructor.
        @buffer = new Buffer(0)

        @_events = @_events || {}

        # Now we set up two callbacks, one to run on successful
        # connection, and the other to run if we fail to connect to
        # the server. They listen to the `"connect"` and `"error"`
        # events respectively. If successful, we set `@open` to
        # `true`, otherwise we pass an error to the connection
        # callback that was passed to the constructor (`callback`)
        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


    # #### Connection _data method
    #
    # This method is responsible for parsing responses from the server
    # after the initial handshake is over. It reads the token number
    # of the response (so we know which query is being responded to),
    # and the response length, then parses the rest of the response as
    # JSON.
    _data: (buf) ->
        # We extend the current buffer with the contents of `buf` that
        # we got from the server.
        @buffer = Buffer.concat([@buffer, buf])

        # The first 8 bytes in a response are the token number of the
        # query the server is responding to. The next 4 bytes indicate
        # how long the response will be. So if the response isn't at
        # least 12 bytes long, there's nothing else to do.
        while @buffer.length >= 12
            # When a query is sent to the server, we write out the
            # token in a way that can be read back later as a native
            # integer. Since Node buffers don't have a method like
            # `readUInt64LE`, we're forced to emulate it ourselves.
            token = @buffer.readUInt32LE(0) + 0x100000000 * @buffer.readUInt32LE(4)

            # Next up is the response length. The protocol dictates
            # this must be a 32 bit unsigned integer in little endian
            # byte order.
            responseLength = @buffer.readUInt32LE(8)
            # This ensures that the responseLength is less than or
            # equal to the amount of data we have in the buffer
            # (including the token and response length itself). If the
            # buffer doesn't have enough data in it, we just break
            # out. More data will be coming later, and it will be
            # added to the end of the buffer, so we'll wait until the
            # full response is available.
            unless @buffer.length >= (12 + responseLength)
                break

            # Since we have enough data, we slice it out of the
            # buffer, and parse it as JSON.
            responseBuffer = @buffer.slice(12, responseLength + 12)
            response = JSON.parse(responseBuffer)

            # Now it's off to `_processResponse` where the data is
            # converted to a format the user will be able to work
            # with, error responses are detected etc.
            @_processResponse response, token

            # Finally, we truncate the buffer to the end of the
            # current response. The buffer may already be queuing up a
            # new response, so it's never safe to clear it or create a
            # new one.
            @buffer = @buffer.slice(12 + responseLength)

    # #### Connection _delQuery method
    #
    # This method just removes the entry in `@outstandingCallbacks`
    # for a given token. It's called whenever a response doesn't
    # return a cursor, a cursor is completely consumed, or the query
    # encounters an error.
    _delQuery: (token) ->
        delete @outstandingCallbacks[token]

    # #### Connection _processResponse method
    #
    # This method is contains the main logic for taking different
    # actions based on the response type. It receives the response as
    # an object (parsed from the JSON coming over the wire), and the
    # token for the query the response corresponds to.
    _processResponse: (response, token) ->
        # For brevity the wire format specifies that the profile key
        # is "p". We give it a more readable variable name here.
        profile = response.p
        # First we need to check if we're even expecting a response
        # for this token. If not it's an error.
        if @outstandingCallbacks[token]?
            {cb:cb, root:root, cursor: cursor, opts: opts, feed: feed} = @outstandingCallbacks[token]
            # Some results for queries are not returned all at once,
            # but in chunks. The driver takes care of making sure the
            # user doesn't see this, and uses `Cursor`s to take
            # multiple responses and make them into one long stream of
            # query results.
            #
            # Depending on the type of result, and whether this is the
            # first response for this token, we may or may not have a
            # cursor defined for this token. If we do have a cursor
            # defined already, we add this response to the cursor.
            if cursor?
                cursor._addResponse(response)

                # `cursor._addResponse` will check if this response is
                # the last one for this token, and if so will set its
                # `cursor._endFlag` to true. If this is the last
                # response for this query and we aren't waiting on any
                # other responses for this cursor (if, for example, we
                # get the responses out of order), then we remove this
                # token's entry in `@outstandingCallbacks`.
                if cursor._endFlag && cursor._outstandingRequests is 0
                    @_delQuery(token)

            # Next we check if we have a callback registered for this
            # token. In [ast](ast.html) we always provide `_start`
            # with a wrapped callback function, so this may as well be
            # an else branch.
            else if cb?
                # The protocol (again for brevity) puts the response
                # type into a key called "t". What we do next depends
                # on that value. We'll be comparing it to the values
                # in the `proto-def` module, in `protoResponseType`.
                switch response.t
                    # ##### Error responses
                    when protoResponseType.COMPILE_ERROR
                        # Compile errors happen during parsing and
                        # type checking the query on the server
                        # side. An example is passing too many
                        # arguments to a function.  We pass an error
                        # object that includes the backtrace from the
                        # response and the original query
                        # (`root`). Then we delete the token from
                        # `@outstandingCallbacks`.
                        cb mkErr(err.RqlCompileError, response, root)
                        @_delQuery(token)
                    when protoResponseType.CLIENT_ERROR
                        # Client errors are returned when the client
                        # is buggy. This can happen if a query isn't
                        # serialized right, or the handshake is done
                        # incorrectly etc. Hopefully end users of the
                        # driver should never see these.
                        cb mkErr(err.RqlClientError, response, root)
                        @_delQuery(token)
                    when protoResponseType.RUNTIME_ERROR
                        # Runtime errors are the most common type of
                        # error. They happen when something goes wrong
                        # that can only be determined by running the
                        # query. For example, if you try to get the
                        # value of a field in an object that doesn't
                        # exist.
                        cb mkErr(err.RqlRuntimeError, response, root)
                        @_delQuery(token)
                    # ##### Success responses
                    when protoResponseType.SUCCESS_ATOM
                        # `SUCCESS_ATOM` is returned when the query
                        # was successful and returned a single ReQL
                        # data type. The `mkAtom` function from the
                        # [util module](util.html) converts all
                        # pseudotypes in this response to their
                        # corresponding native types
                        response = mkAtom response, opts
                        # If the response is an array, we patch it a
                        # bit so it can be used as a stream.
                        if Array.isArray response
                            response = cursors.makeIterable response
                        # If there's a profile available, we nest the
                        # response slightly differently before passing
                        # it to the callback for this token.
                        if profile?
                            response = {profile: profile, value: response}
                        cb null, response
                        # The `SUCCESS_ATOM` response means there will
                        # be no further results for this query, so we
                        # remove it from `@outstandingCallbacks`
                        @_delQuery(token)
                    when protoResponseType.SUCCESS_PARTIAL
                        # `SUCCESS_PARTIAL` indicates the client
                        # should create a cursor and request more data
                        # from the server when it's ready by sending a
                        # `CONTINUE` query with the same token. So, we
                        # create a new Cursor for this token, and add
                        # it to the object stored in
                        # `@outstandingCallbacks`.

                        # We create a new cursor, which is sometimes a
                        # `Feed`, `AtomFeed`, or `OrderByLimitFeed`
                        # depending on the `ResponseNotes`.  (This
                        # usually doesn't matter, but third-party ORMs
                        # might want to treat these differently.)
                        cursor = null
                        for note in response.n
                            switch note
                                when protodef.Response.ResponseNote.SEQUENCE_FEED
                                    cursor ?= new cursors.Feed @, token, opts, root
                                when protodef.Response.ResponseNote.UNIONED_FEED
                                    cursor ?= new cursors.UnionedFeed @, token, opts, root
                                when protodef.Response.ResponseNote.ATOM_FEED
                                    cursor ?= new cursors.AtomFeed @, token, opts, root
                                when protodef.Response.ResponseNote.ORDER_BY_LIMIT_FEED
                                    cursor ?= new cursors.OrderByLimitFeed @, token, opts, root
                        cursor ?= new cursors.Cursor @, token, opts, root

                        # When we've created the cursor, we add it to
                        # the object stored in
                        # `@outstandingCallbacks`.
                        @outstandingCallbacks[token].cursor = cursor

                        # Again, if we have profile information, we
                        # wrap the result given to the callback.  In
                        # either case, we need to add the response to
                        # the new Cursor.
                        if profile?
                            cb null, {profile: profile, value: cursor._addResponse(response)}
                        else
                            cb null, cursor._addResponse(response)
                    when protoResponseType.SUCCESS_SEQUENCE
                        # The `SUCCESS_SEQUENCE` response is sent when
                        # a cursor or feed is complete, and this is
                        # the last response that will be received for
                        # this token. Often, however, the entire
                        # result for a query fits within the initial
                        # batch. In this case, we never see a
                        # `SUCCESS_PARTIAL` response, so there is no
                        # existing Cursor to add the response to. So,
                        # that's what we do here, create a new Cursor,
                        # and delete the token from the
                        # `@outstandingCallbacks`.
                        #
                        # Note that qeries that have already received
                        # a `SUCCESS_PARTIAL` will not be handled
                        # here. They will be handled when we check for
                        # `cursor?` in the conditional up above. In
                        # that branch, we call cursor._addResponse,
                        # which takes care of checking if we received
                        # a `SUCCESS_SEQUENCE`. So this code only gets
                        # executed when the first batch on a cursor is
                        # also our last.
                        cursor = new cursors.Cursor @, token, opts, root
                        @_delQuery(token)
                        if profile?
                            cb null, {profile: profile, value: cursor._addResponse(response)}
                        else
                            cb null, cursor._addResponse(response)
                    when protoResponseType.WAIT_COMPLETE
                        # The `WAIT_COMPLETE` response is sent by the
                        # server after all queries executed with the
                        # optarg `noReply: true` have completed. No
                        # data is returned, so the callback is just provided `null`
                        @_delQuery(token)
                        cb null, null
                    else
                        cb new err.RqlDriverError "Unknown response type"
        else
            # We just ignore tokens we don't have a record of. This is
            # what the other drivers do as well.

    # #### Connection close method
    #
    # A public method that closes the connection. See [API docs for
    # close](http://rethinkdb.com/api/javascript/close/).
    close: (varar 0, 2, (optsOrCallback, callback) ->
        # First determine which argument combination this method was
        # called with, and set the callback and options appropriately.
        if callback?
            # When calling like `.close({...}, function(...){ ... })`
            opts = optsOrCallback
            unless Object::toString.call(opts) is '[object Object]'
                throw new err.RqlDriverError "First argument to two-argument `close` must be an object."
            cb = callback
        else if Object::toString.call(optsOrCallback) is '[object Object]'
            # When calling like `.close({...})`
            opts = optsOrCallback
            cb = null
        else if typeof optsOrCallback is 'function'
            # When calling like `.close(function(...){ ... })`
            opts = {}
            cb = optsOrCallback
        else
            # When calling like `.close({...})`
            opts = optsOrCallback
            cb = null

        for own key of opts
            # Currently, only one optional argument is accepted by
            # `.close`: whether or not to wait for completion of all
            # outstanding `noreply` queries. So we throw an error if
            # anything else is passed.
            unless key in ['noreplyWait']
                throw new err.RqlDriverError "First argument to two-argument `close` must be { noreplyWait: <bool> }."

        # Next we set `@closing` to true. It will be set false once
        # the promise that `.close` returns resolves (see below). The
        # `isOpen` method takes this variable into account.
        @closing = true

        # Should we wait for all outstanding `noreply` writes before
        # considering the connection closed?
        #
        # - if the options object doesn't have a `noreplyWait` key, we
        # default to `true`.
        # - if we do have a `noreplyWait` key, then use that value
        # - if this connection isn't `@open`, then this is all moot
        noreplyWait = ((not opts.noreplyWait?) or opts.noreplyWait) and @open

        # Finally, close returns a promise. This promise must do two
        # things:
        #
        # 1. Set the flags for this connection to false (`@open` &
        # `@closing`)
        #
        # 2. Close all cursors and feeds for outstanding callbacks. We
        # won't be receiving any more responses for them, so they are
        # no longer outstanding. (this is what `@cancel()` does)
        #
        # In addition to these two finalization steps, we may need to
        # wait for all `noreply` queries to finish.
        #
        # You'll notice we don't close sockets or anything like that
        # here. Those steps are done in the subclasses because they
        # have knowledge about what kind of underlying connection is
        # being used.
        new Promise( (resolve, reject) =>
        # Now we create the Promise that `.close` returns. If we
        # aren't waiting for `noreply` queries, then we set the flags
        # and cancel feeds/cursors immediately. If we are waiting,
        # then we run a `noreplyWait` query and set the flags and
        # cancel cursors when that query receives a response.
        #
        # In addition, we chain the callback passed to `.close` on the
        # end, using bluebird's convenience method
        # [nodeify](https://github.com/petkaantonov/bluebird/blob/master/API.md#nodeifyfunction-callback--object-options---promise)
        #
            wrappedCb = (err, result) =>
                @open = false
                @closing = false
                @cancel()
                if err?
                    reject err
                else
                    resolve result

            if noreplyWait
                @noreplyWait(wrappedCb)
            else
                wrappedCb()
        ).nodeify cb
    )

    # #### Connection noreplyWait method
    #
    #  A public method that sends a query to the server that completes
    #  when all previous outstanding queries are completed. See [API
    #  docs for
    #  noreplyWait](http://rethinkdb.com/api/javascript/noreplyWait/).
    noreplyWait: varar 0, 1, (callback) ->

        # If the connection is not open, all of the outstanding
        # `noreply` queries have already been cancelled by the server,
        # so it's an error to call this now. However, `noreplyWait` is
        # called by the `.close` method just before setting the
        # `@open` flag to false. This is why we need the `@closing`
        # flag, so we can allow one last `noreplyWait` call before the
        # connection is closed completely.
        unless @open
            return new Promise( (resolve, reject) ->
                reject(new err.RqlDriverError "Connection is closed.")
            ).nodeify callback

        # Since `noreplyWait` is invoked just like any other query, we
        # need a token.
        token = @nextToken++

        # Here we construct the query itself. There's no body needed,
        # we just have to set the type to the appropriate value from
        # `proto-def` and set the token.
        query = {}
        query.type = protoQueryType.NOREPLY_WAIT
        query.token = token

        new Promise( (resolve, reject) =>
            # The method passed to Promise is invoked immediately
            wrappedCb = (err, result) ->
                # This callback will be invoked when the
                # `NOREPLY_WAIT` query finally returns.
                if (err)
                    reject(err)
                else
                    resolve(result)
            # Here we add an entry for this query to
            # `@outstandingCallbacks`. This promise will be resolved
            # or rejected at that time.
            @outstandingCallbacks[token] = {cb:wrappedCb, root:null, opts:null}
            # Finally, serialize the query and send it to the server
            @_sendQuery(query)
        # After the promise is resolved, the callback passed to
        # `noreplyWait` can be invoked.
        ).nodeify callback

    # #### Connection cancel method
    #
    # This method inserts a dummy error response into all outstanding
    # cursors and feeds, effectively closing them. It also calls all
    # registered callbacks with an error response. Then it clears the
    # `@outstandingCallbacks` registry.
    cancel: ar () ->
        # The type of the dummy response is a runtime error, the
        # reason for the error is "Connection is closed" and the
        # backtrace provided is empty
        response = {t:protoResponseType.RUNTIME_ERROR,r:["Connection is closed."],b:[]}
        for own key, value of @outstandingCallbacks
            if value.cursor?
                value.cursor._addResponse(response)
            else if value.cb?
                value.cb mkErr(err.RqlRuntimeError, response, value.root)

        @outstandingCallbacks = {}

    # #### Connection reconnect method
    #
    # Closes and re-opens a connection. See the [API
    # docs](http://rethinkdb.com/api/javascript/reconnect/)
    reconnect: (varar 0, 2, (optsOrCallback, callback) ->
        # Similar to `.close`, there are a few different ways of
        # calling `.reconnect`, so we need to determine which way it
        # was called, and set the options and callback appropriately.
        if callback?
            # When calling like `reconnect({}, function(..){ ... })`
            opts = optsOrCallback
            cb = callback
        else if typeof optsOrCallback is "function"
            # When calling like `reconnect(function(..){ ... })`
            opts = {}
            cb = optsOrCallback
        else
            # In either of these cases, the callback will be undefined
            if optsOrCallback?
                # Was called like `reconnect({})`
                opts = optsOrCallback
            else
                # Was called like `reconnect()`
                opts = {}
            cb = callback

        # Disconnect and reconnect with the same parameters. Right now
        # this code explicitly mentions the `@rawSocket` attribute,
        # which is only present on the TcpConnection subclass. This
        # code should probably move to that class ultimately.
        new Promise( (resolve, reject) =>
            closeCb = (err) =>
                @rawSocket.removeAllListeners()
                @rawSocket = null # The rawSocket has been closed
                @constructor.call @,
                    host:@host,
                    port:@port
                    timeout:@timeout,
                    authKey:@authKey
                , (err, conn) ->
                    if err?
                        reject err
                    else
                        resolve conn
            @close(opts, closeCb)
        ).nodeify cb
    )

    # #### Connection use method
    #
    # This is a public method, [see the API
    # docs](http://rethinkdb.com/api/javascript/use/). It sets the
    # default db to use when `r.table` is called without specifying a
    # db explicitly. The db variable is sent as a global optarg when
    # querying.
    use: ar (db) ->
        @db = db

    # #### Connection isOpen method
    #
    # This is a non-public method that's used by subclasses to
    # determine if it's safe to call `noreplyWait` when closing. It
    # respects both the `@open` flag and the `@closing` flag, whose
    # behavior is described in the docs for `.close`
    isOpen: () ->
        @open and not @closing

    # #### Connection _start method
    #
    # `_start` combines the raw query term with global optargs,
    # creates a new token for the query and invokes `_sendQuery` with
    # it. `_start` is called by the `.run` method from the [ast
    # module](ast.html)
    _start: (term, cb, opts) ->
        # Obviously, you can't query with a closed connection.
        unless @open then throw new err.RqlDriverError "Connection is closed."

        # Here is where the token for the query is
        # generated. @nextToken is only ever incremented here and in
        # the `.noreplyWait` method
        token = @nextToken++

        # Here we construct the wrapper object for queries
        query = {}

        # The global optargs are the optional arguments passed to
        # `.run`, plus some variables that are set on the connection
        # itself, like `db`.
        query.global_optargs = {}

        # Normal queries begin with `START`. The other options are
        # `CONTINUE` (for feeds and cursors), `STOP` (for closing an
        # open cursor/feed), and `NOREPLY_WAIT` which is a special
        # type of query (see the `.noreplyWait` method).
        query.type = protoQueryType.START
        query.query = term.build()
        query.token = token

        # Now we loop through the options passed to `.run`, convert
        # them to the right format, and put them into the global
        # optargs object for this query
        for own key, value of opts
            # In the protocol, only "snake_case" is accepted for
            # optargs. Because the JS convention is to use
            # "camelCase", we convert it before serializing.
            #
            # In addition, we take the values passed, and convert them
            # into a ReQL term, and dump them in the protocol format,
            # instead of including them directly as they are. This
            # gives resistance to injection attacks.
            query.global_optargs[util.fromCamelCase(key)] = r.expr(value).build()

        # If the user has specified the `db` on the connection (either
        # in the constructor, or with the `.use` method, we add that
        # db to the optargs for the query.
        if @db?
            query.global_optargs['db'] = r.db(@db).build()

        # Next, ensure that the `useOutdated`, `noreply`, and
        # `profile` options (if present) are actual booleans using the
        # `!!` trick. Note that `r.expr(false).build() == false` and
        # `r.expr(null).build() == null`, so the conversion in the
        # loop above won't affect the boolean value the user intended
        # to pass.
        if opts.useOutdated?
            query.global_optargs['use_outdated'] = r.expr(!!opts.useOutdated).build()

        if opts.noreply?
            query.global_optargs['noreply'] = r.expr(!!opts.noreply).build()

        if opts.profile?
            query.global_optargs['profile'] = r.expr(!!opts.profile).build()

        # Here we stash away the callback the user provided in
        # `@outstandingCallbacks`, with the query term and the options
        # passed to `.run`. Note that we don't consider the callback
        # as outstanding if the user has specified `noreply`.  Since
        # the server will never respond, it would sit in
        # `@outstandingCallbacks` forever.
        if (not opts.noreply?) or !opts.noreply
            @outstandingCallbacks[token] = {cb:cb, root:term, opts:opts}

        # Now we send the user's query. `_sendQuery` writes the
        # necessary headers and will serialize it for sending over the
        # underlying connection type (either tcp or http).
        @_sendQuery(query)

        # Finally, if the user called `.run` with both a callback and
        # the `noreply` flag set, we just invoke it immediately, since
        # there won't be a response from the server. Since we have no
        # idea if there's an error and no response, we just pass it
        # null (with error undefined).
        if opts.noreply? and opts.noreply and typeof(cb) is 'function'
            cb null # There is no error and result is `undefined`

    # #### Connection _continueQuery method
    #
    # This method sends a notification to the server that we'd like to
    # receive the next batch from a feed or a cursor
    # (`CONTINUE`). Feeds may block indefinitely, but cursors should
    # return quickly.
    _continueQuery: (token) ->
        unless @open then throw new err.RqlDriverError "Connection is closed."

        query =
            type: protoQueryType.CONTINUE
            token: token

        @_sendQuery(query)

    # #### Connection _endQuery method
    #
    # This method sends a notification to the server that we don't
    # care about the rest of the results on a cursor or feed (`STOP`).
    _endQuery: (token) ->
        unless @open then throw new err.RqlDriverError "Connection is closed."

        query =
            type: protoQueryType.STOP
            token: token

        @_sendQuery(query)

    # #### Connection _sendQuery method
    #
    # This method serializes a javascript object in the rethinkdb json
    # protocol format into a string, and sends it over the underlying
    # connection by calling `_writeQuery`
    _sendQuery: (query) ->
        # The wire protocol doesn't use JSON objects except for things
        # like optargs. Instead it's a bunch of nested arrays, where
        # the first element of the array indicates what the type of
        # that object is. This is a lot like lisp. Here the entire
        # query is wrapped in an array, with the first element the
        # type of the query (`START`, `STOP`, `CONTINUE`, or
        # `NOREPLY_WAIT`).
        data = [query.type]
        # If the query has a body, (which it only does in the `START`
        # case), then we push it into the outer array.
        if !(query.query is undefined)
            data.push(query.query)
            if query.global_optargs? and Object.keys(query.global_optargs).length > 0
                data.push(query.global_optargs)

        # This is the last bit the Connection superclass can
        # do. `_writeQuery` is implemented by the subclasses since
        # they know what underlying connection type is being used.
        # This is also where the query is finally converted into a
        # string.
        @_writeQuery(query.token, JSON.stringify(data))

# ### TcpConnection
#
# This class implements all of the TCP specific behavior for normal
# driver connections. External users of the driver should only use
# `TcpConnection`s since `HttpConnection` are just for the webui.
#
# Fundamentally, this class uses the `net` module to wrap a raw socket
# connection to the server.
class TcpConnection extends Connection

    # #### TcpConnection isAvailable method
    #
    # The TcpConnection should never be used by the webui, so we have
    # an extra method here that decides whether it's available. This
    # is called by the constructor, but also by the `.connect`
    # function which uses it to decide what kind of connection to
    # create.
    @isAvailable: () -> !(process.browser)

    # #### TcpConnection constructor method
    #
    # This sets up all aspects of the connection that relate to
    # TCP. Everything else is done in the Connection superclass
    # constructor.
    constructor: (host, callback) ->
        # Bail out early if we happen to be in the browser
        unless TcpConnection.isAvailable()
            throw new err.RqlDriverError "TCP sockets are not available in this environment"

        # Invoke the superclass's constructor. This initializes the
        # attributes `@host`, `@port`, `@db`, `@authKey` and
        # `@timeout`, `@outstandingCallbacks`, `@nextToken`, `@open`,
        # `@closing`, and `@buffer`. It also registers the callback to
        # be invoked once the `"connect"` event is emitted.
        super(host, callback)

        # Next we create the underlying tcp connection to the server
        # using the net module and store it in the `@rawSocket`
        # attribute.
        @rawSocket = net.connect @port, @host

        # We disable [Nagle's
        # algorithm](http://en.wikipedia.org/wiki/Nagle%27s_algorithm)
        # on the socket because it can impose unnecessary
        # latency.
        @rawSocket.setNoDelay()

        # Here we use the `@timeout` value passed to `.connect` to
        # destroy the socket and emit a timeout error. The timer will
        # be cancelled upon successful connection.
        timeout = setTimeout( (()=>
            @rawSocket.destroy()
            @emit 'error', new err.RqlDriverError "Handshake timedout"
        ), @timeout*1000)

        # If any other kind of error occurs, we also want to cancel
        # the timeout error callback. Otherwise a connection error
        # would be followed shortly by a spurious timeout error.
        @rawSocket.once 'error', => clearTimeout(timeout)

        # Once the TCP socket successfully connects, we can begin the
        # handshake with the server to establish the connection on the
        # server. This is where we decide what protocol to use, and
        # send things like the authKey etc.
        @rawSocket.once 'connect', =>

            # The protocol specifies that the magic number for the
            # version should be given as a little endian 32 bit
            # unsigned integer. The value is given in the `proto-def`
            # module, but it is just a random number that's unlikely
            # to be accidentally the first few bytes of an erroneous
            # connection on the socket.
            version = new Buffer(4)
            version.writeUInt32LE(protoVersion, 0)

            # Since the auth key has a variable length, we need to
            # both encode its length in the wire format (little endian
            # 32 bit), and the bytes themselves.
            auth_buffer = new Buffer(@authKey, 'ascii')
            auth_length = new Buffer(4)
            auth_length.writeUInt32LE(auth_buffer.length, 0)

            # Send the protocol type that we will be using to
            # communicate with the server. This can be either
            # protobuf, or json. The protobuf protocol is deprecated
            # however.
            protocol = new Buffer(4)
            protocol.writeUInt32LE(protoProtocol, 0)

            # Write the version, auth key length, auth key, and
            # protocol number to the socket, in that order.
            @rawSocket.write Buffer.concat([version, auth_length, auth_buffer, protocol])

            # Now we have to wait for a response from the server
            # acknowledging the new connection. The following callback
            # will be invoked when the server sends the first few
            # bytes over the socket.
            handshake_callback = (buf) =>
                # Once we receive a response, extend the current
                # buffer with the data from the server. The reason we
                # extend it, vs. just setting it to the value of `buf`
                # is that this callback may be invoked several times
                # before the server response is received in full. Each
                # time through we check if we've received the entire
                # response, and only disable this event listener at
                # that time.
                @buffer = Buffer.concat([@buffer, buf])

                # Next we read bytes until we get a null byte. This is
                # the response string from the server and should just
                # be "SUCCESS\0". Anything else is an error.
                for b,i in @buffer
                    if b is 0
                        # Once we get the null byte, the server
                        # response has been received and we can turn
                        # off the handshake callback
                        @rawSocket.removeListener('data', handshake_callback)

                        # Here we pull the status string out of the
                        # buffer and convert it into a string.
                        status_buf = @buffer.slice(0, i)
                        @buffer = @buffer.slice(i + 1)
                        status_str = status_buf.toString()

                        # We also want to cancel the timeout error
                        # callback that we set earlier. Even though we
                        # haven't checked `status_str` yet, we've
                        # gotten a response so it isn't a timeout.
                        clearTimeout(timeout)

                        # Finally, check the status string.
                        if status_str == "SUCCESS"
                            # Set up the `_data` method to receive all
                            # further responses from the server. This
                            # callback is only for the initial
                            # connection, all future responses will be
                            # in reply to queries.
                            #
                            # We wrap @_data in a function instead of
                            # just giving it as a callback directly so
                            # that it gets bound to the correct
                            # `this`.
                            @rawSocket.on 'data', (buf) => @_data(buf)

                            # Notify listeners we've connected
                            # successfully. Notably, the Connection
                            # constructor registers a listener for the
                            # `"connect"` event that sets the `@open`
                            # flag on the connection and invokes the
                            # callback passed to the `r.connect`
                            # function.
                            @emit 'connect'
                            return
                        else
                            # The protocol dictates that any other
                            # string but `SUCCESS` is an error
                            # indicating the problem, and that we
                            # should report the error to the user.
                            @emit 'error', new err.RqlDriverError "Server dropped connection with message: \"" + status_str.trim() + "\""
                            return

            # Register the handshake callback on the socket. Once the
            # handshake completes, it will unregister itself and
            # register `_data` on the socket for `"data"` events.
            @rawSocket.on 'data', handshake_callback

        # In case the socket encounters an error, we re-emit the error
        # ourselves. The Connection superclass constructor will invoke
        # the user's `r.connect` callback with the appropriate error
        # object.
        @rawSocket.on 'error', (err) => @emit 'error', err

        # Upon closing the `@rawSocket` for any reason, we close this
        # connection. Passing `noreplyWait: false` ensures that we
        # won't try to emit anything else on the socket.
        @rawSocket.on 'close', =>
            if @isOpen()
                @close({noreplyWait: false})
            @emit 'close'

        # We re-raise the `timeout` event. Node says the socket must
        # be closed by the user though. We should do that rather than
        # just set the `@open` flag to `false`
        @rawSocket.on 'timeout', => @open = false; @emit 'timeout'

    # #### TcpConnection close method
    #
    # This is a public method for closing the current connection. It
    # performs the raw socket related cleanup, and delegates the work
    # of setting `@open` etc to the Connection superclass's
    # implementation of `.close`
    close: (varar 0, 2, (optsOrCallback, callback) ->
        # Here we handle different ways of calling `.close`
        if callback?
            # This is when calling close like `.close({..}, function(..) {})`
            opts = optsOrCallback
            cb = callback
        else if Object::toString.call(optsOrCallback) is '[object Object]'
            # This is when calling close like `.close({..})`
            opts = optsOrCallback
            cb = null
        else if typeof optsOrCallback is "function"
            # This is when calling close like `.close(function(..){..})`
            opts = {}
            cb = optsOrCallback
        else
            # And the default is when calling like `.close()`
            opts = {}

        # Finally, we create the promise for this function. It closes
        # the socket and invokes the superclass to clean up everything
        # else, then invokes the user's callback to this function.
        #
        # In order:
        #
        # 1. The promise is initialized, with `resolve` and `reject` passed
        #    to the anonymous promise callback immediately.
        # 2. The anonymous promise callback calls the superclass's
        #    `close` method on this connection, with the `opts` that
        #    were passed to this function. The only option `close`
        #    accepts is `noreplyWait`
        # 3. The superclass's close method sets `@open` to false, and
        #    also closes all cursors, feeds, and callbacks that are
        #    currently outstanding.
        # 4. Depending on whether `opts.noreplyWait` is true, the
        #    superclass's `close` may immediately invoke `wrappedCb`,
        #    or it may defer it until all outstanding `noreply`
        #    queries are completed.
        # 5. When `wrappedCb` is invoked, it calls `.end()` on the raw
        #    socket for this connection, telling Node we'd like the
        #    tcp connection to be closed.
        # 6. The socket will raise the `"close"` event when it's done
        #    closing, and at that time the promise here will be
        #    resolved or rejected.
        # 7. If the promise was resolved, the callback provided to
        #    this function (`cb`) will be run (it's chained to the
        #    promise).
        new Promise( (resolve, reject) =>
            wrappedCb = (error, result) =>
                @rawSocket.once "close", =>
                    if error?
                        reject error
                    else
                        resolve result

                @rawSocket.end()

            TcpConnection.__super__.close.call(@, opts, wrappedCb)
        ).nodeify cb
    )

    # #### TcpConnection close method
    #
    # This method is called by the superclass in the `.close`
    # method. It simply closes the socket completely and calls the
    # superclass's `cancel`. The superclass's cancel closes all
    # outstanding cursors and feeds, and invokes all outstanding query
    # callbacks with an error.
    #
    # Despite not starting with an underscore, this is not a public
    # method.
    cancel: () ->
        @rawSocket.destroy()
        super()

    # #### TcpConnection _writeQuery method
    #
    # This method emits a query's token on the socket, then invokes
    # `.write` which emits the body of the query.
    _writeQuery: (token, data) ->
        tokenBuf = new Buffer(8)
        # We write the token in two 4-byte chunks: least significant
        # bytes first, then most significant bytes. (This is the same
        # order as little endian 8 bytes would be). The `_data` method
        # in the `Connection` class extracts the token from server
        # responses by reversing this process.
        #
        # The key here is to write the token in a consistent way (so
        # responses can be paired with outstanding requests), and to
        # ensure the number isn't interpreted as a negative number by
        # the server. The server doesn't actually care if the numbers
        # are sequential.
        tokenBuf.writeUInt32LE(token & 0xFFFFFFFF, 0)
        tokenBuf.writeUInt32LE(Math.floor(token / 0xFFFFFFFF), 4)

        # Write out the token to the socket, then invoke `.write` to
        # write the data to the socket.
        @rawSocket.write tokenBuf
        @write new Buffer(data)

    # #### TcpConnection write method
    #
    # This method is called by `_writeQuery` and just writes the body
    # of a query to the socket.
    #
    # Despite not starting with an underscore, it is not a public
    # method.
    write: (chunk) ->
        # Before writing the data, we need to write the length of the
        # data as an unsigned little-endian 4 byte integer.
        lengthBuffer = new Buffer(4)
        lengthBuffer.writeUInt32LE(chunk.length, 0)
        @rawSocket.write lengthBuffer
        # Finally we write the data to the socket. It is serialized
        # from json to a string in the `Connection` class's
        # `_sendQuery` method.
        @rawSocket.write chunk

# ### HttpConnection class
#
# This class is used to run queries from the webui to the RethinkDB
# internal http server using XHR. It's not intended for general usage
# in applications.
class HttpConnection extends Connection
    # We defined the default protocol in case the user doesn't specify
    # one.
    DEFAULT_PROTOCOL: 'http'

    # A static method used by `r.connect` to decide which kind of
    # connection to create in a given environment. Here we check if
    # XHRs are defined. If not, we aren't in the browser and shouldn't
    # be using `HttpConnection`
    @isAvailable: -> typeof XMLHttpRequest isnt "undefined"

    # #### HttpConnection constructor method
    #
    # This method sets up the XMLHttpRequest object that all
    # requests/responses will be sent and received on. Unlike the
    # TcpConnection, the HTTP server only allows one outstanding query
    # at a time per connection. Because of this, currently the web ui
    # creates many different connections.
    #
    # The state of an HTTP connection is actually stored on the
    # server, and they automatically time out after 5 minutes. The way
    # it works is we send a POST request to:
    # `/ajax/reql/open-new-connection`, which gets us a connection
    # id. The browser can also explicitly destroy server connections
    # with the `/ajax/reql/close-connection?conn_id=` url.
    #
    #  Ultimately, the server keeps track of the connections because
    # it needs to give the client a way to fetch more results from
    # cursors and feeds.
    constructor: (host, callback) ->
        # A quick check to ensure we can create an `HttpConnection`
        unless HttpConnection.isAvailable()
            throw new err.RqlDriverError "XMLHttpRequest is not available in this environment"
        # Call the superclass constructor. This initializes the
        # attributes `@host`, `@port`, `@db`, `@authKey`, `@timeout`,
        # `@outstandingCallbacks`, `@nextToken`, `@open`, `@closing`,
        # and `@buffer`. It also registers the callback to be invoked
        # once the `"connect"` event is emitted.
        super(host, callback)

        # A protocol may be supplied in the host object. If the host
        # doesn't have a protocol key, (or is a string, which it's
        # also allowed to be), then use the default protocol, which is
        # 'http'
        protocol = if host.protocol is 'https' then 'https' else @DEFAULT_PROTOCOL

        # Next we construct an XHR with the path to the reql query
        # endpoint on the http server.
        url = "#{protocol}://#{@host}:#{@port}#{host.pathname}ajax/reql/"
        xhr = new XMLHttpRequest

        #The request is a `POST` to ensure the response isn't cached
        # by the browser (which has caused bugs in the past). The
        # `true` argument is setting the request to be asynchronous.
        xhr.open("POST", url+"open-new-connection", true)

        # We also want the response to come back as a buffer, to
        # minimize the code that has to be HttpConnection specific
        xhr.responseType = "arraybuffer"

        # Next, set up the callback to process the response from the
        # server when the new connection is established.
        xhr.onreadystatechange = (e) =>
            # readyState 4 is "DONE", the data has been completely received.
            if xhr.readyState is 4
                # The status is 200 if we get a successful response
                if xhr.status is 200
                    # Now we finally store the `@_url` variable. No
                    # particular reason for not doing this sooner
                    # since the url doesn't change.
                    @_url = url

                    # We create a
                    # [DataView](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/DataView)
                    # for reading the data as an array of bytes. We
                    # grab the value passed back and interpret it as a
                    # signed 32 bit little-endian integer. (The `true`
                    # argument causes it to be interpreted as
                    # little-endian).
                    @_connId = (new DataView xhr.response).getInt32(0, true)

                    # Emit the `"connect"` event. The `Connection`
                    # superclass's constructor listens for this event
                    # to set the `@open` flag to `true` and invoke its
                    # callback (which is callback passed to this
                    # method)
                    @emit 'connect'
                else
                    # In case we get anything other than a 200, raise an error.
                    @emit 'error', new err.RqlDriverError "XHR error, http status #{xhr.status}."
        # Send the xhr, and store it in the instance for later.
        xhr.send()

        @xhr = xhr

    # #### HttpConnection cancel method
    #
    # This sends a POST to the `close-connection` endpoint, so the
    # server can deallocate resources related to the connection. This
    # throws out any data for unconsumed cursors etc and invalidates
    # this connection id.
    cancel: ->
        # Check if the connection id is still open. `@connId` is null
        # if the connection was previously closed or cancelled
        if @_connId?
            # Since only one request can happen at a time, we need to
            # abort any in-progress xhr request before sending a new
            # request to close the connection.
            @xhr.abort()
            xhr = new XMLHttpRequest

            # This sends the close connection request asynchronously.
            xhr.open("POST", "#{@_url}close-connection?conn_id=#{@_connId}", true)

            # We ignore the result, but Firefox doesn't. Without this line it complains
            # about "No element found" when trying to parse the response as xml.
            xhr.responseType = "arraybuffer"

            xhr.send()

            # Null out @conn_Id so we know not to send another
            # close request to the server later.
            @_url = null
            @_connId = null

            # This will end any outstanding cursors and feeds, though
            # there should be at most one since we don't support
            # multiple queries at once on http connections.
            super()

    # #### HttpConnection close method
    #
    # Closes the http connection. Does nothing new or special beyond
    # what the superclass already does. The server connection is
    # closed in `cancel` above, which the superclass calls for us.
    close: (varar 0, 2, (optsOrCallback, callback) ->
        if callback?
            opts = optsOrCallback
            cb = callback
        else if Object::toString.call(optsOrCallback) is '[object Object]'
            opts = optsOrCallback
            cb = null
        else
            opts = {}
            cb = optsOrCallback
        unless not cb? or typeof cb is 'function'
            throw new err.RqlDriverError "Final argument to `close` must be a callback function or object."

        # This would simply be `super(opts, wrappedCb)`, if we were
        # not in the varar anonymous function
        HttpConnection.__super__.close.call(this, opts, cb)
    )

    # #### HttpConnection _writeQuery method
    #
    # This creates the buffer to send to the server, writes the token
    # and data to it, then passes the data and buffer down to `write`
    # to write the data itself to the xhr request.
    #
    # This differs from `TcpConnection`'s version of `_writeQuery`
    # which only writes the token.
    _writeQuery: (token, data) ->
        # We use encodeURI to get the length of the unicode
        # representation of the data.
        buf = new Buffer(encodeURI(data).split(/%..|./).length - 1 + 8)

        # Again, the token is encoded as a little-endian 8 byte
        # unsigned integer.
        buf.writeUInt32LE(token & 0xFFFFFFFF, 0)
        buf.writeUInt32LE(Math.floor(token / 0xFFFFFFFF), 4)

        # Write the data out to the bufferm then invoke `write` with
        # the buffer and token
        buf.write(data, 8)
        @write buf, token

    # #### HttpConnection write method
    #
    # This method takes a Node Buffer (or more accurately, the Buffer
    # class that Browserify polyfills for us) and turns it into an
    # ArrayBufferView, then sends it over the XHR with the current
    # connection id.
    #
    # Despite not starting with an underscore, this is not a public
    # method.
    write: (chunk, token) ->
        # Create a new XHR with to run the query over.
        xhr = new XMLHttpRequest
        xhr.open("POST", "#{@_url}?conn_id=#{@_connId}", true)
        xhr.responseType = "arraybuffer"

        xhr.onreadystatechange = (e) =>
            # When the server responds, the state must be 4 ("DONE"),
            # and the HTTP status code should be success. Otherwise we
            # just do nothing. There's a separate callback if the XHR
            # encounters an error.
            if xhr.readyState is 4 and xhr.status is 200
                # Convert the response from a browser ArrayBuffer into
                # a Node buffer
                buf = new Buffer(b for b in (new Uint8Array(xhr.response)))

                # The data method is defined in the `Connection` class
                # and reads and parses the response from the server.
                @_data(buf)

        # On error we want to invoke the callback for this query with
        # an error.
        xhr.onerror = (e) =>
            @outstandingCallbacks[token].cb(new Error("This HTTP connection is not open"))

        # To serialize the query (held in chunk) for the XHR, we need
        # to convert it from a node buffer into an `ArrayBufferView`
        # (specifically `Uint8Array`), since passing an `ArrayBuffer`
        # in xhr.send is deprecated
        view = new Uint8Array(chunk.length)
        i = 0
        while i < chunk.length
            view[i] = chunk[i]
            i++

        # And send it on its way to the server
        xhr.send view
        @xhr = xhr

# ## isConnection
#
# This is an exported function that determines whether something is a
# Connection. It's used by the [ast module](ast.html) to validate the
# first argument passed to `.run` (which must always be a connection).
module.exports.isConnection = (connection) ->
    return connection instanceof Connection

# ## connect
#
# The main function of this module, which is exposed to end users as
# `r.connect`. See the [API
# docs](http://rethinkdb.com/api/javascript/connect/).
module.exports.connect = varar 0, 2, (hostOrCallback, callback) ->
    if typeof hostOrCallback is 'function'
        # This occurs when called like `r.connect(function(..){..})`
        # Override the callback variable (which should be undefined)
        # and set host to an empty object
        host = {}
        callback = hostOrCallback
    else
        # Otherwise, the `callback` variable is already correctly
        # holding the callback, and the host variable is the first
        # argument
        host = hostOrCallback

    # `r.connect` returns a Promise which does the following:
    #
    # 1. Determines whether to create a `TcpConnection` or an
    #    `HttpConnection` depending on whether we're running on Node
    #    or in the browser.
    # 2. Initializes the connection, and when it's complete invokes
    #    the user's callback
    new Promise( (resolve, reject) ->
        create_connection = (host, callback) =>
            if TcpConnection.isAvailable()
                new TcpConnection host, callback
            else if HttpConnection.isAvailable()
                new HttpConnection host, callback
            else
                throw new err.RqlDriverError "Neither TCP nor HTTP avaiable in this environment"

        wrappedCb = (err, result) ->
            if (err)
                reject(err)
            else
                resolve(result)
        create_connection(host, wrappedCb)
    ).nodeify callback