docs: 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()