class_docstr: Extend Dask's scheduler with routes to handle PubSub machinery
code: def remove_subscriber(self, name=None, worker=None, client=None):
      if worker:
          logger.debug("Remove worker subscriber: %s %s", name, worker)
          self.subscribers[name].remove(worker)
          for pub in self.publishers[name]:
              self.scheduler.worker_send(
                  pub,
                  {
                      "op": "pubsub-remove-subscriber",
                      "address": worker,
                      "name": name,
                  },
              )
      elif client:
          logger.debug("Remove client subscriber: %s %s", name, client)
          self.client_subscribers[name].remove(client)
          if not self.client_subscribers[name]:
              del self.client_subscribers[name]
              for pub in self.publishers[name]:
                  self.scheduler.worker_send(
                      pub,
                      {
                          "op": "pubsub-publish-scheduler",
                          "name": name,
                          "publish": False,
                      },
                  )
  
      if not self.subscribers[name] and not self.publishers[name]:
          logger.debug("Remove PubSub topic %s", name)
          del self.subscribers[name]
          del self.publishers[name]
  
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 5 for calling the PubSubSchedulerExtension obj's underlying member method code with example usage: obj.remove_subscriber(name, worker, client) without return types