text: from __future__ import annotations
  
  import asyncio
  from asyncio import CancelledError
  from asyncio import Queue, QueueEmpty, Task
  import inspect
  from typing import TYPE_CHECKING, Awaitable, Iterable, Callable
  from weakref import WeakSet
  
  from rich.traceback import Traceback
  
  from . import events
  from . import log
  from ._timer import Timer, TimerCallback
  from ._context import active_app
  from .message import Message
  from .reactive import Reactive
  
  
  if TYPE_CHECKING:
      from .app import App
  
  
  class NoParent(Exception):
      pass
  
  
  class CallbackError(Exception):
      pass
  
  
  class MessagePumpClosed(Exception):
      pass
  
  
  class MessagePump:
      def __init__(self, parent: MessagePump | None = None) -> None:
          self._message_queue: Queue[Message | None] = Queue()
          self._parent = parent
          self._running: bool = False
          self._closing: bool = False
          self._closed: bool = False
          self._disabled_messages: set[type[Message]] = set()
          self._pending_message: Message | None = None
          self._task: Task | None = None
          self._child_tasks: WeakSet[Task] = WeakSet()
  
      @property
      def task(self) -> Task:
          assert self._task is not None
          return self._task
  
      @property
      def parent(self) -> MessagePump:
          if self._parent is None:
              raise NoParent(f"{self._parent} has no parent")
          return self._parent
  
      @property
      def app(self) -> "App":
          """Get the current app."""
          return active_app.get()
  
      @property
      def is_parent_active(self):
          return self._parent and not self._parent._closed and not self._parent._closing
  
      @property
      def is_running(self) -> bool:
          return self._running
  
      def log(self, *args) -> None:
          return self.app.log(*args)
  
      def set_parent(self, parent: MessagePump) -> None:
          self._parent = parent
  
      def check_message_enabled(self, message: Message) -> bool:
          return type(message) not in self._disabled_messages
  
      def disable_messages(self, *messages: type[Message]) -> None:
          """Disable message types from being processed."""
          self._disabled_messages.update(messages)
  
      def enable_messages(self, *messages: type[Message]) -> None:
          """Enable processing of messages types."""
          self._disabled_messages.difference_update(messages)
  
      async def get_message(self) -> Message:
          """Get the next event on the queue, or None if queue is closed.
  
          Returns:
              Optional[Event]: Event object or None.
          """
          if self._closed:
              raise MessagePumpClosed("The message pump is closed")
          if self._pending_message is not None:
              try:
                  return self._pending_message
              finally:
                  self._pending_message = None
          message = await self._message_queue.get()
          if message is None:
              self._closed = True
              raise MessagePumpClosed("The message pump is now closed")
          return message
  
      def peek_message(self) -> Message | None:
          """Peek the message at the head of the queue (does not remove it from the queue),
          or return None if the queue is empty.
  
          Returns:
              Optional[Message]: The message or None.
          """
          if self._pending_message is None:
              try:
                  self._pending_message = self._message_queue.get_nowait()
              except QueueEmpty:
                  pass
  
          if self._pending_message is not None:
              return self._pending_message
          return None
  
      def set_timer(
          self,
          delay: float,
          *,
          name: str | None = None,
          callback: TimerCallback = None,
      ) -> Timer:
          timer = Timer(self, delay, self, name=name, callback=callback, repeat=0)
          timer_task = asyncio.get_event_loop().create_task(timer.run())
          self._child_tasks.add(timer_task)
          return timer
  
      def set_interval(
          self,
          interval: float,
          *,
          name: str | None = None,
          callback: TimerCallback = None,
          repeat: int = 0,
      ):
          timer = Timer(
              self, interval, self, name=name, callback=callback, repeat=repeat or None
          )
          self._child_tasks.add(asyncio.get_event_loop().create_task(timer.run()))
          return timer
  
      def close_messages_no_wait(self) -> None:
          self._message_queue.put_nowait(None)
  
      async def close_messages(self, wait: bool = True) -> None:
          """Close message queue, and optionally wait for queue to finish processing."""
          if self._closed:
              return
  
          self._closing = True
  
          await self._message_queue.put(None)
  
          for task in self._child_tasks:
              task.cancel()
              await task
          self._child_tasks.clear()
  
      def start_messages(self) -> None:
          self._task = asyncio.create_task(self.process_messages())
  
      async def process_messages(self) -> None:
          self._running = True
          try:
              return await self._process_messages()
          except CancelledError:
              pass
          finally:
              self._runnning = False
  
      async def _process_messages(self) -> None:
          """Process messages until the queue is closed."""
          _rich_traceback_guard = True
  
          while not self._closed:
              try:
                  message = await self.get_message()
              except MessagePumpClosed:
                  break
              except CancelledError:
                  raise
              except Exception as error:
                  raise error from None
  
              # Combine any pending messages that may supersede this one
              while not (self._closed or self._closing):
                  pending = self.peek_message()
                  if pending is None or not message.can_replace(pending):
                      break
                  try:
                      message = await self.get_message()
                  except MessagePumpClosed:
                      break
  
              try:
                  await self.dispatch_message(message)
              except CancelledError:
                  raise
              except Exception as error:
                  self.app.panic()
                  break
              finally:
                  if isinstance(message, events.Event) and self._message_queue.empty():
                      if not self._closed:
                          event = events.Idle(self)
                          for method in self._get_dispatch_methods("on_idle", event):
                              await method(event)
  
          log("CLOSED", self)
  
      async def dispatch_message(self, message: Message) -> bool | None:
          _rich_traceback_guard = True
          if isinstance(message, events.Event):
              if not isinstance(message, events.Null):
                  await self.on_event(message)
          else:
              return await self.on_message(message)
          return False
  
      def _get_dispatch_methods(
          self, method_name: str, message: Message
      ) -> Iterable[Callable[[Message], Awaitable]]:
          for cls in self.__class__.__mro__:
              if message._no_default_action:
                  break
              method = cls.__dict__.get(method_name, None)
              if method is not None:
                  yield method.__get__(self, cls)
  
      async def on_event(self, event: events.Event) -> None:
          _rich_traceback_guard = True
  
          for method in self._get_dispatch_methods(f"on_{event.name}", event):
              log(event, ">>>", self, verbosity=event.verbosity)
              await method(event)
  
          if event.bubble and self._parent and not event._stop_propagation:
              if event.sender != self._parent and self.is_parent_active:
                  await self._parent.post_message(event)
  
      async def on_message(self, message: Message) -> None:
          _rich_traceback_guard = True
          method_name = f"message_{message.name}"
  
          method = getattr(self, method_name, None)
          if method is not None:
              log(message, ">>>", self, verbosity=message.verbosity)
              await method(message)
  
          if message.bubble and self._parent and not message._stop_propagation:
              if message.sender == self._parent:
                  pass
              elif not self._parent._closed and not self._parent._closing:
                  await self._parent.post_message(message)
  
      def post_message_no_wait(self, message: Message) -> bool:
          if self._closing or self._closed:
              return False
          if not self.check_message_enabled(message):
              return True
          self._message_queue.put_nowait(message)
          return True
  
      async def post_message(self, message: Message) -> bool:
          if self._closing or self._closed:
              return False
          if not self.check_message_enabled(message):
              return True
          await self._message_queue.put(message)
          return True
  
      def post_message_from_child_no_wait(self, message: Message) -> bool:
          if self._closing or self._closed:
              return False
          return self.post_message_no_wait(message)
  
      async def post_message_from_child(self, message: Message) -> bool:
          if self._closing or self._closed:
              return False
          return await self.post_message(message)
  
      async def on_callback(self, event: events.Callback) -> None:
          await event.callback()
  
      def emit_no_wait(self, message: Message) -> bool:
          if self._parent:
              return self._parent.post_message_from_child_no_wait(message)
          else:
              return False
  
      async def emit(self, message: Message) -> bool:
          if self._parent:
              return await self._parent.post_message_from_child(message)
          else:
              return False
  
      async def on_timer(self, event: events.Timer) -> None:
          event.prevent_default()
          event.stop()
          if event.callback is not None:
              try:
                  callback_result = event.callback()
                  if inspect.isawaitable(callback_result):
                      await callback_result
              except Exception as error:
                  raise CallbackError(
                      f"unable to run callback {event.callback!r}; {error}"
                  )
  