(
self, client: AsyncMongoClient[Any], cmd: Optional[MutableMapping[str, Any]]
)
| 189 | self.conn.get_conn.settimeout(timeout) |
| 190 | |
| 191 | def apply_timeout( |
| 192 | self, client: AsyncMongoClient[Any], cmd: Optional[MutableMapping[str, Any]] |
| 193 | ) -> Optional[float]: |
| 194 | # CSOT: use remaining timeout when set. |
| 195 | timeout = _csot.remaining() |
| 196 | if timeout is None: |
| 197 | # Reset the socket timeout unless we're performing a streaming monitor check. |
| 198 | if not self.more_to_come: |
| 199 | self.set_conn_timeout(self.opts.socket_timeout) |
| 200 | return None |
| 201 | # RTT validation. |
| 202 | rtt = _csot.get_rtt() |
| 203 | if rtt is None: |
| 204 | rtt = self.connect_rtt |
| 205 | max_time_ms = timeout - rtt |
| 206 | if max_time_ms < 0: |
| 207 | timeout_details = _get_timeout_details(self.opts) |
| 208 | formatted = format_timeout_details(timeout_details) |
| 209 | # CSOT: raise an error without running the command since we know it will time out. |
| 210 | errmsg = f"operation would exceed time limit, remaining timeout:{timeout:.5f} <= network round trip time:{rtt:.5f} {formatted}" |
| 211 | raise ExecutionTimeout( |
| 212 | errmsg, |
| 213 | 50, |
| 214 | {"ok": 0, "errmsg": errmsg, "code": 50}, |
| 215 | self.max_wire_version, |
| 216 | ) |
| 217 | if cmd is not None: |
| 218 | cmd["maxTimeMS"] = int(max_time_ms * 1000) |
| 219 | self.set_conn_timeout(timeout) |
| 220 | return timeout |
| 221 | |
| 222 | def pin_txn(self) -> None: |
| 223 | self.pinned_txn = True |
no test coverage detected