class RedisPubSubReactiveCommandsImpl<K, V> extends RedisReactiveCommandsImpl<K, V>
        implements RedisPubSubReactiveCommands<K, V> {

    private final PubSubCommandBuilder<K, V> commandBuilder;

    /**
     * Initialize a new connection.
     *
     * @param connection the connection .
     * @param codec Codec used to encode/decode keys and values.
     */
    public RedisPubSubReactiveCommandsImpl(StatefulRedisPubSubConnection<K, V> connection, RedisCodec<K, V> codec) {
        super(connection, codec);
        this.commandBuilder = new PubSubCommandBuilder<>(codec);
    }

    @Override
    public Flux<PatternMessage<K, V>> observePatterns() {
        return observePatterns(FluxSink.OverflowStrategy.BUFFER);
    }

    @Override
    public Flux<PatternMessage<K, V>> observePatterns(FluxSink.OverflowStrategy overflowStrategy) {

        return Flux.create(sink -> {

            RedisPubSubAdapter<K, V> listener = new RedisPubSubAdapter<K, V>() {

                @Override
                public void message(K pattern, K channel, V message) {
                    sink.next(new PatternMessage<>(pattern, channel, message));
                }

            };

            StatefulRedisPubSubConnection<K, V> statefulConnection = getStatefulConnection();
            statefulConnection.addListener(listener);

            sink.onDispose(() -> {
                statefulConnection.removeListener(listener);
            });

        }, overflowStrategy);
    }

    @Override
    public Flux<ChannelMessage<K, V>> observeChannels() {
        return observeChannels(FluxSink.OverflowStrategy.BUFFER);
    }

    @Override
    public Flux<ChannelMessage<K, V>> observeChannels(FluxSink.OverflowStrategy overflowStrategy) {<FILL_FUNCTION_BODY>}

    @Override
    public Mono<Void> psubscribe(K... patterns) {
        return createMono(() -> commandBuilder.psubscribe(patterns)).then();
    }

    @Override
    public Mono<Void> punsubscribe(K... patterns) {
        return createFlux(() -> commandBuilder.punsubscribe(patterns)).then();
    }

    @Override
    public Mono<Void> subscribe(K... channels) {
        return createFlux(() -> commandBuilder.subscribe(channels)).then();
    }

    @Override
    public Mono<Void> unsubscribe(K... channels) {
        return createFlux(() -> commandBuilder.unsubscribe(channels)).then();
    }

    @Override
    public Mono<Long> publish(K channel, V message) {
        return createMono(() -> commandBuilder.publish(channel, message));
    }

    @Override
    public Flux<K> pubsubChannels(K channel) {
        return createDissolvingFlux(() -> commandBuilder.pubsubChannels(channel));
    }

    @Override
    public Mono<Map<K, Long>> pubsubNumsub(K... channels) {
        return createMono(() -> commandBuilder.pubsubNumsub(channels));
    }

    @Override
    public Flux<K> pubsubShardChannels(K channel) {
        return createDissolvingFlux(() -> commandBuilder.pubsubShardChannels(channel));
    }

    @Override
    public Mono<Map<K, Long>> pubsubShardNumsub(K... shardChannels) {
        return createMono(() -> commandBuilder.pubsubShardNumsub(shardChannels));
    }

    @Override
    public Mono<Void> ssubscribe(K... shardChannels) {
        return createFlux(() -> commandBuilder.ssubscribe(shardChannels)).then();
    }

    @Override
    @SuppressWarnings("unchecked")
    public StatefulRedisPubSubConnection<K, V> getStatefulConnection() {
        return (StatefulRedisPubSubConnection<K, V>) super.getStatefulConnection();
    }

}


        return Flux.create(sink -> {

            RedisPubSubAdapter<K, V> listener = new RedisPubSubAdapter<K, V>() {

                @Override
                public void message(K channel, V message) {
                    sink.next(new ChannelMessage<>(channel, message));
                }

            };

            StatefulRedisPubSubConnection<K, V> statefulConnection = getStatefulConnection();
            statefulConnection.addListener(listener);

            sink.onDispose(() -> {
                statefulConnection.removeListener(listener);
            });

        }, overflowStrategy);
    
<methods>public void <init>(StatefulRedisConnection<K,V>, RedisCodec<K,V>) ,public StatefulRedisConnection<K,V> getStatefulConnection() <variables>