From 4ae6b52c001e3e06c334eec8370c8dd96d3b0fe4 Mon Sep 17 00:00:00 2001 From: Attoumane Ahamadi Date: Tue, 28 Jul 2020 14:03:23 -0400 Subject: [PATCH] GH-3226: Add ReactiveRedisStreamMessageProducer Fixes https://github.com/spring-projects/spring-integration/issues/3226 * Implement a `ReactiveRedisStreamMessageProducer` to consume Redis streams * Add support for consumer groups and group auto-creation * Add `@author` to `RedisHeaders` * Rename test methods in the `ReactiveRedisStreamMessageHandlerTests` to project code style * Implement `hashCode()` & `equals()` for `Person` & `Address` testing models * Fix `ReactiveRedisStreamMessageProducerTests` not creating bean for `ConnectionFactory`. Otherwise it caused to create one more not controlled `RedisClient` * Code style clean up in the `ReactiveRedisStreamMessageProducer` * Implement a group creation logic as a reactive stream deferring the call until a subscription happens on the final `messageFlux` * Move a common code for message building as the last `map()` operator in the final `Flux` * Remove an `IntegrationFlow` definition in the `ReactiveRedisStreamMessageProducerTests` as redundant --- .../ReactiveRedisStreamMessageProducer.java | 232 ++++++++++++++++++ .../redis/support/RedisHeaders.java | 11 +- ...activeRedisStreamMessageProducerTests.java | 174 +++++++++++++ ...eactiveRedisStreamMessageHandlerTests.java | 49 ++-- .../integration/redis/util/Address.java | 15 ++ .../integration/redis/util/Person.java | 17 ++ 6 files changed, 474 insertions(+), 24 deletions(-) create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java new file mode 100644 index 0000000000..c1a3b49447 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java @@ -0,0 +1,232 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.redis.inbound; + +import java.time.Duration; + +import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; +import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.Record; +import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.core.ReactiveRedisTemplate; +import org.springframework.data.redis.core.ReactiveStreamOperations; +import org.springframework.data.redis.core.StreamOperations; +import org.springframework.data.redis.serializer.RedisSerializationContext; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.data.redis.stream.StreamReceiver; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.redis.support.RedisHeaders; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * A {@link MessageProducerSupport} for reading messages from a Redis Stream and publishing them into the provided + * output channel. + * By default this adapter reads message as a standalone client {@code XREAD} (Redis command) but can be switched to a + * Consumer Group feature {@code XREADGROUP} by setting {@link #consumerName} field. + * By default the Consumer Group name is an id of this bean {@link #getBeanName()}. + * + * @author Attoumane Ahamadi + * @author Artem Bilan + * + * @since 5.4 + */ +public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport { + + private final ReactiveRedisConnectionFactory reactiveConnectionFactory; + + private final String streamKey; + + private ReactiveStreamOperations reactiveStreamOperations; + + private StreamReceiver.StreamReceiverOptions streamReceiverOptions = + StreamReceiver.StreamReceiverOptions.builder() + .pollTimeout(Duration.ZERO) + .build(); + + private StreamReceiver streamReceiver; + + private ReadOffset readOffset = ReadOffset.latest(); + + private boolean extractPayload = true; + + private boolean autoAck = true; + + @Nullable + private String consumerGroup; + + @Nullable + private String consumerName; + + private boolean createConsumerGroup; + + public ReactiveRedisStreamMessageProducer(ReactiveRedisConnectionFactory reactiveConnectionFactory, + String streamKey) { + + Assert.notNull(reactiveConnectionFactory, "'connectionFactory' must not be null"); + Assert.hasText(streamKey, "'streamKey' must be set"); + this.reactiveConnectionFactory = reactiveConnectionFactory; + this.streamKey = streamKey; + } + + /** + * Define the offset from which we want to read message. By default the {@link ReadOffset#latest()} is used. + * {@link ReadOffset#latest()} is equal to '$', which is the Id used with {@code XREAD} to get new data added to + * the stream. Note that when switching to the Consumer Group feature, we set it to + * {@link ReadOffset#lastConsumed()} if it is still equal to {@link ReadOffset#latest()}. + * @param readOffset the desired offset + */ + public void setReadOffset(ReadOffset readOffset) { + this.readOffset = readOffset; + } + + /** + * Configure this channel adapter to extract or not the message payload. + * @param extractPayload default true + */ + public void setExtractPayload(boolean extractPayload) { + this.extractPayload = extractPayload; + } + + /** + * Set whether or not acknowledge message read in the Consumer Group. {@code true} by default. + * @param autoAck the acknowledge option. + */ + public void setAutoAck(boolean autoAck) { + this.autoAck = autoAck; + } + + /** + * Set the name of the Consumer Group. It is possible to create that Consumer Group if desired, see: + * {@link #createConsumerGroup}. If not set, the defined bean name {@link #getBeanName()} is used. + * @param consumerGroup the Consumer Group on which this adapter should register to listen messages. + */ + public void setConsumerGroup(@Nullable String consumerGroup) { + this.consumerGroup = consumerGroup; + } + + /** + * Set the name of the consumer. When a consumer name is provided, this adapter is switched to the Consumer Group + * feature. Note that this value should be unique in the group. + * @param consumerName the consumer name in the Consumer Group + */ + public void setConsumerName(@Nullable String consumerName) { + this.consumerName = consumerName; + } + + /** + * Create the Consumer Group if and only if it does not exist. + * During the creation we also create the stream, see {@code MKSTREAM}. + * @param createConsumerGroup specify if we should create the Consumer Group, {@code false} by default + */ + public void setCreateConsumerGroup(boolean createConsumerGroup) { + this.createConsumerGroup = createConsumerGroup; + } + + /** + * Set {@link StreamOperations} used to customize the {@link StreamReceiver}. + * It provides a way to set the polling timeout and the serialization context. + * By default the polling timeout is set to infinite and {@link StringRedisSerializer} is used. + * @param streamReceiverOptions the desired receiver options + * */ + public void setStreamReceiverOptions( + @Nullable StreamReceiver.StreamReceiverOptions streamReceiverOptions) { + + this.streamReceiverOptions = streamReceiverOptions; + } + + @Override + public String getComponentType() { + return "redis:stream-inbound-channel-adapter"; + } + + @Override + protected void onInit() { + super.onInit(); + this.streamReceiver = StreamReceiver.create(this.reactiveConnectionFactory, this.streamReceiverOptions); + if (StringUtils.hasText(this.consumerName) && StringUtils.isEmpty(this.consumerGroup)) { + this.consumerGroup = getBeanName(); + } + ReactiveRedisTemplate reactiveRedisTemplate = + new ReactiveRedisTemplate<>(this.reactiveConnectionFactory, RedisSerializationContext.string()); + this.reactiveStreamOperations = reactiveRedisTemplate.opsForStream(); + } + + @Override + protected void doStart() { + super.doStart(); + + StreamOffset offset = StreamOffset.create(this.streamKey, this.readOffset); + + Flux> events; + + if (StringUtils.isEmpty(this.consumerName)) { + events = this.streamReceiver.receive(offset); + } + else { + Mono consumerGroupMono = Mono.empty(); + if (this.createConsumerGroup) { + consumerGroupMono = + this.reactiveStreamOperations.createGroup(this.streamKey, this.consumerGroup) + .onErrorReturn(this.consumerGroup); + } + + Consumer consumer = Consumer.from(this.consumerGroup, this.consumerName); + + if (offset.getOffset().equals(ReadOffset.latest())) { + // for consumer group offset id should be equal '>' + offset = StreamOffset.create(this.streamKey, ReadOffset.lastConsumed()); + } + + events = + this.autoAck + ? this.streamReceiver.receiveAutoAck(consumer, offset) + : this.streamReceiver.receive(consumer, offset); + + events = consumerGroupMono.thenMany(events); + + } + + Flux> messageFlux = + events.map((event) -> { + AbstractIntegrationMessageBuilder builder = + getMessageBuilderFactory() + .withPayload(this.extractPayload ? event.getValue() : event) + .setHeader(RedisHeaders.STREAM_KEY, event.getStream()) + .setHeader(RedisHeaders.STREAM_MESSAGE_ID, event.getId()) + .setHeader(RedisHeaders.CONSUMER_GROUP, this.consumerGroup) + .setHeader(RedisHeaders.CONSUMER, this.consumerName); + if (!this.autoAck) { + builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, + this.reactiveStreamOperations.acknowledge(this.consumerGroup, event) + .subscribe()); + } + return builder.build(); + }); + + subscribeToPublisher(messageFlux); + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java index 28d62788d5..87bdb0119f 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,6 +23,7 @@ package org.springframework.integration.redis.support; * @author Oleg Zhurakousky * @author Gary Russell * @author Artem Bilan + * @author Attoumane Ahamadi * * @since 2.2 */ @@ -45,4 +46,12 @@ public final class 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"; + + public static final String CONSUMER_GROUP = PREFIX + "consumerGroup"; + + public static final String CONSUMER = PREFIX + "consumer"; + } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java new file mode 100644 index 0000000000..65749b5e55 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java @@ -0,0 +1,174 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.redis.inbound; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Duration; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.core.ReactiveRedisTemplate; +import org.springframework.data.redis.serializer.RedisSerializationContext; +import org.springframework.data.redis.stream.StreamReceiver; +import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.integration.handler.ReactiveMessageHandlerAdapter; +import org.springframework.integration.redis.outbound.ReactiveRedisStreamMessageHandler; +import org.springframework.integration.redis.rules.RedisAvailable; +import org.springframework.integration.redis.rules.RedisAvailableRule; +import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.redis.util.Address; +import org.springframework.integration.redis.util.Person; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +import reactor.core.publisher.BaseSubscriber; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +/** + * @author Attoumane Ahamadi + * @author Artem Bilan + * + * @since 5.4 + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests { + + private static final String STREAM_KEY = "myStream"; + + private static final String CONSUMER = "consumer"; + + @Autowired + FluxMessageChannel fluxMessageChannel; + + @Autowired + ReactiveRedisStreamMessageProducer redisStreamMessageProducer; + + @Autowired + ReactiveRedisTemplate template; + + @Autowired + ReactiveMessageHandlerAdapter messageHandler; + + @Before + public void delKey() { + this.template.delete(STREAM_KEY).block(); + } + + @After + public void tearDown() { + this.redisStreamMessageProducer.stop(); + } + + @Test + @RedisAvailable + public void testConsumerGroupCreation() { + this.redisStreamMessageProducer.setCreateConsumerGroup(true); + this.redisStreamMessageProducer.setConsumerName(CONSUMER); + this.redisStreamMessageProducer.afterPropertiesSet(); + this.redisStreamMessageProducer.start(); + + Flux.from(this.fluxMessageChannel).subscribe(); + + this.template.opsForStream() + .groups(STREAM_KEY) + .next() + .as(StepVerifier::create) + .assertNext((infoGroup) -> + assertThat(infoGroup.groupName()).isEqualTo(this.redisStreamMessageProducer.getBeanName())) + .thenCancel() + .verify(); + } + + @Test + @RedisAvailable + public void testReadingMessageAsStandaloneClient() { + Address address = new Address("Rennes 3, France"); + Person person = new Person(address, "Attoumane"); + this.messageHandler.handleMessage(new GenericMessage<>(person)); + + this.redisStreamMessageProducer.setCreateConsumerGroup(false); + this.redisStreamMessageProducer.setConsumerName(null); + this.redisStreamMessageProducer.setReadOffset(ReadOffset.from("0")); + this.redisStreamMessageProducer.afterPropertiesSet(); + this.redisStreamMessageProducer.start(); + + Flux.from(this.fluxMessageChannel) + .as(StepVerifier::create) + .assertNext(message -> assertThat(message.getPayload()).isEqualTo(person)) + .thenCancel() + .verify(Duration.ofSeconds(10)); + } + + @Test + @RedisAvailable + public void testReadingMessageAsConsumerInConsumerGroup() { + //TODO find why the test above does not execute before implementing this one + } + + @Configuration + static class ContextConfig { + + @Bean + ReactiveRedisStreamMessageHandler redisStreamMessageHandler() { + return new ReactiveRedisStreamMessageHandler(RedisAvailableRule.connectionFactory, STREAM_KEY); + } + + @Bean + public ReactiveMessageHandlerAdapter reactiveMessageHandlerAdapter() { + return new ReactiveMessageHandlerAdapter(redisStreamMessageHandler()); + } + + @Bean + ReactiveRedisTemplate reactiveStreamOperations() { + return new ReactiveRedisTemplate<>(RedisAvailableRule.connectionFactory, + RedisSerializationContext.string()); + } + + @Bean + FluxMessageChannel fluxMessageChannel() { + return new FluxMessageChannel(); + } + + @Bean + ReactiveRedisStreamMessageProducer reactiveRedisStreamProducer() { + ReactiveRedisStreamMessageProducer messageProducer = + new ReactiveRedisStreamMessageProducer(RedisAvailableRule.connectionFactory, STREAM_KEY); + messageProducer.setStreamReceiverOptions( + StreamReceiver.StreamReceiverOptions.builder() + .pollTimeout(Duration.ofMillis(100)) + .targetType(Person.class) + .build()); + messageProducer.setAutoStartup(false); + messageProducer.setOutputChannel(fluxMessageChannel()); + return messageProducer; + } + + } + +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java index 91eb67e89f..7896cfa2fa 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java @@ -18,6 +18,9 @@ package org.springframework.integration.redis.outbound; import static org.assertj.core.api.Assertions.assertThat; +import java.util.Arrays; +import java.util.List; + import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -31,7 +34,6 @@ import org.springframework.data.redis.connection.stream.ObjectRecord; import org.springframework.data.redis.connection.stream.StreamOffset; import org.springframework.data.redis.core.ReactiveRedisTemplate; import org.springframework.data.redis.serializer.RedisSerializationContext; -import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.handler.ReactiveMessageHandlerAdapter; import org.springframework.integration.redis.rules.RedisAvailable; @@ -67,9 +69,6 @@ public class ReactiveRedisStreamMessageHandlerTests extends RedisAvailableTests @Autowired private ReactiveMessageHandlerAdapter handlerAdapter; - @Autowired - private ReactiveRedisStreamMessageHandler streamMessageHandler; - @Before public void deleteStreamKey() { ReactiveRedisTemplate template = new ReactiveRedisTemplate<>(this.redisConnectionFactory, @@ -80,15 +79,13 @@ public class ReactiveRedisStreamMessageHandlerTests extends RedisAvailableTests @Test @RedisAvailable - public void integrationStreamOutboundTest() { + public void testIntegrationStreamOutbound() { String messagePayload = "Hello stream message"; - messageChannel.send(new GenericMessage<>(messagePayload)); - - RedisSerializationContext serializationContext = redisSerializationContext(); + this.messageChannel.send(new GenericMessage<>(messagePayload)); ReactiveRedisTemplate template = - new ReactiveRedisTemplate<>(redisConnectionFactory, serializationContext); + new ReactiveRedisTemplate<>(this.redisConnectionFactory, RedisSerializationContext.string()); ObjectRecord record = template.opsForStream() @@ -102,21 +99,33 @@ public class ReactiveRedisStreamMessageHandlerTests extends RedisAvailableTests @Test @RedisAvailable - public void explicitSerializationContextWithModelTest() { + public void testMessageWithListPayload() { + List messagePayload = Arrays.asList("Hello", "stream", "message"); + this.handlerAdapter.handleMessage(new GenericMessage<>(messagePayload)); + + ReactiveRedisTemplate template = new ReactiveRedisTemplate<>(this.redisConnectionFactory, + RedisSerializationContext.string()); + ObjectRecord record = template.opsForStream().read(List.class, StreamOffset + .fromStart(STREAM_KEY)) + .blockFirst(); + + assertThat(record.getStream()).isEqualTo(STREAM_KEY); + assertThat(record.getValue()).isEqualTo(messagePayload); + } + + + @Test + @RedisAvailable + public void testExplicitSerializationContextWithModel() { Address address = new Address("Rennes, France"); Person person = new Person(address, "Attoumane"); Message message = new GenericMessage<>(person); - RedisSerializationContext serializationContext = redisSerializationContext(); - - streamMessageHandler.setSerializationContext(serializationContext); - streamMessageHandler.afterPropertiesSet(); - - handlerAdapter.handleMessage(message); + this.handlerAdapter.handleMessage(message); ReactiveRedisTemplate template = - new ReactiveRedisTemplate<>(redisConnectionFactory, serializationContext); + new ReactiveRedisTemplate<>(this.redisConnectionFactory, RedisSerializationContext.string()); ObjectRecord record = template.opsForStream() @@ -128,12 +137,6 @@ public class ReactiveRedisStreamMessageHandlerTests extends RedisAvailableTests assertThat(record.getValue().getAddress().getAddress()).isEqualTo("Rennes, France"); } - - private RedisSerializationContext redisSerializationContext() { - return RedisSerializationContext.fromSerializer(StringRedisSerializer.UTF_8); - } - - @Configuration public static class ReactiveRedisStreamMessageHandlerTestsContext { diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/Address.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/Address.java index 6551477139..a4f1eaa0ff 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/Address.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/Address.java @@ -17,6 +17,7 @@ package org.springframework.integration.redis.util; import java.io.Serializable; +import java.util.Objects; @SuppressWarnings("serial") public class Address implements Serializable { @@ -37,4 +38,18 @@ public class Address implements Serializable { public Address(String address) { this.address = address; } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + Address address1 = (Address) o; + return Objects.equals(this.address, address1.address); + } + + @Override + public int hashCode() { + return Objects.hash(this.address); + } + } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/Person.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/Person.java index 982aac5c4e..92f352d7e8 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/Person.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/Person.java @@ -17,6 +17,7 @@ package org.springframework.integration.redis.util; import java.io.Serializable; +import java.util.Objects; @SuppressWarnings("serial") @@ -49,4 +50,20 @@ public class Person implements Serializable { public void setName(String name) { this.name = name; } + + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + Person person = (Person) o; + return Objects.equals(this.address, person.address) && + Objects.equals(this.name, person.name); + } + + @Override + public int hashCode() { + return Objects.hash(this.address, this.name); + } + }