(self, operation)
Send a query or getmore operation and handles the response.

        If operation 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.

        Can raise ConnectionFailure.
Send a query or getmore operation and handles the response.
Send a query or getmore operation and handles the response .
def __send_message(self, operation):
        """Send a query or getmore operation and handles the response.

        If operation 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.

        Can raise ConnectionFailure.
        """
        client = self.__collection.database.client
        listeners = client._event_listeners
        publish = listeners.enabled_for_commands
        from_command = False

        if operation:
            kwargs = {
                "read_preference": self.__read_preference,
                "exhaust": self.__exhaust,
            }
            if self.__address is not None:
                kwargs["address"] = self.__address

            try:
                response = client._send_message_with_response(operation,
                                                              **kwargs)
                self.__address = response.address
                if self.__exhaust:
                    # 'response' is an ExhaustResponse.
                    self.__exhaust_mgr = _SocketManager(response.socket_info,
                                                        response.pool)

                cmd_name = operation.name
                data = response.data
                cmd_duration = response.duration
                rqst_id = response.request_id
                from_command = response.from_command
            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.
            rqst_id = 0
            cmd_name = 'getMore'
            if publish:
                # Fake a getMore command.
                cmd = SON([('getMore', self.__id),
                           ('collection', self.__collection.name)])
                if self.__batch_size:
                    cmd['batchSize'] = self.__batch_size
                if self.__max_time_ms:
                    cmd['maxTimeMS'] = self.__max_time_ms
                listeners.publish_command_start(
                    cmd, self.__collection.database.name, 0, self.__address)
                start = datetime.datetime.now()
            try:
                data = self.__exhaust_mgr.sock.receive_message(1, None)
            except Exception as exc:
                if publish:
                    duration = datetime.datetime.now() - start
                    listeners.publish_command_failure(
                        duration, _convert_exception(exc), cmd_name, rqst_id,
                        self.__address)
                if isinstance(exc, ConnectionFailure):
                    self.__die()
                raise
            if publish:
                cmd_duration = datetime.datetime.now() - start

        if publish:
            start = datetime.datetime.now()
        try:
            doc = helpers._unpack_response(response=data,
                                           cursor_id=self.__id,
                                           codec_options=self.__codec_options)
            if from_command:
                helpers._check_command_response(doc['data'][0])
        except OperationFailure as exc:
            self.__killed = True

            # Make sure exhaust socket is returned immediately, if necessary.
            self.__die()

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

            # 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 NotMasterError as exc:
            # 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()

            if publish:
                duration = (datetime.datetime.now() - start) + cmd_duration
                listeners.publish_command_failure(
                    duration, exc.details, cmd_name, 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), cmd_name, rqst_id,
                    self.__address)
            raise

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

        if from_command and cmd_name != "explain":
            cursor = doc['data'][0]['cursor']
            self.__id = cursor['id']
            if cmd_name == 'find':
                documents = cursor['firstBatch']
            else:
                documents = cursor['nextBatch']
            self.__data = deque(documents)
            self.__retrieved += len(documents)
        else:
            self.__id = doc["cursor_id"]
            self.__data = deque(doc["data"])
            self.__retrieved += doc["number_returned"]

        if self.__id == 0:
            self.__killed = True


        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 __send_message ( self , operation ) : client = self . __collection . database . client listeners = client . _event_listeners publish = listeners . enabled_for_commands from_command = False if operation : kwargs = { "read_preference" : self . __read_preference , "exhaust" : self . __exhaust , } if self . __address is not None : kwargs [ "address" ] = self . __address try : response = client . _send_message_with_response ( operation , * * kwargs ) self . __address = response . address if self . __exhaust : # 'response' is an ExhaustResponse. self . __exhaust_mgr = _SocketManager ( response . socket_info , response . pool ) cmd_name = operation . name data = response . data cmd_duration = response . duration rqst_id = response . request_id from_command = response . from_command 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. rqst_id = 0 cmd_name = 'getMore' if publish : # Fake a getMore command. cmd = SON ( [ ( 'getMore' , self . __id ) , ( 'collection' , self . __collection . name ) ] ) if self . __batch_size : cmd [ 'batchSize' ] = self . __batch_size if self . __max_time_ms : cmd [ 'maxTimeMS' ] = self . __max_time_ms listeners . publish_command_start ( cmd , self . __collection . database . name , 0 , self . __address ) start = datetime . datetime . now ( ) try : data = self . __exhaust_mgr . sock . receive_message ( 1 , None ) except Exception as exc : if publish : duration = datetime . datetime . now ( ) - start listeners . publish_command_failure ( duration , _convert_exception ( exc ) , cmd_name , rqst_id , self . __address ) if isinstance ( exc , ConnectionFailure ) : self . __die ( ) raise if publish : cmd_duration = datetime . datetime . now ( ) - start if publish : start = datetime . datetime . now ( ) try : doc = helpers . _unpack_response ( response = data , cursor_id = self . __id , codec_options = self . __codec_options ) if from_command : helpers . _check_command_response ( doc [ 'data' ] [ 0 ] ) except OperationFailure as exc : self . __killed = True # Make sure exhaust socket is returned immediately, if necessary. self . __die ( ) if publish : duration = ( datetime . datetime . now ( ) - start ) + cmd_duration listeners . publish_command_failure ( duration , exc . details , cmd_name , rqst_id , self . __address ) # 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 NotMasterError as exc : # 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 ( ) if publish : duration = ( datetime . datetime . now ( ) - start ) + cmd_duration listeners . publish_command_failure ( duration , exc . details , cmd_name , 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 ) , cmd_name , rqst_id , self . __address ) raise if publish : duration = ( datetime . datetime . now ( ) - start ) + cmd_duration # Must publish in find / getMore / explain command response format. if from_command : res = doc [ 'data' ] [ 0 ] elif cmd_name == "explain" : res = doc [ "data" ] [ 0 ] if doc [ "number_returned" ] else { } else : res = { "cursor" : { "id" : doc [ "cursor_id" ] , "ns" : self . __collection . full_name } , "ok" : 1 } if cmd_name == "find" : res [ "cursor" ] [ "firstBatch" ] = doc [ "data" ] else : res [ "cursor" ] [ "nextBatch" ] = doc [ "data" ] listeners . publish_command_success ( duration , res , cmd_name , rqst_id , self . __address ) if from_command and cmd_name != "explain" : cursor = doc [ 'data' ] [ 0 ] [ 'cursor' ] self . __id = cursor [ 'id' ] if cmd_name == 'find' : documents = cursor [ 'firstBatch' ] else : documents = cursor [ 'nextBatch' ] self . __data = deque ( documents ) self . __retrieved += len ( documents ) else : self . __id = doc [ "cursor_id" ] self . __data = deque ( doc [ "data" ] ) self . __retrieved += doc [ "number_returned" ] if self . __id == 0 : self . __killed = True 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 ( )