def __send_message(self, message):
        """Send a query or getmore message and handles the response.
        """
        db = self.__collection.database
        kwargs = {"_sock": self.__socket,
                  "_must_use_master": self.__must_use_master}
        if self.__connection_id is not None:
            kwargs["_connection_to_use"] = self.__connection_id

        response = db.connection._send_message_with_response(message,
                                                             **kwargs)

        if isinstance(response, tuple):
            (connection_id, response) = response
        else:
            connection_id = None

        self.__connection_id = connection_id

        try:
            response = helpers._unpack_response(response, self.__id)
        except AutoReconnect:
            db.connection._reset()
            raise
        self.__id = response["cursor_id"]

        # starting from doesn't get set on getmore's for tailable cursors
        if not self.__tailable:
            assert response["starting_from"] == self.__retrieved

        self.__retrieved += response["number_returned"]
        self.__data = response["data"]

        if self.__limit and self.__id and self.__limit <= self.__retrieved:
            self.__die()

    def __send_message(self, operation):
        """Send a getmore message and handle the response.
        """
        client = self.__collection.database.client
        try:
            response = client._send_message_with_response(
                operation, address=self.__address)
        except AutoReconnect:
            # Don't try to send kill cursors on another socket
            # or to another server. It can cause a _pinValue
            # assertion on some server releases if we get here
            # due to a socket timeout.
            self.__killed = True
            raise

        try:
            doc = helpers._unpack_response(response.data,
                                           self.__id,
                                           self.__collection.codec_options)
        except CursorNotFound:
            self.__killed = True
            raise
        except NotMasterError:
            # Don't send kill cursors to another server after a "not master"
            # error. It's completely pointless.
            self.__killed = True
            client._reset_server_and_request_check(self.address)
            raise
        self.__id = doc["cursor_id"]
        if self.__id == 0:
            self.__killed = True

        self.__retrieved += doc["number_returned"]
        self.__data = deque(doc["data"])

def command(sock, dbname, spec, slave_ok, is_mongos, read_preference,
            codec_options, check=True, allowable_errors=None):
    """Execute a command over the socket, or raise socket.error.

    :Parameters:
      - `sock`: a raw socket instance
      - `dbname`: name of the database on which to run the command
      - `spec`: a command document as a dict, SON, or mapping object
      - `slave_ok`: whether to set the SlaveOkay wire protocol bit
      - `is_mongos`: are we connected to a mongos?
      - `read_preference`: a read preference
      - `codec_options`: a CodecOptions instance
      - `check`: raise OperationFailure if there are errors
      - `allowable_errors`: errors to ignore if `check` is True
    """
    ns = dbname + '.$cmd'
    flags = 4 if slave_ok else 0
    if is_mongos:
        spec = message._maybe_add_read_preference(spec, read_preference)
    request_id, msg, _ = message.query(flags, ns, 0, -1, spec,
                                       None, codec_options)
    sock.sendall(msg)
    response = receive_message(sock, 1, request_id)
    unpacked = helpers._unpack_response(response, codec_options=codec_options)
    response_doc = unpacked['data'][0]
    msg = "command %s on namespace %s failed: %%s" % (
        repr(spec).replace("%", "%%"), ns)
    if check:
        helpers._check_command_response(response_doc, msg, allowable_errors)
    return response_doc

    def __check_response_to_last_error(self, response):
        """Check a response to a lastError message for errors.

        `response` is a byte string representing a response to the message.
        If it represents an error response we raise OperationFailure.

        Return the response as a document.
        """
        response = helpers._unpack_response(response)

        assert response["number_returned"] == 1
        error = response["data"][0]

        helpers._check_command_response(error, self.disconnect)

        # TODO unify logic with database.error method
        if error.get("err", 0) is None:
            return error
        if error["err"] == "not master":
            self.disconnect()
            raise AutoReconnect("not master")

        if "code" in error:
            if error["code"] in [11000, 11001]:
                raise DuplicateKeyError(error["err"])
            else:
                raise OperationFailure(error["err"], error["code"])
        else:
            raise OperationFailure(error["err"])

    def __auth(self, sock, dbase, user, passwd):
        """Athenticate `sock` against `dbase`
        """
        # TODO: Error handling...
        # Get a nonce
        request_id, msg, _ = message.query(0, dbase + '.$cmd',
                                           0, -1, {'getnonce': 1})
        sock.sendall(msg)
        raw = self.__recv_msg(1, request_id, sock)
        nonce = helpers._unpack_response(raw)['data'][0]['nonce']
        key = helpers._auth_key(nonce, user, passwd)

        # Actually authenticate
        query = {'authenticate': 1, 'user': user, 'nonce': nonce, 'key': key}
        request_id, msg, _ = message.query(0, dbase + '.$cmd', 0, -1, query)
        sock.sendall(msg)
        raw = self.__recv_msg(1, request_id, sock)
        print helpers._unpack_response(raw)['data'][0]

 def __simple_command(self, sock_info, dbname, spec):
     """Send a command to the server.
     """
     rqst_id, msg, _ = message.query(0, dbname + ".$cmd", 0, -1, spec)
     sock_info.sock.sendall(msg)
     response = self.__receive_message_on_socket(1, rqst_id, sock_info)
     response = helpers._unpack_response(response)["data"][0]
     msg = "command %r failed: %%s" % spec
     helpers._check_command_response(response, None, msg)
     return response

 def __simple_command(self, sock, dbname, spec):
     """Send a command to the server.
     """
     rqst_id, msg, _ = message.query(0, dbname + '.$cmd', 0, -1, spec)
     sock.sendall(msg)
     response = self.__recv_msg(1, rqst_id, sock)
     response = helpers._unpack_response(response)['data'][0]
     msg = "command %r failed: %%s" % spec
     helpers._check_command_response(response, None, msg)
     return response

    def __send_message(self, message):
        """Send a query or getmore message and handles the response.
        """
        db = self.__collection.database
        kwargs = {"_must_use_master": self.__must_use_master}
        kwargs["read_preference"] = self.__read_preference
        kwargs["tag_sets"] = self.__tag_sets
        kwargs["secondary_acceptable_latency_ms"] = (
            self.__secondary_acceptable_latency_ms)
        if self.__connection_id is not None:
            kwargs["_connection_to_use"] = self.__connection_id
        kwargs.update(self.__kwargs)

        try:
            response = db.connection._send_message_with_response(message,
                                                                 **kwargs)
        except AutoReconnect:
            # Don't try to send kill cursors on another socket
            # or to another server. It can cause a _pinValue
            # assertion on some server releases if we get here
            # due to a socket timeout.
            self.__killed = True
            raise

        if isinstance(response, tuple):
            (connection_id, response) = response
        else:
            connection_id = None

        self.__connection_id = connection_id

        try:
            response = helpers._unpack_response(response, self.__id,
                                                self.__as_class,
                                                self.__tz_aware,
                                                self.__uuid_subtype)
        except AutoReconnect:
            # Don't send kill cursors to another server after a "not master"
            # error. It's completely pointless.
            self.__killed = True
            db.connection.disconnect()
            raise
        self.__id = response["cursor_id"]

        # starting from doesn't get set on getmore's for tailable cursors
        if not self.__tailable:
            assert response["starting_from"] == self.__retrieved, (
                "Result batch started from %s, expected %s" % (
                    response['starting_from'], self.__retrieved))

        self.__retrieved += response["number_returned"]
        self.__data = deque(response["data"])

        if self.__limit and self.__id and self.__limit <= self.__retrieved:
            self.__die()

 def __is_master(self, candidate):
     """Directly call ismaster.
     """
     # TODO: Error handling...
     request_id, msg, _ = message.query(0, 'admin.$cmd',
                                        0, -1, {'ismaster': 1})
     mongo = pool.Pool(candidate, self.__max_pool_size,
                       self.__net_timeout, self.__conn_timeout)
     sock = mongo.get_socket()[0]
     sock.sendall(msg)
     raw = self.__recv_msg(1, request_id, sock)
     response = helpers._unpack_response(raw)['data'][0]
     return response, mongo

    def _check_with_socket(self, sock_info):
        """Return (IsMaster, round_trip_time).

        Can raise ConnectionFailure or OperationFailure.
        """
        start = _time()
        request_id, msg, max_doc_size = message.query(
            0, 'admin.$cmd', 0, -1, {'ismaster': 1},
            None, DEFAULT_CODEC_OPTIONS)

        # TODO: use sock_info.command()
        sock_info.send_message(msg, max_doc_size)
        raw_response = sock_info.receive_message(1, request_id)
        result = helpers._unpack_response(raw_response)
        return IsMaster(result['data'][0]), _time() - start

    def __simple_command(self, sock_info, dbname, spec):
        """Send a command to the server.
        """
        rqst_id, msg, _ = message.query(0, dbname + '.$cmd', 0, -1, spec)
        start = time.time()
        try:
            sock_info.sock.sendall(msg)
            response = self.__receive_message_on_socket(1, rqst_id, sock_info)
        except:
            sock_info.close()
            raise

        end = time.time()
        response = helpers._unpack_response(response)['data'][0]
        msg = "command %r failed: %%s" % spec
        helpers._check_command_response(response, None, msg)
        return response, end - start

    def write_command(self, request_id, msg):
        """Send "insert" etc. command, returning response as a dict.

        Can raise ConnectionFailure or OperationFailure.

        :Parameters:
          - `request_id`: an int.
          - `msg`: bytes, the command message.
        """
        self.send_message(msg, 0)
        response = helpers._unpack_response(self.receive_message(1, request_id))
        assert response['number_returned'] == 1
        result = response['data'][0]

        # Raises NotMasterError or OperationFailure.
        helpers._check_command_response(result)
        return result

    def nextback(self,callback,response):
       
        if response is not None:
            if isinstance(response, tuple):
                (connection_id, response) = response
            else:
                connection_id = None
    
            self.__connection_id = connection_id
    
            try:
                response = helpers._unpack_response(response, self.__id,as_class=self.__as_class)
            except AutoReconnect:
                db.connection._reset()
                raise
                
     
            self.__id = response["cursor_id"]
            
            # starting from doesn't get set on getmore's for tailable selfs
            if not self.__tailable:
                assert response["starting_from"] == self.__retrieved
    
            self.__retrieved += response["number_returned"]

            self.__data = response["data"]
         
            
            self.__id = response["cursor_id"]
    
    
        if self.__limit and self.__id and self.__limit <= self.__retrieved or not self.__id:
            self.__die()
    
        db = self.__collection.database
        
        if len(self.__data):
            callback(db._fix_outgoing(self.__data.pop(0), self.__collection)) 
        else:
            callback(StopIteration)

    def __send_message(self, msg):
        """Send a getmore message and handle the response.
        """
        client = self.__collection.database.connection
        try:
            res = client._send_message_with_response(
                msg, _connection_to_use=self.__conn_id)
            self.__conn_id, (response, dummy0, dummy1) = res
        except AutoReconnect:
            # Don't try to send kill cursors on another socket
            # or to another server. It can cause a _pinValue
            # assertion on some server releases if we get here
            # due to a socket timeout.
            self.__killed = True
            raise

        try:
            response = helpers._unpack_response(response,
                                                self.__id,
                                                *self.__decode_opts)
        except CursorNotFound:
            self.__killed = True
            raise
        except AutoReconnect:
            # Don't send kill cursors to another server after a "not master"
            # error. It's completely pointless.
            self.__killed = True
            client.disconnect()
            raise
        self.__id = response["cursor_id"]
        if self.__id == 0:
            self.__killed = True

        assert response["starting_from"] == self.__retrieved, (
            "Result batch started from %s, expected %s" % (
                response['starting_from'], self.__retrieved))

        self.__retrieved += response["number_returned"]
        self.__data = deque(response["data"])

    def __send_message(self, msg):
        """Send a getmore message and handle the response.
        """
        client = self.__collection.database.connection
        try:
            res = client._send_message_with_response(
                msg, _connection_to_use=self.__conn_id)
            self.__conn_id, (response, dummy0, dummy1) = res
        except AutoReconnect:
            # Don't try to send kill cursors on another socket
            # or to another server. It can cause a _pinValue
            # assertion on some server releases if we get here
            # due to a socket timeout.
            self.__killed = True
            raise

        try:
            response = helpers._unpack_response(
                response,
                self.__id,
                self.__codec_options.document_class,
                self.__codec_options.tz_aware,
                self.__codec_options.uuid_representation,
                self.__compile_re)
        except CursorNotFound:
            self.__killed = True
            raise
        except AutoReconnect:
            # Don't send kill cursors to another server after a "not master"
            # error. It's completely pointless.
            self.__killed = True
            client.disconnect()
            raise
        self.__id = response["cursor_id"]
        if self.__id == 0:
            self.__killed = True

        self.__retrieved += response["number_returned"]
        self.__data = deque(response["data"])

    def __check_response_to_last_error(self, response):
        """Check a response to a lastError message for errors.

        `response` is a byte string representing a response to the message.
        If it represents an error response we raise OperationFailure.

        Return the response as a document.
        """
        response = helpers._unpack_response(response)

        assert response["number_returned"] == 1
        error = response["data"][0]

        helpers._check_command_response(error, self.disconnect)

        error_msg = error.get("err", "")
        if error_msg is None:
            return error
        if error_msg.startswith("not master"):
            self.disconnect()
            raise AutoReconnect(error_msg)

        details = error
        # mongos returns the error code in an error object
        # for some errors.
        if "errObjects" in error:
            for errobj in error["errObjects"]:
                if errobj["err"] == error_msg:
                    details = errobj
                    break

        if "code" in details:
            if details["code"] in (11000, 11001, 12582):
                raise DuplicateKeyError(details["err"], details["code"])
            else:
                raise OperationFailure(details["err"], details["code"])
        else:
            raise OperationFailure(details["err"])

    def __send_message(self, operation):
        """Send a getmore message and handle the response.
        """
        client = self.__collection.database.client
        listeners = client._event_listeners
        publish = listeners.enabled_for_commands
        try:
            response = client._send_message_with_response(
                operation, address=self.__address)
        except AutoReconnect:
            # Don't try to send kill cursors on another socket
            # or to another server. It can cause a _pinValue
            # assertion on some server releases if we get here
            # due to a socket timeout.
            self.__killed = True
            raise

        cmd_duration = response.duration
        rqst_id = response.request_id
        if publish:
            start = datetime.datetime.now()
        try:
            doc = helpers._unpack_response(response.data,
                                           self.__id,
                                           self.__collection.codec_options)
        except OperationFailure as exc:
            self.__killed = True

            if publish:
                duration = (datetime.datetime.now() - start) + cmd_duration
                listeners.publish_command_failure(
                    duration, exc.details, "getMore", rqst_id, self.__address)

            raise
        except NotMasterError as exc:
            # Don't send kill cursors to another server after a "not master"
            # error. It's completely pointless.
            self.__killed = True

            if publish:
                duration = (datetime.datetime.now() - start) + cmd_duration
                listeners.publish_command_failure(
                    duration, exc.details, "getMore", rqst_id, self.__address)

            client._reset_server_and_request_check(self.address)
            raise
        except Exception as exc:
            if publish:
                duration = (datetime.datetime.now() - start) + cmd_duration
                listeners.publish_command_failure(
                    duration, _convert_exception(exc), "getMore", rqst_id,
                    self.__address)
            raise

        if publish:
            duration = (datetime.datetime.now() - start) + cmd_duration
            # Must publish in getMore command response format.
            res = {"cursor": {"id": doc["cursor_id"],
                              "ns": self.__collection.full_name,
                              "nextBatch": doc["data"]},
                   "ok": 1}
            listeners.publish_command_success(
                duration, res, "getMore", rqst_id, self.__address)

        self.__id = doc["cursor_id"]
        if self.__id == 0:
            self.__killed = True

        self.__retrieved += doc["number_returned"]
        self.__data = deque(doc["data"])

def command(sock, dbname, spec, slave_ok, is_mongos,
            read_preference, codec_options, check=True,
            allowable_errors=None, address=None,
            check_keys=False, listeners=None, max_bson_size=None,
            read_concern=DEFAULT_READ_CONCERN):
    """Execute a command over the socket, or raise socket.error.

    :Parameters:
      - `sock`: a raw socket instance
      - `dbname`: name of the database on which to run the command
      - `spec`: a command document as a dict, SON, or mapping object
      - `slave_ok`: whether to set the SlaveOkay wire protocol bit
      - `is_mongos`: are we connected to a mongos?
      - `read_preference`: a read preference
      - `codec_options`: a CodecOptions instance
      - `check`: raise OperationFailure if there are errors
      - `allowable_errors`: errors to ignore if `check` is True
      - `address`: the (host, port) of `sock`
      - `check_keys`: if True, check `spec` for invalid keys
      - `listeners`: An instance of :class:`~pymongo.monitoring.EventListeners`
      - `max_bson_size`: The maximum encoded bson size for this server
      - `read_concern`: The read concern for this command.
    """
    name = next(iter(spec))
    ns = dbname + '.$cmd'
    flags = 4 if slave_ok else 0
    # Publish the original command document.
    orig = spec
    if is_mongos:
        spec = message._maybe_add_read_preference(spec, read_preference)
    if read_concern.level:
        spec['readConcern'] = read_concern.document

    publish = listeners is not None and listeners.enabled_for_commands
    if publish:
        start = datetime.datetime.now()

    request_id, msg, size = message.query(flags, ns, 0, -1, spec,
                                          None, codec_options, check_keys)

    if (max_bson_size is not None
            and size > max_bson_size + message._COMMAND_OVERHEAD):
        message._raise_document_too_large(
            name, size, max_bson_size + message._COMMAND_OVERHEAD)

    if publish:
        encoding_duration = datetime.datetime.now() - start
        listeners.publish_command_start(orig, dbname, request_id, address)
        start = datetime.datetime.now()

    try:
        sock.sendall(msg)
        response = receive_message(sock, 1, request_id)
        unpacked = helpers._unpack_response(
            response, codec_options=codec_options)

        response_doc = unpacked['data'][0]
        if check:
            helpers._check_command_response(response_doc, None, allowable_errors)
    except Exception as exc:
        if publish:
            duration = (datetime.datetime.now() - start) + encoding_duration
            if isinstance(exc, (NotMasterError, OperationFailure)):
                failure = exc.details
            else:
                failure = message._convert_exception(exc)
            listeners.publish_command_failure(
                duration, failure, name, request_id, address)
        raise
    if publish:
        duration = (datetime.datetime.now() - start) + encoding_duration
        listeners.publish_command_success(
            duration, response_doc, name, request_id, address)
    return response_doc

    def __send_message(self, message):
        """Send a query or getmore message and handles the response.

        If message is ``None`` this is an exhaust cursor, which reads
        the next result batch off the exhaust socket instead of
        sending getMore messages to the server.
        """
        client = self.__collection.database.connection

        if message:
            kwargs = {"_must_use_master": self.__must_use_master}
            kwargs["read_preference"] = self.__read_preference
            kwargs["tag_sets"] = self.__tag_sets
            kwargs["secondary_acceptable_latency_ms"] = self.__secondary_acceptable_latency_ms
            kwargs["exhaust"] = self.__exhaust
            if self.__connection_id is not None:
                kwargs["_connection_to_use"] = self.__connection_id
            kwargs.update(self.__kwargs)

            try:
                res = client._send_message_with_response(message, **kwargs)
                self.__connection_id, (response, sock, pool) = res
                if self.__exhaust:
                    self.__exhaust_mgr = _SocketManager(sock, pool)
            except AutoReconnect:
                # Don't try to send kill cursors on another socket
                # or to another server. It can cause a _pinValue
                # assertion on some server releases if we get here
                # due to a socket timeout.
                self.__killed = True
                raise
        else:
            # Exhaust cursor - no getMore message.
            try:
                response = client._exhaust_next(self.__exhaust_mgr.sock)
            except AutoReconnect:
                self.__killed = True
                self.__exhaust_mgr.error()
                raise

        try:
            response = helpers._unpack_response(
                response, self.__id, self.__as_class, self.__tz_aware, self.__uuid_subtype, self.__compile_re
            )
        except OperationFailure:
            self.__killed = True
            # Make sure exhaust socket is returned immediately, if necessary.
            self.__die()
            # If this is a tailable cursor the error is likely
            # due to capped collection roll over. Setting
            # self.__killed to True ensures Cursor.alive will be
            # False. No need to re-raise.
            if self.__query_flags & _QUERY_OPTIONS["tailable_cursor"]:
                return
            raise
        except AutoReconnect:
            # Don't send kill cursors to another server after a "not master"
            # error. It's completely pointless.
            self.__killed = True
            # Make sure exhaust socket is returned immediately, if necessary.
            self.__die()
            client.disconnect()
            raise

        self.__id = response["cursor_id"]

        # starting from doesn't get set on getmore's for tailable cursors
        if not (self.__query_flags & _QUERY_OPTIONS["tailable_cursor"]):
            assert response["starting_from"] == self.__retrieved, "Result batch started from %s, expected %s" % (
                response["starting_from"],
                self.__retrieved,
            )

        self.__retrieved += response["number_returned"]
        self.__data = deque(response["data"])

        if self.__limit and self.__id and self.__limit <= self.__retrieved:
            self.__die()

        # Don't wait for garbage collection to call __del__, return the
        # socket to the pool now.
        if self.__exhaust and self.__id == 0:
            self.__exhaust_mgr.close()

def command(
    sock,
    dbname,
    spec,
    slave_ok,
    is_mongos,
    read_preference,
    codec_options,
    check=True,
    allowable_errors=None,
    address=None,
    user=False,
    check_keys=False,
):
    """Execute a command over the socket, or raise socket.error.

    :Parameters:
      - `sock`: a raw socket instance
      - `dbname`: name of the database on which to run the command
      - `spec`: a command document as a dict, SON, or mapping object
      - `slave_ok`: whether to set the SlaveOkay wire protocol bit
      - `is_mongos`: are we connected to a mongos?
      - `read_preference`: a read preference
      - `codec_options`: a CodecOptions instance
      - `check`: raise OperationFailure if there are errors
      - `allowable_errors`: errors to ignore if `check` is True
      - `address`: the (host, port) of `sock`
      - `user`: is this a user command or internal?
      - `check_keys`: if True, check `spec` for invalid keys
    """
    name = next(iter(spec))
    ns = dbname + ".$cmd"
    flags = 4 if slave_ok else 0
    if is_mongos:
        spec = message._maybe_add_read_preference(spec, read_preference)

    publish = user and monitoring.enabled()
    if publish:
        start = datetime.datetime.now()

    request_id, msg, _ = message.query(flags, ns, 0, -1, spec, None, codec_options, check_keys)

    if publish:
        encoding_duration = datetime.datetime.now() - start
        monitoring.publish_command_start(spec, dbname, request_id, address)
        start = datetime.datetime.now()

    sock.sendall(msg)
    response = receive_message(sock, 1, request_id)
    try:
        unpacked = helpers._unpack_response(response, codec_options=codec_options)

        response_doc = unpacked["data"][0]
        if check:
            msg = "command %s on namespace %s failed: %%s" % (repr(spec).replace("%", "%%"), ns)
            helpers._check_command_response(response_doc, msg, allowable_errors)
    except (NotMasterError, OperationFailure) as exc:
        if publish:
            duration = (datetime.datetime.now() - start) + encoding_duration
            monitoring.publish_command_failure(duration, exc.details, name, request_id, address)
        raise
    if publish:
        duration = (datetime.datetime.now() - start) + encoding_duration
        monitoring.publish_command_success(duration, response_doc, name, request_id, address)
    return response_doc