def _process_pubsub(self, obj, *, process_waiters=True):
        """Processes pubsub messages."""
        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)
Processes pubsub messages.