WIP : Channel Adapters for Redis Streams #3227

akuma8 · 2020-03-24T15:29:09Z

#3226
The PR is in WIP because I need information for improvements.

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

...n/java/org/springframework/integration/redis/outbound/RedisStreamOutboundChannelAdapter.java

artembilan

I hope you find my superficial review sufficient and we are good to continue a work here.

Thank you for your effort!

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

...n/java/org/springframework/integration/redis/outbound/RedisStreamOutboundChannelAdapter.java

artembilan · 2020-03-27T18:24:35Z

@akuma8 ,

please, don't squash commits when you push them to the PR: it is going to be much easier for us to review when change history is present.
Also it would be great to comment the changes in commit: kinda what has been done and the reason.
It also helps us to review faster.

See more info about good commits in Chris Beams' article: https://chris.beams.io/posts/git-commit/

Thanks for understanding.

P.S. for now it is OK because PR is small yet, but it is going to grow I believe.

akuma8 · 2020-03-27T19:10:40Z

@artembilan Thank you for the remark, I will be careful in the future. It's a personal habit.
I let you a comment because I am stuck, please let me know when you will read it.

artembilan · 2020-03-27T19:15:50Z

please let me know when you will read it.

I'm in review process.
Stay tuned. BTW not reason to be in hurry: we are OK to release it whenever it is ready.

artembilan

You know I have just discovered for myself a ReactiveStreamOperations in Spring Data Redis.
I my experience it is better to develop Reactive component and let it to be used in the blocking scenarios.

Can we revise your solution in favor of ReactiveStreamOperations?
See ReactiveMongoDbStoringMessageHandler for example.

The RedisStreamInboundChannelAdapter could also be based on the ReactiveStreamOperations.read(). The returned Flux could be damped into a ReactiveStreamsSubscribableChannel if configured that way or as regular send() otherwise.

Let me know WDYT?

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

.../src/main/java/org/springframework/integration/redis/outbound/RedisStreamMessageHandler.java

akuma8

Can we revise your solution in favor of ReactiveStreamOperations?

Sure, I saw ReactiveStreamOperations in the doc but I thought it only be relevant to use it in a full reactive context which is not our case but I will revise my first solution if this one is better.

...n/java/org/springframework/integration/redis/outbound/RedisStreamOutboundChannelAdapter.java

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

...n/java/org/springframework/integration/redis/outbound/RedisStreamOutboundChannelAdapter.java

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

akuma8 · 2020-04-02T20:40:37Z

@artembilan I tried the reactive approach without success. I followed this https://docs.spring.io/spring-data/redis/docs/2.3.0.M4/reference/html/#redis.streams.receive.containers with StreamReceiver but when I publish a message to the stream it is not received in the inbound side. At application startup messages are well read but not at runtime. I didn't have that issue with the blocking approach.
When talking about ReactiveStreamOperations I was'nt sure if you also want the outbound side to be reactive so I didn't change it. I just modified the inbound side. Please let me know if I have to change it too.
I need other hints to solve the inbound issue, I don't know why I don't receive messages.
Thanks a lot

artembilan · 2020-04-03T15:26:45Z

Hi, @akuma8 !

Thank you for update! I'll review your solution and try to figure problems the next week.
Meanwhile the answer to your question: yes, I want an outbound part as reactive as well. You need to implement a ReactiveMessageHandler over there. See a ReactiveMongoDbStoringMessageHandler for example.

Right now I'm working on some reactive streams functionality for the MessageProducerSupport. I somehow believe your inbound channel adapter would benefit from that.
Let's see the next week how it is going!

akuma8 · 2020-04-03T19:11:53Z

Hi @artembilan,
Thanks for this feedback. I modified the outbound to take account your remark (see the 2nd commit) and I renamed it to ReactiveRedisStreamMessageHandler. As usual the outbound side works as expected but it's not the case for the inbound. Please let me know when your reactive streams functionality will be available.
Thanks

artembilan · 2020-04-05T01:56:33Z

I have just raised a PR #3240 about reactive Message Producer.
I have over there an implementation for MongoDb change stream.
I think that one should be a sample how this Redis Streams has to be implemented on the inbound side.
Let's hope it will be merged next week, so you would have some foundation to rely on!

artembilan

@akuma8 ,

tell, me, please, how that would be hard for you to divide this PR into two.
One for already working functionality on the outbound side.
And then we can come back to be inbound one when PR #3240 is merged.
This way we won't have everything blocked.

Thank for understanding.

akuma8 · 2020-04-09T11:56:50Z

Hi @artembilan
Since Monday I am sick, it’s not the Covid-19 😁.
It’s a good idea to split this PR in 2, I’ll do that next week, I hope getting better.

artembilan · 2020-04-09T14:23:47Z

Stay safe, @akuma8 , and get well soon!

No need to be in a hurry: looks like no pressure for the feature yet. So, we are good to bring it whenever it is ready.

When you come back from your sick time off, please, divide this into two PRs - inbound & outbound.
Please, make sure you contribute some tests for this new channel adapters.

At a glance MessageHandler part is OK - should be merged shortly after an appropriate PR.

I haven't looked into an inbound part because wanted to be sure that MessageProducerSupport.subscribeToPublisher(Publisher<Message<?>>) approach is correct.

Now it is merged to master and you can start looking into the MongoDbChangeStreamMessageProducer to make your inbound part similar way.

Fingers crossed for your health!

artembilan

Hi @akuma8 !

Any chances to see an update from you on the matter?

What we have so far and what we still need to do to have that Redis Stream Inbound Channel Adapter?

Thanks

akuma8 · 2020-07-16T09:31:23Z

Hi @artembilan,
I am a bit busy but I started looking at the inbound part, I hope opening a first PR next week.
I also opened that issue about reading List from a Redis Stream
I'll let you know.
Thanks

artembilan · 2020-07-16T14:21:21Z

Hi @akuma8 !

No problem with your load!

Just let me know that you can make it before September or should I pull this task from you and fix myself!
Great to see the fix in Spring Data Redis. I think it's time reinstate your test for list. Nothing more we need to do on the matter.
See a related PR in Spring Data: they have some tests to investigate.

akuma8 · 2020-07-16T18:21:31Z

@artembilan,

... you can make it before September...

Yes, I think so. If we keep doing it like we did for the outbound side I can make it before September.
I am also looking for a way to test the fix from Spring Data Redis, if you can include the related version, please let me know.

artembilan · 2020-07-17T19:42:09Z

if you can include the related version, please let me know.

Just did: ebec500

Upgraded to Spring Data 2020.0.0-SNAPSHOT

akuma8 · 2020-07-21T14:46:58Z

Hi @artembilan,
I have some questions about implementation choices.
Redis Stream Api provides 2 ways to read data, we can read as a simple client with the XREAD command or as a client belonging to a Consumer Group: XREADGROUP. Should we provide one inbound implementation including the 2 options or devide them in 2 different implementations?
Also, should the implementations use the reactive model (as we did for the outbound side)? If yes, using MongoDbChangeStreamMessageProducer as an example is still valid?
Thanks a lot

artembilan · 2020-07-21T15:19:15Z

Yes, if that is possible, I'd prefer to have a one inbound component with a consumerGroup property (optional).
And yes, it would be better to have it as reactive. I think the MongoDbChangeStreamMessageProducer is a good sample to borrow ideas.

Thanks

- Changed the imperative `StreamMessageListenerContainer` listener in favor of the reactive one `StreamReceiver` - Added RedisHeaders values for Redis Stream Outbound: - Added a new constructor taking `Expression` as parameter - Changed the imperative `StreamOperations` in favor of `ReactiveStreamOperations` - Renaming class from `RedisStreamMessageHandler` to `ReactiveRedisStreamMessageHandler` Took into account other improvement remarks: copyright, setters position, methods names

artembilan

This is good progress, but definitely there is still some work to do: tests, docs, JavaDocs etc.

Thanks

artembilan · 2020-07-28T16:41:36Z

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

@@ -0,0 +1,171 @@
+/*
+ * Copyright 2020-2021 the original author or authors.


I hope we can merge this still in 2020 😄

artembilan · 2020-07-28T16:42:06Z

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

+import org.springframework.util.StringUtils;
+
+/**
+ * Read Message from a Redis Stream and publish it to the indicated output channel.


@since 5.4 and your name to @author

artembilan · 2020-07-28T16:44:10Z

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

+	private Message<?> convertMessage( Record<Object, Object> record ) {
+		Map<String, Object> headers = new HashMap<>();
+		headers.put( RedisHeaders.STREAM_KEY, record.getStream() );
+		headers.put( RedisHeaders.STREAM_MESSAGE_ID, record.getId() );


We probably need also populate a consumerGroup header.

artembilan · 2020-07-28T16:47:04Z

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

+		Map<String, Object> headers = new HashMap<>();
+		headers.put( RedisHeaders.STREAM_KEY, record.getStream() );
+		headers.put( RedisHeaders.STREAM_MESSAGE_ID, record.getId() );
+		return this.messageConverter.toMessage( record.getValue(), new MessageHeaders( headers ) );


Hm. Doesn't look like we do something fancy with the converter. What is wrong with just using a plain getMessageBuilderFactory()?

artembilan · 2020-07-28T16:48:30Z

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

+		this.createGroupIfNotExist = createGroupIfNotExist;
+	}
+
+	public void setReceiver( StreamReceiver receiver ) {


Why do we need this option?
Looks like you create it in the registerListener() anyway.

artembilan · 2020-07-28T16:54:39Z

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

+		this.readOffset = readOffset;
+	}
+
+	void setTemplate( RedisTemplate template ) {


Why do we need to set this one?
Why can't we create it based on the provided connection factory?

artembilan · 2020-07-28T16:55:29Z

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

+		this.consumerName = consumerName;
+	}
+
+	public void setCreateGroupIfNotExist( boolean createGroupIfNotExist ) {


I would just call it setCreateGroup() and explain a behavior in JavaDocs.

artembilan · 2020-07-28T16:59:03Z

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

+			}
+			catch ( Exception e ) {
+				// An exception is thrown when the group already exists
+				e.printStackTrace();


Probably better to just log INFO that group already exist.
Anyway it's probably better to look into a ReactiveRedisTemplate and perform the creation operation together with subsequent receiver as a reactive stream composition.

artembilan · 2020-07-28T16:59:42Z

...egration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java

@@ -45,4 +45,8 @@ private RedisHeaders() {

 	public static final String MESSAGE_SOURCE = PREFIX + "messageSource";

+	public static final String STREAM_KEY = PREFIX + "streamKey";
+
+	public static final String STREAM_MESSAGE_ID = PREFIX + "streamMessageID";


Maybe streamMessageId? No ID, please.

artembilan · 2020-07-28T17:04:01Z

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java

+			Assert.hasText( consumerName, "'consumerName' must be set" );
+			Consumer consumer = Consumer.from( this.consumerGroupName, this.consumerName );
+			this.receiver
+					.receiveAutoAck( consumer, offset )


Looks like we also need to think about manual acknowledgement:

* Every record must be acknowledged using * {@link org.springframework.data.redis.connection.ReactiveStreamCommands#xAck(ByteBuffer, String, String...)}

See receive() API for the StreamReceiver.
This has to be populated as a IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK.
And also see SimpleAcknowledgment abstraction to wrap the call to that xAck. for the particular record returned.

Saying that I'd expect an autoAck(true|false) option on this channel adapter.

akuma8 · 2020-07-28T17:57:07Z

Hi @artembilan,
Sorry but I forgot to tell you to not review this, I just rebased master branch.
I planned to delete this PR and opening a new one. Like I did for the outbound.

artembilan · 2020-07-28T17:58:46Z

OK. So, I'm closing this one to avoid confusion.

Thanks for confirmation!

akuma8 commented Mar 24, 2020

View reviewed changes

...ain/java/org/springframework/integration/redis/inbound/RedisStreamInboundChannelAdapter.java Outdated Show resolved Hide resolved

akuma8 commented Mar 24, 2020

View reviewed changes

...n/java/org/springframework/integration/redis/outbound/RedisStreamOutboundChannelAdapter.java Outdated Show resolved Hide resolved

artembilan requested changes Mar 24, 2020

View reviewed changes

akuma8 force-pushed the gh-3226 branch from 4f4f43e to 88c7dbe Compare March 27, 2020 18:20

artembilan requested changes Mar 27, 2020

View reviewed changes

akuma8 commented Mar 27, 2020

View reviewed changes

akuma8 force-pushed the gh-3226 branch from 60a05d3 to 7372f18 Compare April 3, 2020 18:50

artembilan requested changes Apr 6, 2020

View reviewed changes

akuma8 added a commit to akuma8/spring-integration that referenced this pull request Apr 23, 2020

Outbound side of this PR: spring-projects#3227

bdc6c65

akuma8 mentioned this pull request Apr 23, 2020

Outbound channel adapter for Redis Stream #3259

Closed

artembilan requested changes Jul 13, 2020

View reviewed changes

akuma8 force-pushed the gh-3226 branch 2 times, most recently from 8deed7a to 9c482ec Compare July 28, 2020 16:32

akuma8 added 2 commits July 28, 2020 18:36

Adding Channel Adapters for Redis Streams

79fdd55

akuma8 force-pushed the gh-3226 branch from 9c482ec to 6320804 Compare July 28, 2020 16:38

artembilan requested changes Jul 28, 2020

View reviewed changes

artembilan closed this Jul 28, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

WIP : Channel Adapters for Redis Streams #3227

WIP : Channel Adapters for Redis Streams #3227

akuma8 commented Mar 24, 2020

artembilan left a comment

artembilan commented Mar 27, 2020

akuma8 commented Mar 27, 2020

artembilan commented Mar 27, 2020

artembilan left a comment

akuma8 left a comment •

edited

Loading

akuma8 commented Apr 2, 2020

artembilan commented Apr 3, 2020

akuma8 commented Apr 3, 2020

artembilan commented Apr 5, 2020

artembilan left a comment

akuma8 commented Apr 9, 2020

artembilan commented Apr 9, 2020

artembilan left a comment

akuma8 commented Jul 16, 2020 •

edited

Loading

artembilan commented Jul 16, 2020

akuma8 commented Jul 16, 2020

artembilan commented Jul 17, 2020

akuma8 commented Jul 21, 2020 •

edited

Loading

artembilan commented Jul 21, 2020

artembilan left a comment

artembilan Jul 28, 2020

artembilan Jul 28, 2020

artembilan Jul 28, 2020

artembilan Jul 28, 2020

artembilan Jul 28, 2020

artembilan Jul 28, 2020

artembilan Jul 28, 2020

artembilan Jul 28, 2020

artembilan Jul 28, 2020

artembilan Jul 28, 2020

akuma8 commented Jul 28, 2020 •

edited

Loading

artembilan commented Jul 28, 2020

		@@ -0,0 +1,171 @@
		/*
		* Copyright 2020-2021 the original author or authors.

WIP : Channel Adapters for Redis Streams #3227

WIP : Channel Adapters for Redis Streams #3227

Conversation

akuma8 commented Mar 24, 2020

artembilan left a comment

Choose a reason for hiding this comment

artembilan commented Mar 27, 2020

akuma8 commented Mar 27, 2020

artembilan commented Mar 27, 2020

artembilan left a comment

Choose a reason for hiding this comment

akuma8 left a comment • edited Loading

Choose a reason for hiding this comment

akuma8 commented Apr 2, 2020

artembilan commented Apr 3, 2020

akuma8 commented Apr 3, 2020

artembilan commented Apr 5, 2020

artembilan left a comment

Choose a reason for hiding this comment

akuma8 commented Apr 9, 2020

artembilan commented Apr 9, 2020

artembilan left a comment

Choose a reason for hiding this comment

akuma8 commented Jul 16, 2020 • edited Loading

artembilan commented Jul 16, 2020

akuma8 commented Jul 16, 2020

artembilan commented Jul 17, 2020

akuma8 commented Jul 21, 2020 • edited Loading

artembilan commented Jul 21, 2020

artembilan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akuma8 commented Jul 28, 2020 • edited Loading

artembilan commented Jul 28, 2020

akuma8 left a comment •

edited

Loading

akuma8 commented Jul 16, 2020 •

edited

Loading

akuma8 commented Jul 21, 2020 •

edited

Loading

akuma8 commented Jul 28, 2020 •

edited

Loading