Addition of the inbound part of Redis Stream support.

[akuma8]

#3226
This commit is for a first review since tests are not yet added.

[akuma8]

@artembilan,
This is the inbound PR. I am working on adding tests while you are reviewing.
Thanks

[artembilan]

Some review while you're writing tests...

[artembilan]

You should have a ReactiveRedisStreamMessageHandler as a bean. This not a ReactiveMessageHandlerAdapter responsibility to initialize its delegate.

[akuma8]

Hi @artembilan,
I completed outbound side tests by adding the test with the List class.
I have an issue with inbound tests where one of them turn into an infinite loop.
I am not enough familiar with the Reactive world, which explains my problems.
Could you take a look and let me know?
Thanks a lot

[akuma8]

Hi @artembilan,
I did requested changes.
I still have the same issue with my tests.
I think there is an issue with the configuration of streamReceiver.
I breakpointed to check the content of messageFlux (result of the this.streamReceiver.receive()), I can see a StackOverflow exception, see attached image.

I don't kknow what I am doing wrong since I use a default streamReceiver:

StreamReceiver.StreamReceiverOptions<String, ?> streamReceiverOptions = StreamReceiver.StreamReceiverOptions.builder().pollTimeout(Duration.ZERO).build();
this.streamReceiver = StreamReceiver.create(this.reactiveConnectionFactory, this.streamReceiverOptions);

I think there is an issue with the receive() method.

[artembilan]

Pulling locally for review and possible merge after cleaning up...

[artembilan]

Merged as 4ae6b52.

Please, pull the latest master locally to investigate what I did for the code.

I'm really looking forward for more tests since we have fixed the problem with connection factory.
Plus it would be great to have some docs in the redis.adoc on the matter.

Thank you for contribution!

[akuma8]

@artembilan thanks for your investigation.
Did the problem come from the connection factory?
I am working on adding further tests.
About documentation, as I don't speak english it will be hard for me to write a good documentation.
I will try and let you know.
Thank you

[artembilan]

Hi @akuma8 !

Yes, I think the problem for hanging test suite was really about the second call for the ConnectionFactory.afterPropertiesSet() where we created uncontrolled RedisClient.

Sure! More tests are welcome!
Re. Docs: I'm not good in English, too: @garyrussell is always helping me to make my docs PRs better.
So, feel free to raise whatever you can do best and we together will bring it into the public!

Thank you!