class_docstr: Extend Dask's scheduler with routes to handle PubSub machinery
code: def handle_message(
      self, name=None, msg=None, worker=None, client=None
  ):
      for c in list(self.client_subscribers[name]):
          try:
              self.scheduler.client_comms[c].send(
                  {"op": "pubsub-msg", "name": name, "msg": msg}
              )
          except (KeyError, CommClosedError):
              self.remove_subscriber(name=name, client=c)
  
      if client:
          for sub in self.subscribers[name]:
              self.scheduler.worker_send(
                  sub, {"op": "pubsub-msg", "name": name, "msg": msg}
              )
  
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 6 for calling the PubSubSchedulerExtension obj's underlying member method code with example usage: obj.handle_message(name, msg, worker, client) without return types