class_docstr: Extend Dask's scheduler with routes to handle PubSub machinery
code: def __init__(self, scheduler):
      self.scheduler = scheduler
      self.publishers = defaultdict(set)
      self.subscribers = defaultdict(set)
      self.client_subscribers = defaultdict(set)
  
      self.scheduler.handlers.update(
          {"pubsub_add_publisher": self.add_publisher}
      )
  
      self.scheduler.stream_handlers.update(
          {
              "pubsub-add-subscriber": self.add_subscriber,
              "pubsub-remove-publisher": self.remove_publisher,
              "pubsub-remove-subscriber": self.remove_subscriber,
              "pubsub-msg": self.handle_message,
          }
      )
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import threading
  import weakref
  from collections import defaultdict, deque
  from dask.utils import parse_timedelta
  from distributed.core import CommClosedError
  from distributed.metrics import time
  from distributed.protocol.serialize import to_serialize
  from distributed.utils import TimeoutError, sync, wait_for
label_desc: Use image node_id 1 to create a new PubSubSchedulerExtension object  with example: obj = PubSubSchedulerExtension(scheduler)