Process a new ServerDescription on an opened topology. Hold the lock when calling this.
(
self,
server_description: ServerDescription,
reset_pool: bool = False,
interrupt_connections: bool = False,
)
| 483 | ) |
| 484 | |
| 485 | async def _process_change( |
| 486 | self, |
| 487 | server_description: ServerDescription, |
| 488 | reset_pool: bool = False, |
| 489 | interrupt_connections: bool = False, |
| 490 | ) -> None: |
| 491 | """Process a new ServerDescription on an opened topology. |
| 492 | |
| 493 | Hold the lock when calling this. |
| 494 | """ |
| 495 | td_old = self._description |
| 496 | sd_old = td_old._server_descriptions[server_description.address] |
| 497 | if _is_stale_server_description(sd_old, server_description): |
| 498 | # This is a stale hello response. Ignore it. |
| 499 | return |
| 500 | |
| 501 | new_td = updated_topology_description(self._description, server_description) |
| 502 | # CMAP: Ensure the pool is "ready" when the server is selectable. |
| 503 | if server_description.is_readable or ( |
| 504 | server_description.is_server_type_known and new_td.topology_type == TOPOLOGY_TYPE.Single |
| 505 | ): |
| 506 | server = self._servers.get(server_description.address) |
| 507 | if server: |
| 508 | await server.pool.ready() |
| 509 | |
| 510 | suppress_event = sd_old == server_description |
| 511 | if self._publish_server and not suppress_event: |
| 512 | assert self._events is not None |
| 513 | assert self._listeners is not None |
| 514 | self._events.put( |
| 515 | ( |
| 516 | self._listeners.publish_server_description_changed, |
| 517 | (sd_old, server_description, server_description.address, self._topology_id), |
| 518 | ) |
| 519 | ) |
| 520 | |
| 521 | self._description = new_td |
| 522 | await self._update_servers() |
| 523 | |
| 524 | if self._publish_tp and not suppress_event: |
| 525 | assert self._events is not None |
| 526 | assert self._listeners is not None |
| 527 | self._events.put( |
| 528 | ( |
| 529 | self._listeners.publish_topology_description_changed, |
| 530 | (td_old, self._description, self._topology_id), |
| 531 | ) |
| 532 | ) |
| 533 | if _SDAM_LOGGER.isEnabledFor(logging.DEBUG) and not suppress_event: |
| 534 | _debug_log( |
| 535 | _SDAM_LOGGER, |
| 536 | message=_SDAMStatusMessage.TOPOLOGY_CHANGE, |
| 537 | topologyId=self._topology_id, |
| 538 | previousDescription=repr(td_old), |
| 539 | newDescription=repr(self._description), |
| 540 | ) |
| 541 | |
| 542 | # Shutdown SRV polling for unsupported cluster types. |
no test coverage detected