Outbound channel adapter for Redis Stream #3259

akuma8 · 2020-04-23T17:58:27Z

This is the outbound side of that PR: #3227 related to #3226
TODO:
I have to add more tests cases.

artembilan · 2020-04-23T18:30:18Z

Glad to see you back! Hope you feel good and ready to go 😄

So, thank you first of all for your effort!
Not sure that we can make it into the current 5.3 version, but it is definitely will go to the next one.

There bear with us if we don't merge it soon enough.

Please, take a look into Travis report:

/home/travis/build/spring-projects/spring-integration/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandler.java:52: warning: [rawtypes] found raw type: ReactiveStreamOperations

	private ReactiveStreamOperations reactiveStreamOperations;

We are against rawtypes - should be fixed. Even if it is just ? or Object it has to be fixed respectively.
We probably can go with the @SuppressWarnings("rawtypes") as a final resort.

artembilan

Are there any chances to see real tests that after calling this ReactiveRedisStreamMessageHandler we get some data in the Redis?

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

...a/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java

akuma8 · 2020-05-05T18:51:02Z

Hi @artembilan !
I am finally back.
Thanks for your reviews, I make changes and let you know.

...a/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java

akuma8 · 2020-05-17T04:24:56Z

Hi @artembilan,
I noticed that ReactiveRedisTemplate has this method:

ReactiveStreamOperations<K, HK, HV> opsForStream( RedisSerializationContext<K, ?> serializationContext, @Nullable HashMapper<? super K, ? super HK, ? super HV> hashMapper)

Do you think that I should add a org.springframework.data.redis.hash.HashMapper property to the ReactiveRedisStreamMessageHandler to let the client customize the HashMapper? If not provided we'll use the default one.

artembilan · 2020-05-18T19:37:09Z

o you think that I should add a org.springframework.data.redis.hash.HashMapper property

Yes, sounds reasonable!

artembilan · 2020-05-19T19:15:40Z

Please, rebase your branch to the latest master. It is now 5.4.x.
We will review your effort shortly after that.
Thanks

akuma8 · 2020-05-21T18:47:12Z

@artembilan rebase done.
The CI failed because of sftp tests failing.
Thanks

akuma8 · 2020-05-22T19:06:34Z

@artembilan There are 2 tests that I am unable to make them pass, the first one is:

	public void explicitJsonSerializationContextTest() {
		String streamKey = "myStream";
		List<String> messagePayload = Arrays.asList("Bonjour", "à", "tous", "les", "confinés");

		RedisSerializationContext<String, Object> jsonSerializationContext = redisStringOrJsonSerializationContext(false, List.class);

		ReactiveRedisStreamMessageHandler streamMessageHandler = new ReactiveRedisStreamMessageHandler(streamKey, redisConnectionFactory);
		streamMessageHandler.setSerializationContext(jsonSerializationContext);
		
               //initializes reactiveRedisStreamOperations
		invokeOnInitMethod(streamMessageHandler);

		ReactiveMessageHandlerAdapter handlerAdapter = new ReactiveMessageHandlerAdapter(streamMessageHandler);
		handlerAdapter.handleMessage(new GenericMessage<>(messagePayload));

		ReactiveRedisTemplate<String, String> template = new ReactiveRedisTemplate(redisConnectionFactory,
				jsonSerializationContext);
		ObjectRecord<String, List> record = template.opsForStream().read(List.class, StreamOffset.fromStart(streamKey))
				.blockFirst();

		assertThat(record.getStream()).isEqualTo(streamKey);
		assertThat(record.getValue()).isEqualTo("[\"Bonjour\", \"à\", \"tous\", \"les\", \"confinés\"]");
		template.delete(streamKey).block();
	}

The goal is to override the default RedisSerializationContext by setting explicitly another one with a Json serializer. To verify that I defined this list
List<String> messagePayload = Arrays.asList("Bonjour", "à", "tous", "les", "confinés");
as payload. As I use a Json serializer I expect this result:
[\"Bonjour\", \"à\", \"tous\", \"les\", \"confinés\"] but I have a strange result when reading data from the stream, it is not deserialized.

The 2nd test is similar to the 1st one but instead of a list of strings I use this model RedisMessageStoreTests.Person as payload:

	public void explicitJsonSerializationContextWithModelTest() {
		String streamKey = "myStream";

		RedisMessageStoreTests.Address address = new RedisMessageStoreTests.Address().withAddress("Rennes, France");
		RedisMessageStoreTests.Person person = new RedisMessageStoreTests.Person(address, "Attoumane");

		Message message = new GenericMessage(person);

		RedisSerializationContext<String, Object> jsonSerializationContext = redisStringOrJsonSerializationContext(false, RedisMessageStoreTests.Person.class);

		ReactiveRedisStreamMessageHandler streamMessageHandler = new ReactiveRedisStreamMessageHandler(streamKey, redisConnectionFactory);
		streamMessageHandler.setSerializationContext(jsonSerializationContext);
		
                //initializes reactiveRedisStreamOperations
		invokeOnInitMethod(streamMessageHandler);

		ReactiveMessageHandlerAdapter handlerAdapter = new ReactiveMessageHandlerAdapter(streamMessageHandler);
		handlerAdapter.handleMessage(message);

		ReactiveRedisTemplate<String, String> template = new ReactiveRedisTemplate(redisConnectionFactory, jsonSerializationContext);
		ObjectRecord<String, RedisMessageStoreTests.Person> record = template.opsForStream().read(RedisMessageStoreTests.Person.class, StreamOffset.fromStart(streamKey)).blockFirst();

		assertThat(record.getStream()).isEqualTo(streamKey);
		assertThat(record.getValue().getName()).isEqualTo("Attoumane");
		assertThat(record.getValue().getAddress().getAddress()).isEqualTo("Rennes, France");
		template.delete(streamKey).block();
	}

The result is the same, data are not deserialized. If we solve this, the inboud side will be simpler. Please let me know if you find the cause.

Thanks a lot

artembilan

Here is a review for the current state of code.
I'll take a look into your failing tests when you address this review.
Right now the change is a bit confusing especially in test configuration, so highly possible that something is wrong with your tests.
Please, come back to me with fixes or objection and we will figure out the rest together.

Thank you!

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

...pringframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests-context.xml

...a/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java

akuma8 · 2020-06-10T10:03:15Z

Hi @artembilan,
Do you know why sftp tests often fail when I push on this branch?

> Task :spring-integration-sftp:test
SftpStreamingMessageSourceTests > testMaxFetchLambdaFilter() FAILED
    org.springframework.messaging.MessagingException at SftpStreamingMessageSourceTests.java:159
        Caused by: org.springframework.integration.util.PoolItemNotAvailableException at SftpStreamingMessageSourceTests.java:159
            Caused by: java.lang.IllegalStateException at SftpStreamingMessageSourceTests.java:159
                Caused by: java.lang.IllegalStateException at SftpStreamingMessageSourceTests.java:159
                    Caused by: com.jcraft.jsch.JSchException at SftpStreamingMessageSourceTests.java:159

artembilan · 2020-06-10T18:16:49Z

Hi @akuma8 !

Thank you for coming back with some update!

Today I'm busy with releases, but going to take a look into your effort tomorrow in the morning.

Do you know why sftp tests often fail when I push on this branch?

Well, it really fails even for me. Not sure yet what is going on. Some race condition, probably...
If it bother you too much, you can disable/ignore it for time being.

artembilan

Here is some review.

Also, please, consider to fix the RedisAvailableRule. The base.evaluate(); should go after that catch (Exception e) { with Assume.assumeTrue().

Thank you!

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

...pringframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTestsContext.java

...a/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java

akuma8 · 2020-06-12T15:02:10Z

@artembilan Thanks for your reviews. I made requested changes. I also change my tests, I use a RedisSerializationContext which simply use a StringRedisSerializer. I don't understand why with its Json counterpart nothing works. Also even with that RedisSerializationContext there is a test that does not pass. This one ReactiveRedisStreamMessageHandlerTests#explicitSerializationContextTest().

template.opsForStream().read(List.class, StreamOffset.fromStart(ReactiveRedisStreamMessageHandlerTestsContext.STREAM_KEY))
				.blockFirst();

fails.
Have a look and please let me know.
Thanks a gain

akuma8 · 2020-06-22T12:49:00Z

Hi @artembilan,
I don't know why but all redis tests fail during the CI build. On my machine I don't have these issues, all tests pass.

ReactiveRedisStreamMessageHandlerTests > explicitSerializationContextWithModelTest FAILED
    org.springframework.data.redis.RedisSystemException at LettuceExceptionConverter.java:54
        Caused by: io.lettuce.core.RedisCommandExecutionException at ExceptionFactory.java:135
ReactiveRedisStreamMessageHandlerTests > integrationStreamOutboundTest FAILED
    org.springframework.data.redis.RedisSystemException at LettuceExceptionConverter.java:54
        Caused by: io.lettuce.core.RedisCommandExecutionException at ExceptionFactory.java:135
> Task :spring-integration-scripting:test
> Task :spring-integration-redis:test FAILED
RedisQueueMessageDrivenEndpointTests > testInt3932ReadFromLeft FAILED
    org.opentest4j.MultipleFailuresError at TestRun.java:196
RedisQueueMessageDrivenEndpointTests > testInt3014Default FAILED
    org.opentest4j.MultipleFailuresError at TestRun.java:196
RedisQueueMessageDrivenEndpointTests > testInt3014ExpectMessageTrue FAILED
    org.opentest4j.MultipleFailuresError at TestRun.java:196
RedisQueueMessageDrivenEndpointTests > testInt3017IntegrationSymmetrical FAILED
    org.opentest4j.MultipleFailuresError at TestRun.java:196
RedisQueueMessageDrivenEndpointTests > testInt3017IntegrationInbound FAILED
    org.opentest4j.MultipleFailuresError at TestRun.java:196
RedisQueueMessageDrivenEndpointTests > testInt3442ProperlyStop FAILED
    org.opentest4j.MultipleFailuresError at TestRun.java:196

Let me know if I have to make further changes.
Thanks

artembilan · 2020-06-22T15:06:07Z

I don't know why but all redis tests fail during the CI build. On my machine I don't have these issues, all tests pass.

That is probably because Travis uses an old Redis version unlike your local machine.

I recently installed Linux sub-system onto my Windows 10 and use the latest Redis 6.0.5 from there: works really well.

So, let's don't bother ourselves for a while!

I'll review your PR today: very likely this one is going to make it into the release this Wednesday.

Thank you for keeping me busy! 😉

artembilan

@akuma8 ,

I have just fixed failed Redis tests and that RedisAvailableRule: 9495685

Please, rebase your PR and let me know how it works for you and probably we are ready to merge this.

Thanks

artembilan · 2020-06-23T20:28:39Z

...a/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java

+
+
+	/*@Test
+	@RedisAvailable*/


Please, let me know if this test works well

I rebased but still have the same issue, the .opsForStream().read(....) method returns null.

java.lang.IllegalArgumentException: Value must not be null! at org.springframework.util.Assert.notNull(Assert.java:201) at org.springframework.data.redis.connection.stream.Record.of(Record.java:81) at org.springframework.data.redis.connection.stream.MapRecord.toObjectRecord(MapRecord.java:147) at org.springframework.data.redis.core.StreamObjectMapper.toObjectRecord(StreamObjectMapper.java:132) at org.springframework.data.redis.core.ReactiveStreamOperations.lambda$read$4(ReactiveStreamOperations.java:426) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100) at reactor.core.publisher.FluxUsingWhen$UsingWhenSubscriber.onNext(FluxUsingWhen.java:355) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242) at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73) at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) at io.lettuce.core.RedisPublisher$ImmediateSubscriber.onNext(RedisPublisher.java:886) at io.lettuce.core.RedisPublisher$RedisSubscription.onNext(RedisPublisher.java:279) at io.lettuce.core.output.StreamingOutput$Subscriber.onNext(StreamingOutput.java:64) at io.lettuce.core.output.StreamReadOutput.complete(StreamReadOutput.java:92) at io.lettuce.core.protocol.RedisStateMachine.decode(RedisStateMachine.java:401) at io.lettuce.core.protocol.CommandHandler.decode(CommandHandler.java:707) at io.lettuce.core.protocol.CommandHandler.decode0(CommandHandler.java:671) at io.lettuce.core.protocol.CommandHandler.decode(CommandHandler.java:666) at io.lettuce.core.protocol.CommandHandler.decode(CommandHandler.java:587) at io.lettuce.core.protocol.CommandHandler.channelRead(CommandHandler.java:556) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:714) at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:650) at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:576) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493) at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) at java.lang.Thread.run(Thread.java:748) Suppressed: java.lang.Exception: #block terminated with an error at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:99) at reactor.core.publisher.Flux.blockFirst(Flux.java:2452) at org.springframework.integration.redis.outbound.ReactiveRedisStreamMessageHandlerTests.explicitSerializationContextTest(ReactiveRedisStreamMessageHandlerTests.java:119) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50) at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12) at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47) at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17) at org.springframework.test.context.junit4.statements.RunBeforeTestExecutionCallbacks.evaluate(RunBeforeTestExecutionCallbacks.java:74) at org.springframework.test.context.junit4.statements.RunAfterTestExecutionCallbacks.evaluate(RunAfterTestExecutionCallbacks.java:84) at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:75) at org.springframework.test.context.junit4.statements.RunAfterTestMethodCallbacks.evaluate(RunAfterTestMethodCallbacks.java:86) at org.springframework.integration.redis.rules.RedisAvailableRule$1.evaluate(RedisAvailableRule.java:86) at org.springframework.test.context.junit4.statements.SpringRepeat.evaluate(SpringRepeat.java:84) at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325) at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:251) at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97) at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290) at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71) at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288) at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58) at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268) at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61) at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70) at org.junit.runners.ParentRunner.run(ParentRunner.java:363) at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190) at org.junit.runner.JUnitCore.run(JUnitCore.java:137) at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68) at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47) at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242) at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)

I checked the stream and it well contains data:

1) "1593030360926-0" 2) 1) "_class" 2) "java.util.Arrays$ArrayList" 3) "a.[0]._class" 4) "java.lang.String" 5) "a.[0]" 6) "Hello" 7) "a.[1]._class" 8) "java.lang.String" 9) "a.[1]" 10) "stream" 11) "a.[2]._class" 12) "java.lang.String" 13) "a.[2]" 14) "message" 15) "modCount" 16) "0"

It seems like reading List causes issue.
Thanks for your help

It really likely a List.class as a return type is not possible from that read(....).

I think we need to disable this test for time being.

…shing message to the actual stream using ReactiveStreamOperations

Some tests with explicit RedisSerializationContext fail.

…eactiveRedisStreamMessageHandlerTests#explicitSerializationContextTest

artembilan · 2020-06-24T21:45:51Z

...a/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java

+
+
+	/*@Test
+	@RedisAvailable*/


It really likely a List.class as a return type is not possible from that read(....).

I think we need to disable this test for time being.

artembilan

Please run gradlew check Task locally to see Checkstyle problems. Looks like we are good to merge this soon enough!
👍

artembilan · 2020-06-25T18:20:10Z

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

+ *
+ * @since 5.4
+ */
+@SuppressWarnings({"rawtypes", "unchecked"})


May we kinda get rid of rawtypes?
Probably simple wildcard for genercis (<?>) should be enough for us...

Removing rawtypes causes compilation issue:

error: no suitable method found for opsForStream(HashMapper<CAP#1,CAP#2,CAP#3>) template.opsForStream(this.hashMapper); ^ method ReactiveRedisOperations.<HK#1,HV#1>opsForStream() is not applicable (cannot infer type-variable(s) HK#1,HV#1 (actual and formal argument lists differ in length))

on

this.reactiveStreamOperations = this.hashMapper == null ? template.opsForStream() : template.opsForStream(this.hashMapper);

OK. Let me play with that locally!
Then I'll merge it since we need something for today's release.

You feel free to go ahead with Inbound Channel Adapter implementation and some docs for both of this.

Thanks

artembilan · 2020-06-25T18:23:21Z

...a/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java

+
+		ReactiveRedisTemplate<String, String> template = new ReactiveRedisTemplate(redisConnectionFactory, serializationContext);
+
+		ObjectRecord<String, String> record = template.opsForStream().read(String.class, StreamOffset.fromStart(ReactiveRedisStreamMessageHandlerTestsContext.STREAM_KEY)).blockFirst();


Please, consider to not exceed 120 line length: much easier to review.

artembilan · 2020-06-26T15:30:42Z

Merged as afa79d8 after some clean up.

Please, review carefully what I did with your code in the end.
Would be great to be consistent in the Inbound part.
Also don't hesitate to raise docs PR on the matter.

Thank you and looking forward for more!

akuma8 · 2020-06-29T08:44:45Z

@artembilan Thanks for merging. I reviewed your modifications which seem cleaner than what I did.
I need this feature in my application, that's why I'm stubbornly trying to implement it. It's not easy since it's my first contrinution. I'll let you know when I'll start the inbound part.
Thanks for your help.

akuma8 force-pushed the gh-3226-outbound branch from bdc6c65 to 6b7fca5 Compare April 23, 2020 18:05

akuma8 changed the title ~~Ooutbound channel adapter for Redis Stream~~ Outbound channel adapter for Redis Stream Apr 23, 2020

artembilan requested changes Apr 23, 2020

View reviewed changes

akuma8 force-pushed the gh-3226-outbound branch from ae17008 to e6e53ea Compare May 9, 2020 21:41

akuma8 commented May 10, 2020

View reviewed changes

...a/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java Outdated Show resolved Hide resolved

akuma8 commented May 10, 2020

View reviewed changes

...a/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java Outdated Show resolved Hide resolved

akuma8 force-pushed the gh-3226-outbound branch 2 times, most recently from ce00199 to c14ae67 Compare May 10, 2020 15:52

artembilan added this to the 5.4.x milestone May 12, 2020

akuma8 requested a review from artembilan May 17, 2020 04:25

akuma8 force-pushed the gh-3226-outbound branch 2 times, most recently from 2935f7e to 4a26f09 Compare May 21, 2020 18:35

artembilan requested changes May 26, 2020

View reviewed changes

akuma8 force-pushed the gh-3226-outbound branch 3 times, most recently from 8929515 to d404aa8 Compare June 6, 2020 13:23

akuma8 requested a review from artembilan June 10, 2020 10:03

artembilan requested changes Jun 11, 2020

View reviewed changes

akuma8 requested a review from artembilan June 12, 2020 15:02

akuma8 force-pushed the gh-3226-outbound branch from d87cf4e to 9e8c463 Compare June 12, 2020 17:21

artembilan requested changes Jun 23, 2020

View reviewed changes

akuma8 added 5 commits June 24, 2020 21:39

Redis stream message handler support. This is the outbound part publi…

0bd13db

…shing message to the actual stream using ReactiveStreamOperations

Addition of more test cases with one using MessageChannel.

242375b

Some tests with explicit RedisSerializationContext fail.

Improvements after PR review.

f10f7d4

Improvements after PR review. Also only one test does not pass, see R…

e44930a

…eactiveRedisStreamMessageHandlerTests#explicitSerializationContextTest

Still have to review one failing test

a0f9770

akuma8 force-pushed the gh-3226-outbound branch from 9e8c463 to a0f9770 Compare June 24, 2020 20:59

artembilan requested changes Jun 24, 2020

View reviewed changes

artembilan requested changes Jun 25, 2020

View reviewed changes

Removed failed test reading List from a Stream

739edc0

akuma8 force-pushed the gh-3226-outbound branch from 030633e to 739edc0 Compare June 25, 2020 19:35

artembilan closed this Jun 26, 2020

artembilan modified the milestones: 5.4.x, 5.4 M1 Jun 26, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Outbound channel adapter for Redis Stream #3259

Outbound channel adapter for Redis Stream #3259

akuma8 commented Apr 23, 2020

artembilan commented Apr 23, 2020

artembilan left a comment

akuma8 commented May 5, 2020

akuma8 commented May 17, 2020

artembilan commented May 18, 2020

artembilan commented May 19, 2020

akuma8 commented May 21, 2020 •

edited

Loading

akuma8 commented May 22, 2020

artembilan left a comment

akuma8 commented Jun 10, 2020

artembilan commented Jun 10, 2020

artembilan left a comment

akuma8 commented Jun 12, 2020 •

edited

Loading

akuma8 commented Jun 22, 2020

artembilan commented Jun 22, 2020

artembilan left a comment

artembilan Jun 23, 2020

akuma8 Jun 24, 2020 •

edited

Loading

artembilan Jun 24, 2020

artembilan Jun 24, 2020

artembilan left a comment

artembilan Jun 25, 2020

akuma8 Jun 25, 2020

artembilan Jun 26, 2020

artembilan Jun 25, 2020

artembilan commented Jun 26, 2020

akuma8 commented Jun 29, 2020 •

edited

Loading


		ReactiveRedisTemplate<String, String> template = new ReactiveRedisTemplate(redisConnectionFactory, serializationContext);

		ObjectRecord<String, String> record = template.opsForStream().read(String.class, StreamOffset.fromStart(ReactiveRedisStreamMessageHandlerTestsContext.STREAM_KEY)).blockFirst();

Outbound channel adapter for Redis Stream #3259

Outbound channel adapter for Redis Stream #3259

Conversation

akuma8 commented Apr 23, 2020

artembilan commented Apr 23, 2020

artembilan left a comment

Choose a reason for hiding this comment

akuma8 commented May 5, 2020

akuma8 commented May 17, 2020

artembilan commented May 18, 2020

artembilan commented May 19, 2020

akuma8 commented May 21, 2020 • edited Loading

akuma8 commented May 22, 2020

artembilan left a comment

Choose a reason for hiding this comment

akuma8 commented Jun 10, 2020

artembilan commented Jun 10, 2020

artembilan left a comment

Choose a reason for hiding this comment

akuma8 commented Jun 12, 2020 • edited Loading

akuma8 commented Jun 22, 2020

artembilan commented Jun 22, 2020

artembilan left a comment

Choose a reason for hiding this comment

artembilan Jun 23, 2020

Choose a reason for hiding this comment

akuma8 Jun 24, 2020 • edited Loading

Choose a reason for hiding this comment

artembilan Jun 24, 2020

Choose a reason for hiding this comment

artembilan Jun 24, 2020

Choose a reason for hiding this comment

artembilan left a comment

Choose a reason for hiding this comment

artembilan Jun 25, 2020

Choose a reason for hiding this comment

akuma8 Jun 25, 2020

Choose a reason for hiding this comment

artembilan Jun 26, 2020

Choose a reason for hiding this comment

artembilan Jun 25, 2020

Choose a reason for hiding this comment

artembilan commented Jun 26, 2020

akuma8 commented Jun 29, 2020 • edited Loading

akuma8 commented May 21, 2020 •

edited

Loading

akuma8 commented Jun 12, 2020 •

edited

Loading

akuma8 Jun 24, 2020 •

edited

Loading

akuma8 commented Jun 29, 2020 •

edited

Loading