Get or create a AsyncConnection. Can raise ConnectionFailure.
(
self, checkout_started_time: float, handler: Optional[_MongoClientErrorHandler] = None
)
| 1199 | ) |
| 1200 | |
| 1201 | async def _get_conn( |
| 1202 | self, checkout_started_time: float, handler: Optional[_MongoClientErrorHandler] = None |
| 1203 | ) -> AsyncConnection: |
| 1204 | """Get or create a AsyncConnection. Can raise ConnectionFailure.""" |
| 1205 | # We use the pid here to avoid issues with fork / multiprocessing. |
| 1206 | # See test.test_client:TestClient.test_fork for an example of |
| 1207 | # what could go wrong otherwise |
| 1208 | if self.pid != os.getpid(): |
| 1209 | await self.reset_without_pause() |
| 1210 | |
| 1211 | if self.closed: |
| 1212 | duration = time.monotonic() - checkout_started_time |
| 1213 | if self.enabled_for_cmap: |
| 1214 | assert self.opts._event_listeners is not None |
| 1215 | self.opts._event_listeners.publish_connection_check_out_failed( |
| 1216 | self.address, ConnectionCheckOutFailedReason.POOL_CLOSED, duration |
| 1217 | ) |
| 1218 | if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG): |
| 1219 | _debug_log( |
| 1220 | _CONNECTION_LOGGER, |
| 1221 | message=_ConnectionStatusMessage.CHECKOUT_FAILED, |
| 1222 | clientId=self._client_id, |
| 1223 | serverHost=self.address[0], |
| 1224 | serverPort=self.address[1], |
| 1225 | reason="Connection pool was closed", |
| 1226 | error=ConnectionCheckOutFailedReason.POOL_CLOSED, |
| 1227 | durationMS=duration, |
| 1228 | ) |
| 1229 | raise _PoolClosedError( |
| 1230 | "Attempted to check out a connection from closed connection pool" |
| 1231 | ) |
| 1232 | |
| 1233 | async with self.lock: |
| 1234 | self.operation_count += 1 |
| 1235 | |
| 1236 | # Get a free socket or create one. |
| 1237 | if _csot.get_timeout(): |
| 1238 | deadline = _csot.get_deadline() |
| 1239 | elif self.opts.wait_queue_timeout: |
| 1240 | deadline = time.monotonic() + self.opts.wait_queue_timeout |
| 1241 | else: |
| 1242 | deadline = None |
| 1243 | |
| 1244 | async with self.size_cond: |
| 1245 | self._raise_if_not_ready(checkout_started_time, emit_event=True) |
| 1246 | while not (self.requests < self.max_pool_size): |
| 1247 | timeout = deadline - time.monotonic() if deadline else None |
| 1248 | if not await _async_cond_wait(self.size_cond, timeout): |
| 1249 | # Timed out, notify the next thread to ensure a |
| 1250 | # timeout doesn't consume the condition. |
| 1251 | if self.requests < self.max_pool_size: |
| 1252 | self.size_cond.notify() |
| 1253 | self._raise_wait_queue_timeout(checkout_started_time) |
| 1254 | self._raise_if_not_ready(checkout_started_time, emit_event=True) |
| 1255 | self.requests += 1 |
| 1256 | |
| 1257 | # We've now acquired the semaphore and must release it on error. |
| 1258 | conn = None |
no test coverage detected