Method _process_alerts

cmd2/cmd2.py:3594–3639

Background worker that processes queued alerts and dynamic prompt updates.

(self)

        return self._read_raw_input(prompt, temp_session, is_password=True)

    def _process_alerts(self) -> None:
        """Background worker that processes queued alerts and dynamic prompt updates."""
        while True:
            with self._alert_condition:
                # Wait until we have alerts and are allowed to display them, or shutdown is signaled.
                self._alert_condition.wait_for(
                    lambda: (len(self._alert_queue) > 0 and self._alert_allowed) or self._alert_shutdown
                )

                # Shutdown immediately even if we have alerts.
                if self._alert_shutdown:
                    break

                # Hold the condition lock while printing to block command execution. This
                # prevents async alerts from printing once a command starts.

                # Print all alerts at once to reduce flicker.
                alert_text = "\n".join(alert.msg for alert in self._alert_queue if alert.msg)

                # Find the latest prompt update among all pending alerts.
                latest_prompt = None
                for alert in reversed(self._alert_queue):
                    if (
                        alert.prompt is not None
                        and alert.prompt != self.prompt
                        and alert.timestamp > self._alert_prompt_timestamp
                    ):
                        latest_prompt = alert.prompt
                        self._alert_prompt_timestamp = alert.timestamp
                        break

                # Clear the alerts
                self._alert_queue.clear()

                if latest_prompt is not None:
                    # Update prompt so patch_stdout() or get_app().invalidate() can redraw it.
                    self.prompt = latest_prompt

                if alert_text:
                    # Print the alert messages above the prompt.
                    with patch_stdout():
                        print_formatted_text(pt_filter_style(alert_text))

                elif latest_prompt is not None:
                    # Refresh UI immediately to show the new prompt
                    get_app().invalidate()

    def _read_command_line(self, prompt: str) -> str:
        """Read the next command line from the input stream.