class_docstr: Subscribe to a Publish/Subscribe topic
  
  See Also
  --------
  Pub: for full docstring
code: def __init__(self, name, worker=None, client=None):
      if worker is None and client is None:
          from distributed.worker import get_client, get_worker
  
          try:
              worker = get_worker()
          except Exception:
              client = get_client()
  
      self.worker = worker
      self.client = client
      if self.worker:
          self.loop = self.worker.loop
      elif self.client:
          self.loop = self.client.loop
      self.name = name
      self.buffer = deque()
  
      if self.worker:
          pubsub = self.worker.extensions["pubsub"]
      elif self.client:
          pubsub = self.client.extensions["pubsub"]
      self.loop.add_callback(pubsub.subscribers[name].add, self)
  
      msg = {"op": "pubsub-add-subscriber", "name": self.name}
      if self.worker:
          self.loop.add_callback(self.worker.batched_stream.send, msg)
      elif self.client:
          self.loop.add_callback(self.client.scheduler_comm.send, msg)
      else:
          raise Exception()
  
      weakref.finalize(self, pubsub.trigger_cleanup)
  
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 Sub object  with example: obj = Sub(name, worker, client)