Close this connection with a reason.
(self, reason: Optional[str])
| 543 | ) |
| 544 | |
| 545 | async def close_conn(self, reason: Optional[str]) -> None: |
| 546 | """Close this connection with a reason.""" |
| 547 | if self.closed: |
| 548 | return |
| 549 | await self._close_conn() |
| 550 | if reason: |
| 551 | if self.enabled_for_cmap: |
| 552 | assert self.listeners is not None |
| 553 | self.listeners.publish_connection_closed(self.address, self.id, reason) |
| 554 | if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG): |
| 555 | _debug_log( |
| 556 | _CONNECTION_LOGGER, |
| 557 | message=_ConnectionStatusMessage.CONN_CLOSED, |
| 558 | clientId=self._client_id, |
| 559 | serverHost=self.address[0], |
| 560 | serverPort=self.address[1], |
| 561 | driverConnectionId=self.id, |
| 562 | reason=_verbose_connection_error_reason(reason), |
| 563 | error=reason, |
| 564 | ) |
| 565 | |
| 566 | async def _close_conn(self) -> None: |
| 567 | """Close this connection.""" |
no test coverage detected