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

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<String, ?, ?> reactiveStreamOperations;
+
+	private StreamReceiver.StreamReceiverOptions<String, ?> streamReceiverOptions =
+			StreamReceiver.StreamReceiverOptions.builder()
+					.pollTimeout(Duration.ZERO)
+					.build();
+
+	private StreamReceiver<String, ?> 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<String, ?> 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<String, ?> reactiveRedisTemplate =
+				new ReactiveRedisTemplate<>(this.reactiveConnectionFactory, RedisSerializationContext.string());
+		this.reactiveStreamOperations = reactiveRedisTemplate.opsForStream();
+	}
+
+	@Override
+	protected void doStart() {
+		super.doStart();
+
+		StreamOffset<String> offset = StreamOffset.create(this.streamKey, this.readOffset);
+
+		Flux<? extends Record<String, ?>> 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<? extends Message<?>> 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);
+	}
+
+}

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";
+
 }

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<String, ?> 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<String, ?> 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;
+		}
+
+	}
+
+}

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<String, String> 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<String, ?> serializationContext = redisSerializationContext();
+		this.messageChannel.send(new GenericMessage<>(messagePayload));
 
 		ReactiveRedisTemplate<String, ?> template =
-				new ReactiveRedisTemplate<>(redisConnectionFactory, serializationContext);
+				new ReactiveRedisTemplate<>(this.redisConnectionFactory, RedisSerializationContext.string());
 
 		ObjectRecord<String, String> record =
 				template.opsForStream()
@@ -102,21 +99,33 @@ public class ReactiveRedisStreamMessageHandlerTests extends RedisAvailableTests
 
 	@Test
 	@RedisAvailable
-	public void explicitSerializationContextWithModelTest() {
+	public void testMessageWithListPayload() {
+		List<String> messagePayload = Arrays.asList("Hello", "stream", "message");
+		this.handlerAdapter.handleMessage(new GenericMessage<>(messagePayload));
+
+		ReactiveRedisTemplate<String, ?> template = new ReactiveRedisTemplate<>(this.redisConnectionFactory,
+				RedisSerializationContext.string());
+		ObjectRecord<String, ?> 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<String, ?> serializationContext = redisSerializationContext();
-
-		streamMessageHandler.setSerializationContext(serializationContext);
-		streamMessageHandler.afterPropertiesSet();
-
-		handlerAdapter.handleMessage(message);
+		this.handlerAdapter.handleMessage(message);
 
 		ReactiveRedisTemplate<String, ?> template =
-				new ReactiveRedisTemplate<>(redisConnectionFactory, serializationContext);
+				new ReactiveRedisTemplate<>(this.redisConnectionFactory, RedisSerializationContext.string());
 
 		ObjectRecord<String, Person> record =
 				template.opsForStream()
@@ -128,12 +137,6 @@ public class ReactiveRedisStreamMessageHandlerTests extends RedisAvailableTests
 		assertThat(record.getValue().getAddress().getAddress()).isEqualTo("Rennes, France");
 	}
 
-
-	private RedisSerializationContext<String, ?> redisSerializationContext() {
-		return RedisSerializationContext.fromSerializer(StringRedisSerializer.UTF_8);
-	}
-
-
 	@Configuration
 	public static class ReactiveRedisStreamMessageHandlerTestsContext {
 

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);
+	}
+
 }

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);
+	}
+
 }