code: def _process_pubsub(self, obj, *, process_waiters=True):
          
          kind, *args, data = obj
          if kind in (b'subscribe', b'unsubscribe'):
              chan, = args
              if process_waiters and self._in_pubsub and self._waiters:
                  self._process_data(obj)
              if kind == b'unsubscribe':
                  ch = self._pubsub_channels.pop(chan, None)
                  if ch:
                      ch.close()
              self._in_pubsub = data
          elif kind in (b'psubscribe', b'punsubscribe'):
              chan, = args
              if process_waiters and self._in_pubsub and self._waiters:
                  self._process_data(obj)
              if kind == b'punsubscribe':
                  ch = self._pubsub_patterns.pop(chan, None)
                  if ch:
                      ch.close()
              self._in_pubsub = data
          elif kind == b'message':
              chan, = args
              self._pubsub_channels[chan].put_nowait(data)
          elif kind == b'pmessage':
              pattern, chan = args
              self._pubsub_patterns[pattern].put_nowait((chan, data))
          elif kind == b'pong':
              if process_waiters and self._in_pubsub and self._waiters:
                  self._process_data(data or b'PONG')
          else:
              logger.warning("Unknown pubsub message received %r", obj)
docs: Processes pubsub messages.