async def execute_command(self, *args, **kwargs):
        """
        Execute a publish/subscribe command.

        Taken code from redis-py and tweak to make it work within a cluster.
        """
        # NOTE: don't parse the response in this function -- it could pull a
        # legitimate message off the stack if the connection is already
        # subscribed to one or more channels
        await self.connection_pool.initialize()

        if self.connection is None:
            self.connection = self.connection_pool.get_connection(
                'pubsub',
                channel=args[1],
            )
            # register a callback that re-subscribes to any channels we
            # were listening to when we were disconnected
            self.connection.register_connect_callback(self.on_connect)
        connection = self.connection
        await self._execute(connection, connection.send_command, *args)
async def execute_command ( self , * args , ** kwargs ) : await self . connection_pool . initialize ( ) if self . connection is None : self . connection = self . connection_pool . get_connection ( 'pubsub' , channel = args [ 1 ] , ) self . connection . register_connect_callback ( self . on_connect ) connection = self . connection await self . _execute ( connection , connection . send_command , * args )
Execute a publish/subscribe command.

        Taken code from redis-py and tweak to make it work within a cluster.