Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3226: Redis Stream Outbound Channel Adapter

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3226

* Redis stream message handler support.
* This is the outbound part publishing message to the actual stream using ReactiveStreamOperations

* Addition of more test cases with one using `MessageChannel`.

* Improvements after PR review.

* Removed failed test reading List from a Stream
* Code style clean up
* Remove `rawtypes` usage
* Remove redundant inner classes for test model
* Add `What's New` note
This commit is contained in:
Attoumane AHAMADI
2020-04-23 19:27:38 +02:00
committed by Artem Bilan
parent 4761800528
commit afa79d868b
7 changed files with 449 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

153
spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandler.java Normal file
View File

@@ -0,0 +1,153 @@
/*
* 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.outbound;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.ReactiveStreamOperations;
import org.springframework.data.redis.hash.HashMapper;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.handler.AbstractReactiveMessageHandler;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import reactor.core.publisher.Mono;
/**
* Implementation of {@link org.springframework.messaging.ReactiveMessageHandler} which writes
* Message payload or Message itself (see {@link #extractPayload}) into a Redis stream using Reactive Stream operations.
*
* @author Attoumane Ahamadi
* @author Artem Bilan
*
* @since 5.4
*/
public class ReactiveRedisStreamMessageHandler extends AbstractReactiveMessageHandler {
private final Expression streamKeyExpression;
private final ReactiveRedisConnectionFactory connectionFactory;
private EvaluationContext evaluationContext;
private boolean extractPayload = true;
private ReactiveStreamOperations<String, ?, ?> reactiveStreamOperations;
private RedisSerializationContext<String, ?> serializationContext = RedisSerializationContext.string();
@Nullable
private HashMapper<String, ?, ?> hashMapper;
/**
* Create an instance based on provided {@link ReactiveRedisConnectionFactory} and key for stream.
* @param connectionFactory the {@link ReactiveRedisConnectionFactory} to use
* @param streamKey the key for stream
*/
public ReactiveRedisStreamMessageHandler(ReactiveRedisConnectionFactory connectionFactory, String streamKey) {
this(connectionFactory, new LiteralExpression(streamKey));
}
/**
* Create an instance based on provided {@link ReactiveRedisConnectionFactory} and expression for stream key.
* @param connectionFactory the {@link ReactiveRedisConnectionFactory} to use
* @param streamKeyExpression the SpEL expression to evaluate a key for stream
*/
public ReactiveRedisStreamMessageHandler(ReactiveRedisConnectionFactory connectionFactory,
Expression streamKeyExpression) {
Assert.notNull(streamKeyExpression, "'streamKeyExpression' must not be null");
Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
this.streamKeyExpression = streamKeyExpression;
this.connectionFactory = connectionFactory;
}
public void setSerializationContext(RedisSerializationContext<String, ?> serializationContext) {
Assert.notNull(serializationContext, "'serializationContext' must not be null");
this.serializationContext = serializationContext;
}
/**
* (Optional) Set the {@link HashMapper} used to create {@link #reactiveStreamOperations}.
* The default {@link HashMapper} is defined from the provided {@link RedisSerializationContext}
* @param hashMapper the wanted hashMapper
* */
public void setHashMapper(@Nullable HashMapper<String, ?, ?> hashMapper) {
this.hashMapper = hashMapper;
}
/**
* Set to {@code true} to extract the payload; otherwise
* the entire message is sent. Default {@code true}.
* @param extractPayload false to not extract.
*/
public void setExtractPayload(boolean extractPayload) {
this.extractPayload = extractPayload;
}
@Override
public String getComponentType() {
return "redis:stream-outbound-channel-adapter";
}
@Override
@SuppressWarnings("unchecked")
protected void onInit() {
super.onInit();
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
ReactiveRedisTemplate<String, ?> template =
new ReactiveRedisTemplate<>(this.connectionFactory, this.serializationContext);
this.reactiveStreamOperations =
this.hashMapper == null
? template.opsForStream()
: template.opsForStream(
(HashMapper<? super String, ? super Object, ? super Object>) this.hashMapper);
}
@Override
protected Mono<Void> handleMessageInternal(Message<?> message) {
return Mono
.fromSupplier(() -> {
String streamKey = this.streamKeyExpression.getValue(this.evaluationContext, message, String.class);
Assert.notNull(streamKey, "'streamKey' must not be null");
return streamKey;
})
.flatMap((streamKey) -> {
Object value = message;
if (this.extractPayload) {
value = message.getPayload();
}
Record<String, ?> record =
StreamRecords.objectBacked(value)
.withStreamKey(streamKey);
return this.reactiveStreamOperations.add(record);
})
.then();
}
}

170
spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandlerTests.java Normal file
View File

@@ -0,0 +1,170 @@
/*
* 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.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
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;
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.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
/**
* @author Attoumane Ahamadi
* @author Artem Bilan
*
* @since 5.4
*/
@RunWith(SpringRunner.class)
@DirtiesContext
public class ReactiveRedisStreamMessageHandlerTests extends RedisAvailableTests {
private static final String STREAM_KEY = "myStream";
@Autowired
@Qualifier("streamChannel")
private MessageChannel messageChannel;
@Autowired
private ReactiveRedisConnectionFactory redisConnectionFactory;
@Autowired
private ReactiveMessageHandlerAdapter handlerAdapter;
@Autowired
private ReactiveRedisStreamMessageHandler streamMessageHandler;
@Before
public void deleteStreamKey() {
ReactiveRedisTemplate<String, String> template = new ReactiveRedisTemplate<>(this.redisConnectionFactory,
RedisSerializationContext.string());
template.delete(STREAM_KEY).block();
}
@Test
@RedisAvailable
public void integrationStreamOutboundTest() {
String messagePayload = "Hello stream message";
messageChannel.send(new GenericMessage<>(messagePayload));
RedisSerializationContext<String, ?> serializationContext = redisSerializationContext();
ReactiveRedisTemplate<String, ?> template =
new ReactiveRedisTemplate<>(redisConnectionFactory, serializationContext);
ObjectRecord<String, String> record =
template.opsForStream()
.read(String.class, StreamOffset.fromStart(STREAM_KEY))
.blockFirst();
assertThat(record.getStream()).isEqualTo(STREAM_KEY);
assertThat(record.getValue()).isEqualTo(messagePayload);
}
@Test
@RedisAvailable
public void explicitSerializationContextWithModelTest() {
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);
ReactiveRedisTemplate<String, ?> template =
new ReactiveRedisTemplate<>(redisConnectionFactory, serializationContext);
ObjectRecord<String, Person> record =
template.opsForStream()
.read(Person.class, StreamOffset.fromStart(STREAM_KEY))
.blockFirst();
assertThat(record.getStream()).isEqualTo(STREAM_KEY);
assertThat(record.getValue().getName()).isEqualTo("Attoumane");
assertThat(record.getValue().getAddress().getAddress()).isEqualTo("Rennes, France");
}
private RedisSerializationContext<String, ?> redisSerializationContext() {
return RedisSerializationContext.fromSerializer(StringRedisSerializer.UTF_8);
}
@Configuration
public static class ReactiveRedisStreamMessageHandlerTestsContext {
@Bean
public MessageChannel streamChannel(ReactiveMessageHandlerAdapter messageHandlerAdapter) {
DirectChannel directChannel = new DirectChannel();
directChannel.subscribe(messageHandlerAdapter);
directChannel.setMaxSubscribers(1);
return directChannel;
}
@Bean
public ReactiveRedisStreamMessageHandler streamMessageHandler(
ReactiveRedisConnectionFactory connectionFactory) {
return new ReactiveRedisStreamMessageHandler(connectionFactory, STREAM_KEY);
}
@Bean
public ReactiveMessageHandlerAdapter reactiveMessageHandlerAdapter(
ReactiveRedisStreamMessageHandler streamMessageHandler) {
return new ReactiveMessageHandlerAdapter(streamMessageHandler);
}
@Bean
public ReactiveRedisConnectionFactory reactiveRedisConnectionFactory() {
return RedisAvailableRule.connectionFactory;
}
}
}

89
spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2007-2019 the original author or authors.
* Copyright 2007-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.
@@ -18,7 +18,6 @@ package org.springframework.integration.redis.store;
import static org.assertj.core.api.Assertions.assertThat;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
@@ -35,6 +34,8 @@ import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.redis.util.Address;
import org.springframework.integration.redis.util.Person;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
@@ -43,6 +44,7 @@ import org.springframework.messaging.support.GenericMessage;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*
*/
public class RedisMessageStoreTests extends RedisAvailableTests {
@@ -57,7 +59,7 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testGetNonExistingMessage() {
RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisConnectionFactory jcf = getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
Message<?> message = store.getMessage(UUID.randomUUID());
assertThat(message).isNull();
@@ -66,7 +68,7 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testGetMessageCountWhenEmpty() {
RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisConnectionFactory jcf = getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
assertThat(store.getMessageCount()).isEqualTo(0);
}
@@ -76,8 +78,8 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
public void testAddStringMessage() {
RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
Message<String> stringMessage = new GenericMessage<String>("Hello Redis");
Message<String> storedMessage = store.addMessage(stringMessage);
Message<String> stringMessage = new GenericMessage<>("Hello Redis");
Message<String> storedMessage = store.addMessage(stringMessage);
assertThat(storedMessage).isNotSameAs(stringMessage);
assertThat(storedMessage.getPayload()).isEqualTo("Hello Redis");
}
@@ -85,14 +87,14 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testAddSerializableObjectMessage() {
RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisConnectionFactory jcf = getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
Address address = new Address();
address.setAddress("1600 Pennsylvania Av, Washington, DC");
Person person = new Person(address, "Barak Obama");
Message<Person> objectMessage = new GenericMessage<Person>(person);
Message<Person> storedMessage = store.addMessage(objectMessage);
Message<Person> objectMessage = new GenericMessage<>(person);
Message<Person> storedMessage = store.addMessage(objectMessage);
assertThat(storedMessage).isNotSameAs(objectMessage);
assertThat(storedMessage.getPayload().getName()).isEqualTo("Barak Obama");
}
@@ -100,10 +102,10 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
@Test(expected = IllegalArgumentException.class)
@RedisAvailable
public void testAddNonSerializableObjectMessage() {
RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisConnectionFactory jcf = getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
Message<Foo> objectMessage = new GenericMessage<Foo>(new Foo());
Message<Foo> objectMessage = new GenericMessage<>(new Foo());
store.addMessage(objectMessage);
}
@@ -111,9 +113,9 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testAddAndGetStringMessage() {
RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisConnectionFactory jcf = getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
Message<String> stringMessage = new GenericMessage<String>("Hello Redis");
Message<String> stringMessage = new GenericMessage<>("Hello Redis");
store.addMessage(stringMessage);
Message<String> retrievedMessage = (Message<String>) store.getMessage(stringMessage.getHeaders().getId());
assertThat(retrievedMessage).isNotNull();
@@ -124,9 +126,9 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testAddAndGetWithPrefix() {
RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisConnectionFactory jcf = getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf, "foo");
Message<String> stringMessage = new GenericMessage<String>("Hello Redis");
Message<String> stringMessage = new GenericMessage<>("Hello Redis");
store.addMessage(stringMessage);
Message<String> retrievedMessage = (Message<String>) store.getMessage(stringMessage.getHeaders().getId());
assertThat(retrievedMessage).isNotNull();
@@ -142,9 +144,9 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testAddAndRemoveStringMessage() {
RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisConnectionFactory jcf = getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
Message<String> stringMessage = new GenericMessage<String>("Hello Redis");
Message<String> stringMessage = new GenericMessage<>("Hello Redis");
store.addMessage(stringMessage);
Message<String> retrievedMessage = (Message<String>) store.removeMessage(stringMessage.getHeaders().getId());
assertThat(retrievedMessage).isNotNull();
@@ -154,11 +156,11 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testWithMessageHistory() throws Exception {
public void testWithMessageHistory() {
RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
Message<?> message = new GenericMessage<String>("Hello");
Message<?> message = new GenericMessage<>("Hello");
DirectChannel fooChannel = new DirectChannel();
fooChannel.setBeanName("fooChannel");
DirectChannel barChannel = new DirectChannel();
@@ -178,11 +180,11 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testAddAndRemoveMessagesFromMessageGroup() throws Exception {
public void testAddAndRemoveMessagesFromMessageGroup() {
RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore messageStore = new RedisMessageStore(jcf);
String groupId = "X";
List<Message<?>> messages = new ArrayList<Message<?>>();
List<Message<?>> messages = new ArrayList<>();
for (int i = 0; i < 25; i++) {
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.addMessagesToGroup(groupId, message);
@@ -194,51 +196,6 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
messageStore.removeMessageGroup("X");
}
@SuppressWarnings("serial")
public static class Person implements Serializable {
private Address address;
private String name;
public Person(Address address, String name) {
this.address = address;
this.name = name;
}
public Address getAddress() {
return address;
}
public void setAddress(Address address) {
this.address = address;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
@SuppressWarnings("serial")
public static class Address implements Serializable {
private String address;
public String getAddress() {
return address;
}
public void setAddress(String address) {
this.address = address;
}
}
public static class Foo {
}

40
spring-integration-redis/src/test/java/org/springframework/integration/redis/util/Address.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* Copyright 2013-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.util;
import java.io.Serializable;
@SuppressWarnings("serial")
public class Address implements Serializable {
private String address;
public String getAddress() {
return address;
}
public void setAddress(String address) {
this.address = address;
}
public Address() {
}
public Address(String address) {
this.address = address;
}
}

52
spring-integration-redis/src/test/java/org/springframework/integration/redis/util/Person.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* Copyright 2013-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.util;
import java.io.Serializable;
@SuppressWarnings("serial")
public class Person implements Serializable {
private Address address;
private String name;
public Person(Address address, String name) {
this.address = address;
this.name = name;
}
public Person() {
}
public Address getAddress() {
return address;
}
public void setAddress(Address address) {
this.address = address;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}

5
src/reference/asciidoc/redis.adoc
View File

@@ -788,3 +788,8 @@ However, the resources protected by such a lock may have been compromised, so su
You should set the expiry at a large enough value to prevent this condition, but set it low enough that the lock can be recovered after a server failure in a reasonable amount of time.
Starting with version 5.0, the `RedisLockRegistry` implements `ExpirableLockRegistry`, which removes locks last acquired more than `age` ago and that are not currently locked.
[[redis-stream-outbound]]
=== Redis Stream Outbound Channel Adapter
TBD

7
src/reference/asciidoc/whats-new.adoc
View File

@@ -23,7 +23,12 @@ See <<./kafka.adoc#kafka,Spring for Apache Kafka Support>> for more information.
==== R2DBC Channel Adapters
The Channel Adapters for R2DBC database interaction have been introduced.
See <<./r2dbc.adoc#r2dbc,R2DBC Support>> for more information.
See <<./r2dbc.adoc#r2dbc,R2DBC Support>> for more information.
==== Redis Stream Support
The Channel Adapters for Redis Stream support have been introduced.
See <<./redis.adoc#redis-stream-outbound,Redis Stream Outbound Channel Adapter>> for more information.
[[x5.4-general]]
=== General Changes