Send a kill cursors message with the given ids.
(
self,
cursor_ids: Sequence[int],
address: Optional[_CursorAddress],
topology: Topology,
session: Optional[ClientSession],
)
| 2195 | self._close_cursor_soon(cursor_id, address) |
| 2196 | |
| 2197 | def _kill_cursors( |
| 2198 | self, |
| 2199 | cursor_ids: Sequence[int], |
| 2200 | address: Optional[_CursorAddress], |
| 2201 | topology: Topology, |
| 2202 | session: Optional[ClientSession], |
| 2203 | ) -> None: |
| 2204 | """Send a kill cursors message with the given ids.""" |
| 2205 | if address: |
| 2206 | # address could be a tuple or _CursorAddress, but |
| 2207 | # select_server_by_address needs (host, port). |
| 2208 | server = topology.select_server_by_address(tuple(address), _Op.KILL_CURSORS) # type: ignore[arg-type] |
| 2209 | else: |
| 2210 | # Application called close_cursor() with no address. |
| 2211 | server = topology.select_server(writable_server_selector, _Op.KILL_CURSORS) |
| 2212 | |
| 2213 | with self._checkout(server, session) as conn: |
| 2214 | assert address is not None |
| 2215 | self._kill_cursor_impl(cursor_ids, address, session, conn) |
| 2216 | |
| 2217 | def _kill_cursor_impl( |
| 2218 | self, |
no test coverage detected