Add support for SubscriptionListener using Pub/Sub and introduce ReactiveRedisMessageListenerContainer.receiveLater(…)

[mp911de]

Support for SubscriptionListener when using MessageListener for subscription confirmation callbacks. ReactiveRedisMessageListenerContainer and ReactiveRedisOperations provide receiveLater(…) and listenToLater(…) methods to await until Redis acknowledges the subscription.

A MessageListener can additionally implement SubscriptionListener to receive notifications upon subscription/unsubscribe confirmation. Listening to subscription notifications can be useful when synchronizing invocations.

To await and ensure proper subscription, you can use the receiveLater method that returns a Mono<Flux<ChannelMessage>>.
The resulting Mono completes with an inner publisher as a result of completing the subscription to the given topics. By intercepting onNext signals, you can synchronize server-side subscriptions.

ReactiveRedisConnectionFactory factory = …
ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(factory);

Mono<Flux<ChannelMessage<String, String>>> stream = container.receiveLater(ChannelTopic.of("my-channel"));

stream.doOnNext(inner -> // notification hook when Redis subscriptions are synchronized with the server)
    .flatMapMany(Function.identity())
    .…;

Partially addresses #1622 as this is the required infrastructure to rewrite RedisMessageListenerContainer.start() and addMessageListener(…) into a blocking form.

Closes #2054