Moving the Redis Binder out to its own repo
removed redis starter
This commit is contained in:
Mark Fisher
committed by
Marius Bogoevici
Marius Bogoevici
parent
f9f92ae184
commit
2b397b45a7
@@ -1,7 +1,6 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>spring-cloud-stream-binders-parent</artifactId>
|
||||
<packaging>pom</packaging>
|
||||
<url>https://github.com/spring-cloud/spring-cloud-stream</url>
|
||||
@@ -9,6 +8,7 @@
|
||||
<name>Pivotal Software, Inc.</name>
|
||||
<url>http://www.spring.io</url>
|
||||
</organization>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-parent</artifactId>
|
||||
@@ -16,12 +16,11 @@
|
||||
</parent>
|
||||
|
||||
<modules>
|
||||
<module>spring-cloud-stream-binder-test</module>
|
||||
<module>spring-cloud-stream-binder-rabbit</module>
|
||||
<module>spring-cloud-stream-binder-redis</module>
|
||||
<module>spring-cloud-stream-binder-kafka</module>
|
||||
<module>spring-cloud-stream-binder-rabbit</module>
|
||||
<module>spring-cloud-stream-binder-test</module>
|
||||
</modules>
|
||||
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
|
||||
@@ -1,64 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>spring-cloud-stream-binder-redis</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<name>spring-cloud-stream-binder-redis</name>
|
||||
<description>Redis binder implementation</description>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binders-parent</artifactId>
|
||||
<version>1.0.0.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-codec</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-autoconfigure</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-core</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-redis</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-redis</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>joda-time</groupId>
|
||||
<artifactId>joda-time</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -1,408 +0,0 @@
|
||||
/*
|
||||
* Copyright 2014-2016 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
|
||||
*
|
||||
* http://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.cloud.stream.binder.redis;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
|
||||
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
|
||||
import org.springframework.cloud.stream.binder.AbstractBinder;
|
||||
import org.springframework.cloud.stream.binder.BinderHeaders;
|
||||
import org.springframework.cloud.stream.binder.Binding;
|
||||
import org.springframework.cloud.stream.binder.ConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.DefaultBinding;
|
||||
import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
|
||||
import org.springframework.cloud.stream.binder.MessageValues;
|
||||
import org.springframework.cloud.stream.binder.PartitionHandler;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.core.RedisOperations;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.expression.spel.standard.SpelExpressionParser;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.endpoint.EventDrivenConsumer;
|
||||
import org.springframework.integration.endpoint.MessageProducerSupport;
|
||||
import org.springframework.integration.handler.AbstractMessageHandler;
|
||||
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
|
||||
import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
|
||||
import org.springframework.integration.redis.outbound.RedisQueueOutboundChannelAdapter;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.retry.RecoveryCallback;
|
||||
import org.springframework.retry.RetryCallback;
|
||||
import org.springframework.retry.RetryContext;
|
||||
import org.springframework.retry.support.RetryTemplate;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by Redis.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
* @author Gary Russell
|
||||
* @author David Turanski
|
||||
* @author Jennifer Hickey
|
||||
*/
|
||||
public class RedisMessageChannelBinder extends AbstractBinder<MessageChannel, ConsumerProperties, ProducerProperties> {
|
||||
|
||||
private static final String ERROR_HEADER = "errorKey";
|
||||
|
||||
static final String CONSUMER_GROUPS_KEY_PREFIX = "groups.";
|
||||
|
||||
private static final SpelExpressionParser parser = new SpelExpressionParser();
|
||||
|
||||
private final String[] headersToMap;
|
||||
|
||||
private final RedisOperations<String, String> redisOperations;
|
||||
|
||||
private final RedisConnectionFactory connectionFactory;
|
||||
|
||||
private final EmbeddedHeadersMessageConverter embeddedHeadersMessageConverter = new
|
||||
EmbeddedHeadersMessageConverter();
|
||||
|
||||
private final RedisQueueOutboundChannelAdapter errorAdapter;
|
||||
|
||||
public RedisMessageChannelBinder(RedisConnectionFactory connectionFactory) {
|
||||
this(connectionFactory, new String[0]);
|
||||
}
|
||||
|
||||
public RedisMessageChannelBinder(RedisConnectionFactory connectionFactory, String... headersToMap) {
|
||||
Assert.notNull(connectionFactory, "connectionFactory must not be null");
|
||||
this.connectionFactory = connectionFactory;
|
||||
StringRedisTemplate template = new StringRedisTemplate(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
this.redisOperations = template;
|
||||
if (headersToMap != null && headersToMap.length > 0) {
|
||||
String[] combinedHeadersToMap =
|
||||
Arrays.copyOfRange(BinderHeaders.STANDARD_HEADERS, 0, BinderHeaders.STANDARD_HEADERS.length
|
||||
+ headersToMap.length);
|
||||
System.arraycopy(headersToMap, 0, combinedHeadersToMap, BinderHeaders.STANDARD_HEADERS.length,
|
||||
headersToMap.length);
|
||||
this.headersToMap = combinedHeadersToMap;
|
||||
}
|
||||
else {
|
||||
this.headersToMap = BinderHeaders.STANDARD_HEADERS;
|
||||
}
|
||||
this.errorAdapter = new RedisQueueOutboundChannelAdapter(
|
||||
parser.parseExpression("headers['" + ERROR_HEADER + "']"), connectionFactory);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onInit() {
|
||||
this.errorAdapter.setIntegrationEvaluationContext(this.evaluationContext);
|
||||
this.errorAdapter.setBeanFactory(getBeanFactory());
|
||||
this.errorAdapter.afterPropertiesSet();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Binding<MessageChannel> doBindConsumer(final String name, String group, MessageChannel moduleInputChannel, ConsumerProperties properties) {
|
||||
if (!StringUtils.hasText(group)) {
|
||||
group = "anonymous." + UUID.randomUUID().toString();
|
||||
}
|
||||
String queueName = groupedName(name, group);
|
||||
if (properties.isPartitioned()) {
|
||||
queueName += "-" + properties.getInstanceIndex();
|
||||
}
|
||||
MessageProducerSupport adapter = createInboundAdapter(properties, queueName);
|
||||
return doRegisterConsumer(name, group, queueName, moduleInputChannel, adapter, properties);
|
||||
}
|
||||
|
||||
private MessageProducerSupport createInboundAdapter(ConsumerProperties accessor, String queueName) {
|
||||
MessageProducerSupport adapter;
|
||||
int concurrency = accessor.getConcurrency();
|
||||
concurrency = concurrency > 0 ? concurrency : 1;
|
||||
if (concurrency == 1) {
|
||||
RedisQueueMessageDrivenEndpoint single = new RedisQueueMessageDrivenEndpoint(queueName,
|
||||
this.connectionFactory);
|
||||
single.setBeanFactory(getBeanFactory());
|
||||
single.setSerializer(null);
|
||||
adapter = single;
|
||||
}
|
||||
else {
|
||||
adapter = new CompositeRedisQueueMessageDrivenEndpoint(queueName, concurrency);
|
||||
}
|
||||
return adapter;
|
||||
}
|
||||
|
||||
private Binding<MessageChannel> doRegisterConsumer(String bindingName, String group, String channelName, MessageChannel moduleInputChannel,
|
||||
MessageProducerSupport adapter, final ConsumerProperties properties) {
|
||||
DirectChannel bridgeToModuleChannel = new DirectChannel();
|
||||
bridgeToModuleChannel.setBeanFactory(this.getBeanFactory());
|
||||
bridgeToModuleChannel.setBeanName(channelName + ".bridge");
|
||||
MessageChannel bridgeInputChannel = addRetryIfNeeded(channelName, bridgeToModuleChannel, properties);
|
||||
adapter.setOutputChannel(bridgeInputChannel);
|
||||
adapter.setBeanName("inbound." + channelName);
|
||||
adapter.afterPropertiesSet();
|
||||
DefaultBinding<MessageChannel> consumerBinding = new DefaultBinding<MessageChannel>(bindingName, group, moduleInputChannel, adapter) {
|
||||
|
||||
@Override
|
||||
protected void afterUnbind() {
|
||||
String key = RedisMessageChannelBinder.CONSUMER_GROUPS_KEY_PREFIX + getName();
|
||||
RedisMessageChannelBinder.this.redisOperations.boundZSetOps(key).incrementScore(getGroup(), -1);
|
||||
}
|
||||
};
|
||||
ReceivingHandler convertingBridge = new ReceivingHandler();
|
||||
convertingBridge.setOutputChannel(moduleInputChannel);
|
||||
convertingBridge.setBeanName(channelName + ".bridge.handler");
|
||||
convertingBridge.afterPropertiesSet();
|
||||
bridgeToModuleChannel.subscribe(convertingBridge);
|
||||
this.redisOperations.boundZSetOps(CONSUMER_GROUPS_KEY_PREFIX + bindingName).incrementScore(group, 1);
|
||||
adapter.start();
|
||||
return consumerBinding;
|
||||
}
|
||||
|
||||
/**
|
||||
* If retry is enabled, wrap the bridge channel in another that will invoke send() within the scope of a retry
|
||||
* template.
|
||||
* @param name The name.
|
||||
* @param bridgeToModuleChannel The channel.
|
||||
* @param properties The properties.
|
||||
* @return The channel, or a wrapper.
|
||||
*/
|
||||
private MessageChannel addRetryIfNeeded(final String name, final DirectChannel bridgeToModuleChannel,
|
||||
ConsumerProperties properties) {
|
||||
final RetryTemplate retryTemplate = buildRetryTemplateIfRetryEnabled(properties);
|
||||
if (retryTemplate == null) {
|
||||
return bridgeToModuleChannel;
|
||||
}
|
||||
else {
|
||||
DirectChannel channel = new DirectChannel() {
|
||||
|
||||
@Override
|
||||
protected boolean doSend(final Message<?> message, final long timeout) {
|
||||
try {
|
||||
return retryTemplate.execute(new RetryCallback<Boolean, Exception>() {
|
||||
|
||||
@Override
|
||||
public Boolean doWithRetry(RetryContext context) throws Exception {
|
||||
return bridgeToModuleChannel.send(message, timeout);
|
||||
}
|
||||
|
||||
}, new RecoveryCallback<Boolean>() {
|
||||
|
||||
/**
|
||||
* Send the failed message to 'ERRORS:[name]'.
|
||||
*/
|
||||
@Override
|
||||
public Boolean recover(RetryContext context) throws Exception {
|
||||
logger.error(
|
||||
"Failed to deliver message; retries exhausted; message sent to queue 'ERRORS:"
|
||||
+ name + "' ", context.getLastThrowable());
|
||||
errorAdapter.handleMessage(getMessageBuilderFactory().fromMessage(message)
|
||||
.setHeader(ERROR_HEADER, "ERRORS:" + name)
|
||||
.build());
|
||||
return true;
|
||||
}
|
||||
|
||||
});
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to deliver message", e);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
};
|
||||
channel.setBeanName(name + ".bridge");
|
||||
return channel;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Binding<MessageChannel> doBindProducer(final String name, MessageChannel moduleOutputChannel, ProducerProperties properties) {
|
||||
Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel);
|
||||
return doRegisterProducer(name, moduleOutputChannel, properties);
|
||||
}
|
||||
|
||||
private RedisQueueOutboundChannelAdapter createProducerEndpoint(String name, ProducerProperties properties) {
|
||||
RedisQueueOutboundChannelAdapter queue;
|
||||
if (!properties.isPartitioned()) {
|
||||
queue = new RedisQueueOutboundChannelAdapter(name, this.connectionFactory);
|
||||
}
|
||||
else {
|
||||
queue = new RedisQueueOutboundChannelAdapter(
|
||||
parser.parseExpression(buildPartitionRoutingExpression(name)), this.connectionFactory);
|
||||
}
|
||||
queue.setIntegrationEvaluationContext(this.evaluationContext);
|
||||
queue.setBeanFactory(this.getBeanFactory());
|
||||
queue.afterPropertiesSet();
|
||||
return queue;
|
||||
}
|
||||
|
||||
private Binding<MessageChannel> doRegisterProducer(final String name, MessageChannel moduleOutputChannel,
|
||||
ProducerProperties properties) {
|
||||
Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel);
|
||||
MessageHandler handler = new SendingHandler(name, properties);
|
||||
EventDrivenConsumer consumer = new EventDrivenConsumer((SubscribableChannel) moduleOutputChannel, handler);
|
||||
consumer.setBeanFactory(this.getBeanFactory());
|
||||
consumer.setBeanName("outbound." + name);
|
||||
consumer.afterPropertiesSet();
|
||||
DefaultBinding<MessageChannel> producerBinding =
|
||||
new DefaultBinding<>(name, null, moduleOutputChannel, consumer);
|
||||
String[] requiredGroups = properties.getRequiredGroups();
|
||||
if (!ObjectUtils.isEmpty(requiredGroups)) {
|
||||
for (String group : requiredGroups) {
|
||||
this.redisOperations.boundZSetOps(CONSUMER_GROUPS_KEY_PREFIX + name).incrementScore(group, 1);
|
||||
}
|
||||
}
|
||||
consumer.start();
|
||||
return producerBinding;
|
||||
}
|
||||
|
||||
private class SendingHandler extends AbstractMessageHandler {
|
||||
|
||||
private final String bindingName;
|
||||
|
||||
private final ProducerProperties producerProperties;
|
||||
|
||||
private final Map<String, RedisQueueOutboundChannelAdapter> adapters = new HashMap<>();
|
||||
|
||||
private final PartitionHandler partitionHandler;
|
||||
|
||||
private SendingHandler(String bindingName, ProducerProperties producerProperties) {
|
||||
this.bindingName = bindingName;
|
||||
this.producerProperties = producerProperties;
|
||||
ConfigurableListableBeanFactory beanFactory = RedisMessageChannelBinder.this.getBeanFactory();
|
||||
this.setBeanFactory(beanFactory);
|
||||
this.partitionHandler = new PartitionHandler(beanFactory, evaluationContext, partitionSelector, producerProperties);
|
||||
refreshChannelAdapters();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void handleMessageInternal(Message<?> message) throws Exception {
|
||||
MessageValues transformed = serializePayloadIfNecessary(message);
|
||||
|
||||
if (producerProperties.isPartitioned()) {
|
||||
transformed.put(PARTITION_HEADER, this.partitionHandler.determinePartition(message));
|
||||
}
|
||||
|
||||
byte[] messageToSend = embeddedHeadersMessageConverter.embedHeaders(transformed,
|
||||
RedisMessageChannelBinder.this.headersToMap);
|
||||
|
||||
refreshChannelAdapters();
|
||||
for (RedisQueueOutboundChannelAdapter adapter : adapters.values()) {
|
||||
adapter.handleMessage((MessageBuilder.withPayload(messageToSend).copyHeaders(transformed).build()));
|
||||
}
|
||||
}
|
||||
|
||||
private void refreshChannelAdapters() {
|
||||
Set<String> groups = redisOperations.boundZSetOps(CONSUMER_GROUPS_KEY_PREFIX + bindingName).rangeByScore(1, Double.MAX_VALUE);
|
||||
for (String group : groups) {
|
||||
if (!adapters.containsKey(group)) {
|
||||
String channel = String.format("%s.%s", this.bindingName, group);
|
||||
adapters.put(group, createProducerEndpoint(channel, producerProperties));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class ReceivingHandler extends AbstractReplyProducingMessageHandler {
|
||||
|
||||
public ReceivingHandler() {
|
||||
super();
|
||||
this.setBeanFactory(RedisMessageChannelBinder.this.getBeanFactory());
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Override
|
||||
protected Object handleRequestMessage(Message<?> requestMessage) {
|
||||
MessageValues theRequestMessage;
|
||||
try {
|
||||
theRequestMessage = embeddedHeadersMessageConverter.extractHeaders((Message<byte[]>) requestMessage, true);
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error(EmbeddedHeadersMessageConverter.decodeExceptionMessage(requestMessage), e);
|
||||
theRequestMessage = new MessageValues(requestMessage);
|
||||
}
|
||||
return deserializePayloadIfNecessary(theRequestMessage).toMessage(getMessageBuilderFactory());
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean shouldCopyRequestHeaders() {
|
||||
// prevent returned message from being copied in superclass
|
||||
return false;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Provides concurrency by creating a list of message-driven endpoints.
|
||||
*/
|
||||
private class CompositeRedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
|
||||
|
||||
private final List<RedisQueueMessageDrivenEndpoint> consumers = new
|
||||
ArrayList<RedisQueueMessageDrivenEndpoint>();
|
||||
|
||||
public CompositeRedisQueueMessageDrivenEndpoint(String queueName, int concurrency) {
|
||||
for (int i = 0; i < concurrency; i++) {
|
||||
RedisQueueMessageDrivenEndpoint adapter = new RedisQueueMessageDrivenEndpoint(queueName,
|
||||
connectionFactory);
|
||||
adapter.setBeanFactory(RedisMessageChannelBinder.this.getBeanFactory());
|
||||
adapter.setSerializer(null);
|
||||
adapter.setBeanName("inbound." + queueName + "." + i);
|
||||
this.consumers.add(adapter);
|
||||
}
|
||||
this.setBeanFactory(RedisMessageChannelBinder.this.getBeanFactory());
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void onInit() {
|
||||
for (RedisQueueMessageDrivenEndpoint consumer : consumers) {
|
||||
consumer.afterPropertiesSet();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doStart() {
|
||||
for (RedisQueueMessageDrivenEndpoint consumer : consumers) {
|
||||
consumer.start();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doStop() {
|
||||
for (RedisQueueMessageDrivenEndpoint consumer : consumers) {
|
||||
consumer.stop();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setOutputChannel(MessageChannel outputChannel) {
|
||||
for (RedisQueueMessageDrivenEndpoint consumer : consumers) {
|
||||
consumer.setOutputChannel(outputChannel);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setErrorChannel(MessageChannel errorChannel) {
|
||||
for (RedisQueueMessageDrivenEndpoint consumer : consumers) {
|
||||
consumer.setErrorChannel(errorChannel);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,63 +0,0 @@
|
||||
/*
|
||||
* Copyright 2015-2016 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
|
||||
*
|
||||
* http://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.cloud.stream.binder.redis.config;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.redis.RedisMessageChannelBinder;
|
||||
import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.integration.codec.Codec;
|
||||
|
||||
/**
|
||||
* @author David Turanski
|
||||
*/
|
||||
@Configuration
|
||||
@Import({PropertyPlaceholderAutoConfiguration.class, KryoCodecAutoConfiguration.class})
|
||||
@ConfigurationProperties(prefix = "spring.cloud.stream.binder.redis")
|
||||
public class RedisMessageChannelBinderConfiguration {
|
||||
|
||||
private String[] headers;
|
||||
|
||||
@Autowired
|
||||
private Codec codec;
|
||||
|
||||
|
||||
@Autowired
|
||||
private RedisConnectionFactory redisConnectionFactory;
|
||||
|
||||
@Bean
|
||||
public RedisMessageChannelBinder redisMessageChannelBinder() {
|
||||
|
||||
RedisMessageChannelBinder redisMessageChannelBinder = new RedisMessageChannelBinder(this.redisConnectionFactory,
|
||||
this.headers);
|
||||
redisMessageChannelBinder.setCodec(this.codec);
|
||||
return redisMessageChannelBinder;
|
||||
}
|
||||
|
||||
public String[] getHeaders() {
|
||||
return this.headers;
|
||||
}
|
||||
|
||||
public void setHeaders(String[] headers) {
|
||||
this.headers = headers;
|
||||
}
|
||||
}
|
||||
@@ -1,77 +0,0 @@
|
||||
/*
|
||||
* Copyright 2015-2016 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
|
||||
*
|
||||
* http://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.cloud.stream.binder.redis.config;
|
||||
|
||||
import org.springframework.boot.actuate.health.HealthIndicator;
|
||||
import org.springframework.boot.actuate.health.RedisHealthIndicator;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
|
||||
import org.springframework.boot.autoconfigure.cloud.CloudAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
|
||||
import org.springframework.cloud.Cloud;
|
||||
import org.springframework.cloud.CloudFactory;
|
||||
import org.springframework.cloud.stream.binder.Binder;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.context.annotation.Profile;
|
||||
import org.springframework.context.annotation.PropertySource;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
|
||||
/**
|
||||
* Bind to services, either locally or in a cloud environment.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
* @author Dave Syer
|
||||
* @author David Turanski
|
||||
* @author Eric Bottard
|
||||
*/
|
||||
@Configuration
|
||||
@ConditionalOnMissingBean(Binder.class)
|
||||
@Import(RedisMessageChannelBinderConfiguration.class)
|
||||
@PropertySource("classpath:/META-INF/spring-cloud-stream/redis-binder.properties")
|
||||
@AutoConfigureBefore({CloudAutoConfiguration.class, RedisAutoConfiguration.class})
|
||||
public class RedisServiceAutoConfiguration {
|
||||
|
||||
@Configuration
|
||||
@ConditionalOnClass(Cloud.class)
|
||||
@Profile("cloud")
|
||||
protected static class CloudConfig {
|
||||
|
||||
@Bean
|
||||
public Cloud cloud() {
|
||||
return new CloudFactory().getCloud();
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(RedisConnectionFactory.class)
|
||||
RedisConnectionFactory redisConnectionFactory(Cloud cloud) {
|
||||
return cloud.getSingletonServiceConnector(RedisConnectionFactory.class, null);
|
||||
}
|
||||
}
|
||||
|
||||
@Profile("!cloud")
|
||||
@Import(RedisAutoConfiguration.class)
|
||||
protected static class NoCloudConfig {
|
||||
}
|
||||
|
||||
@Bean
|
||||
public HealthIndicator binderHealthIndicator(RedisConnectionFactory redisConnectionFactory) {
|
||||
return new RedisHealthIndicator(redisConnectionFactory);
|
||||
}
|
||||
}
|
||||
@@ -1,5 +0,0 @@
|
||||
/**
|
||||
* This package contains an implementation of the {@link org.springframework.cloud.stream.binder.Binder} for Redis.
|
||||
*/
|
||||
|
||||
package org.springframework.cloud.stream.binder.redis;
|
||||
@@ -1,5 +0,0 @@
|
||||
spring.cloud.stream.binder.redis.default.backOffInitialInterval: 1000
|
||||
spring.cloud.stream.binder.redis.default.backOffMaxInterval: 10000
|
||||
spring.cloud.stream.binder.redis.default.backOffMultiplier: 2.0
|
||||
spring.cloud.stream.binder.redis.default.concurrency: 1
|
||||
spring.cloud.stream.binder.redis.default.maxAttempts: 3
|
||||
@@ -1,2 +0,0 @@
|
||||
redis:\
|
||||
org.springframework.cloud.stream.binder.redis.config.RedisServiceAutoConfiguration
|
||||
@@ -1,156 +0,0 @@
|
||||
/*
|
||||
* Copyright 2002-2016 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
|
||||
*
|
||||
* http://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.cloud.stream.binder.redis;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import java.util.Date;
|
||||
import java.util.HashMap;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
|
||||
import org.joda.time.DateTime;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.data.redis.serializer.RedisSerializer;
|
||||
import org.springframework.tuple.Tuple;
|
||||
import org.springframework.tuple.TupleBuilder;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonCreator;
|
||||
import com.fasterxml.jackson.annotation.JsonProperty;
|
||||
|
||||
/**
|
||||
* @author David Turanski
|
||||
*
|
||||
*/
|
||||
public abstract class AbstractRedisSerializerTests {
|
||||
|
||||
private RedisSerializer<Object> serializer;
|
||||
|
||||
protected abstract RedisSerializer<Object> getSerializer();
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
serializer = getSerializer();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRandomObjectSerialization() {
|
||||
Foo foo = new Foo("hello");
|
||||
byte[] bytes = serializer.serialize(foo);
|
||||
Object obj = serializer.deserialize(bytes);
|
||||
assertTrue(obj instanceof Foo);
|
||||
assertEquals("hello", ((Foo) obj).bar);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDateSerialization() {
|
||||
Date d = new Date();
|
||||
byte[] bytes = serializer.serialize(d);
|
||||
Date obj = (Date) serializer.deserialize(bytes);
|
||||
assertEquals(d, obj);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDateTimeSerialization() {
|
||||
DateTime d = new DateTime();
|
||||
byte[] bytes = serializer.serialize(d);
|
||||
DateTime obj = (DateTime) serializer.deserialize(bytes);
|
||||
assertEquals(d, obj);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testStringSerialization() {
|
||||
String s = new String("hello");
|
||||
byte[] bytes = serializer.serialize(s);
|
||||
Object obj = serializer.deserialize(bytes);
|
||||
assertEquals(s, obj);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testLongSerialization() {
|
||||
byte[] bytes = serializer.serialize(100L);
|
||||
long obj = (Long) serializer.deserialize(bytes);
|
||||
assertEquals(100, obj);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFloatSerialization() {
|
||||
byte[] bytes = serializer.serialize(99.9f);
|
||||
double obj = (Double) serializer.deserialize(bytes);
|
||||
assertEquals(99.9, obj, 0.1);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBooleanSerialization() {
|
||||
byte[] bytes = serializer.serialize(true);
|
||||
boolean obj = (Boolean) serializer.deserialize(bytes);
|
||||
assertTrue(obj);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMapSerialization() {
|
||||
Map<String, String> map = new HashMap<String, String>();
|
||||
map.put("foo", "bar");
|
||||
byte[] bytes = serializer.serialize(map);
|
||||
Map<?, ?> obj = (Map<?, ?>) serializer.deserialize(bytes);
|
||||
assertEquals("bar", obj.get("foo"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testListSerialization() {
|
||||
List<String> list = new LinkedList<String>();
|
||||
list.add("foo");
|
||||
byte[] bytes = serializer.serialize(list);
|
||||
List<?> obj = (List<?>) serializer.deserialize(bytes);
|
||||
assertEquals("foo", obj.get(0));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSetSerialization() {
|
||||
Set<String> set = new TreeSet<String>();
|
||||
set.add("foo");
|
||||
byte[] bytes = serializer.serialize(set);
|
||||
Set<?> obj = (Set<?>) serializer.deserialize(bytes);
|
||||
assertEquals("foo", obj.iterator().next());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testTupleSerialization() {
|
||||
Tuple t = TupleBuilder.tuple().of("foo", "bar");
|
||||
byte[] bytes = serializer.serialize(t);
|
||||
|
||||
Tuple obj = (Tuple) serializer.deserialize(bytes);
|
||||
assertEquals("bar", obj.getString("foo"));
|
||||
}
|
||||
|
||||
public static class Foo {
|
||||
|
||||
@JsonCreator
|
||||
public Foo(@JsonProperty("bar") String val) {
|
||||
bar = val;
|
||||
}
|
||||
|
||||
public String bar;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,259 +0,0 @@
|
||||
/*
|
||||
* Copyright 2013-2016 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
|
||||
*
|
||||
* http://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.cloud.stream.binder.redis;
|
||||
|
||||
import static org.hamcrest.Matchers.containsString;
|
||||
import static org.hamcrest.Matchers.instanceOf;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertSame;
|
||||
import static org.junit.Assert.assertThat;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.cloud.stream.binder.Binding;
|
||||
import org.springframework.cloud.stream.binder.ConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
|
||||
import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.cloud.stream.binder.Spy;
|
||||
import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||||
import org.springframework.expression.Expression;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.endpoint.AbstractEndpoint;
|
||||
import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
|
||||
import org.springframework.integration.redis.outbound.RedisQueueOutboundChannelAdapter;
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.retry.support.RetryTemplate;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @author David Turanski
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public class RedisBinderTests extends PartitionCapableBinderTests<RedisTestBinder, ConsumerProperties, ProducerProperties> {
|
||||
|
||||
private final String CLASS_UNDER_TEST_NAME = RedisMessageChannelBinder.class.getSimpleName();
|
||||
|
||||
@Rule
|
||||
public RedisTestSupport redisAvailableRule = new RedisTestSupport();
|
||||
|
||||
private RedisTemplate<String, Object> redisTemplate;
|
||||
|
||||
private static final EmbeddedHeadersMessageConverter embeddedHeadersMessageConverter =
|
||||
new EmbeddedHeadersMessageConverter();
|
||||
|
||||
@Override
|
||||
protected RedisTestBinder getBinder() {
|
||||
if (testBinder == null) {
|
||||
testBinder = new RedisTestBinder(redisAvailableRule.getResource());
|
||||
}
|
||||
return testBinder;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected ConsumerProperties createConsumerProperties() {
|
||||
return new ConsumerProperties();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected ProducerProperties createProducerProperties() {
|
||||
return new ProducerProperties();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean usesExplicitRouting() {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testConsumerProperties() throws Exception {
|
||||
RedisTestBinder binder = getBinder();
|
||||
ConsumerProperties properties1 = new ConsumerProperties();
|
||||
properties1.setMaxAttempts(1);
|
||||
Binding<MessageChannel> binding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties1);
|
||||
AbstractEndpoint endpoint = extractEndpoint(binding);
|
||||
assertThat(endpoint, instanceOf(RedisQueueMessageDrivenEndpoint.class));
|
||||
assertSame(DirectChannel.class, TestUtils.getPropertyValue(endpoint, "outputChannel").getClass());
|
||||
binding.unbind();
|
||||
assertFalse(endpoint.isRunning());
|
||||
|
||||
ConsumerProperties properties2 = new ConsumerProperties();
|
||||
properties2.setBackOffInitialInterval(2000);
|
||||
properties2.setBackOffMaxInterval(20000);
|
||||
properties2.setBackOffMultiplier(5.0);
|
||||
properties2.setConcurrency(2);
|
||||
properties2.setMaxAttempts(23);
|
||||
properties2.setInstanceIndex(0);
|
||||
|
||||
binding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties2);
|
||||
endpoint = extractEndpoint(binding);
|
||||
verifyConsumer(endpoint);
|
||||
|
||||
binding.unbind();
|
||||
assertFalse(endpoint.isRunning());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testProducerProperties() throws Exception {
|
||||
RedisTestBinder binder = getBinder();
|
||||
Binding<MessageChannel> consumerBinding = binder.bindConsumer("props.0", "test", new DirectChannel(), createConsumerProperties());
|
||||
Binding<MessageChannel> producerBinding = binder.bindProducer("props.0", new DirectChannel(), createProducerProperties());
|
||||
AbstractEndpoint producerEndpoint = extractEndpoint(producerBinding);
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, RedisQueueOutboundChannelAdapter> adapters = TestUtils.getPropertyValue(producerEndpoint, "handler.adapters", Map.class);
|
||||
RedisQueueOutboundChannelAdapter adapter = adapters.get("test");
|
||||
assertEquals(
|
||||
"props.0.test",
|
||||
TestUtils.getPropertyValue(adapter, "queueNameExpression", Expression.class).getExpressionString());
|
||||
producerBinding.unbind();
|
||||
assertFalse(producerEndpoint.isRunning());
|
||||
|
||||
ProducerProperties producerProperties = new ProducerProperties();
|
||||
producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("'foo'"));
|
||||
producerProperties.setPartitionKeyExtractorClass(AbstractRedisSerializerTests.Foo.class);
|
||||
producerProperties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("0"));
|
||||
producerProperties.setPartitionSelectorClass(AbstractRedisSerializerTests.Foo.class);
|
||||
producerProperties.setPartitionCount(1);
|
||||
|
||||
producerBinding = binder.bindProducer("props.0", new DirectChannel(), producerProperties);
|
||||
producerEndpoint = extractEndpoint(producerBinding);
|
||||
adapter = (RedisQueueOutboundChannelAdapter) TestUtils.getPropertyValue(producerEndpoint, "handler.adapters", Map.class).get("test");
|
||||
assertEquals(
|
||||
"'props.0.test-' + headers['partition']",
|
||||
TestUtils.getPropertyValue(adapter, "queueNameExpression", Expression.class).getExpressionString());
|
||||
|
||||
producerBinding.unbind();
|
||||
consumerBinding.unbind();
|
||||
assertFalse(producerEndpoint.isRunning());
|
||||
assertFalse(extractEndpoint(consumerBinding).isRunning());
|
||||
}
|
||||
|
||||
private void verifyConsumer(AbstractEndpoint endpoint) {
|
||||
assertThat(endpoint.getClass().getName(), containsString("CompositeRedisQueueMessageDrivenEndpoint"));
|
||||
assertEquals(2, TestUtils.getPropertyValue(endpoint, "consumers", Collection.class).size());
|
||||
DirectChannel channel = TestUtils.getPropertyValue(
|
||||
TestUtils.getPropertyValue(endpoint, "consumers", List.class).get(0),
|
||||
"outputChannel", DirectChannel.class);
|
||||
assertThat(
|
||||
channel.getClass().getName(), containsString(getClassUnderTestName() + "$")); // retry wrapper
|
||||
assertThat(
|
||||
TestUtils.getPropertyValue(TestUtils.getPropertyValue(endpoint, "consumers", List.class).get(1),
|
||||
"outputChannel").getClass().getName(), containsString(getClassUnderTestName() + "$")); // retry wrapper
|
||||
RetryTemplate retry = TestUtils.getPropertyValue(channel, "val$retryTemplate", RetryTemplate.class);
|
||||
assertEquals(23, TestUtils.getPropertyValue(retry, "retryPolicy.maxAttempts"));
|
||||
assertEquals(2000L, TestUtils.getPropertyValue(retry, "backOffPolicy.initialInterval"));
|
||||
assertEquals(20000L, TestUtils.getPropertyValue(retry, "backOffPolicy.maxInterval"));
|
||||
assertEquals(5.0, TestUtils.getPropertyValue(retry, "backOffPolicy.multiplier"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRetryFail() {
|
||||
RedisTestBinder binder = getBinder();
|
||||
DirectChannel channel = new DirectChannel();
|
||||
binder.bindProducer("retry.0", channel, createProducerProperties());
|
||||
ConsumerProperties consumerProperties = new ConsumerProperties();
|
||||
consumerProperties.setMaxAttempts(2);
|
||||
consumerProperties.setBackOffInitialInterval(100);
|
||||
consumerProperties.setBackOffMultiplier(1.0);
|
||||
Binding<MessageChannel> consumerBinding = binder.bindConsumer("retry.0", "test", new DirectChannel(), consumerProperties); // no subscriber
|
||||
channel.send(new GenericMessage<>("foo"));
|
||||
RedisTemplate<String, Object> template = createTemplate();
|
||||
Object rightPop = template.boundListOps("ERRORS:retry.0.test").rightPop(5, TimeUnit.SECONDS);
|
||||
assertNotNull(rightPop);
|
||||
assertThat(new String((byte[]) rightPop), containsString("foo"));
|
||||
consumerBinding.unbind();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMoreHeaders() {
|
||||
RedisMessageChannelBinder binder = new RedisMessageChannelBinder(mock(RedisConnectionFactory.class), "foo", "bar");
|
||||
Collection<String> headers = Arrays.asList(TestUtils.getPropertyValue(binder, "headersToMap", String[].class));
|
||||
assertEquals(7, headers.size());
|
||||
assertTrue(headers.contains("foo"));
|
||||
assertTrue(headers.contains("bar"));
|
||||
}
|
||||
|
||||
private RedisTemplate<String, Object> createTemplate() {
|
||||
if (this.redisTemplate != null) {
|
||||
return this.redisTemplate;
|
||||
}
|
||||
RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
|
||||
template.setConnectionFactory(this.redisAvailableRule.getResource());
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.setEnableDefaultSerializer(false);
|
||||
template.afterPropertiesSet();
|
||||
this.redisTemplate = template;
|
||||
return template;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
protected String getEndpointRouting(AbstractEndpoint endpoint) {
|
||||
Map<String, RedisQueueOutboundChannelAdapter> adapters = TestUtils.getPropertyValue(endpoint, "handler.adapters", Map.class);
|
||||
return TestUtils.getPropertyValue(adapters.values().iterator().next(), "queueNameExpression", Expression.class).getExpressionString();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String getExpectedRoutingBaseDestination(String name, String group) {
|
||||
return name + "." + group;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String getClassUnderTestName() {
|
||||
return CLASS_UNDER_TEST_NAME;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Spy spyOn(final String queue) {
|
||||
final RedisTemplate<String, Object> template = createTemplate();
|
||||
return new Spy() {
|
||||
|
||||
@Override
|
||||
public Object receive(boolean expectNull) throws Exception {
|
||||
byte[] bytes = (byte[]) template.boundListOps(queue).rightPop(50, TimeUnit.MILLISECONDS);
|
||||
if (bytes == null) {
|
||||
return null;
|
||||
}
|
||||
bytes = (byte[]) embeddedHeadersMessageConverter.extractHeaders(new GenericMessage<byte[]>(bytes), false).getPayload();
|
||||
return new String(bytes, "UTF-8");
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void binderBindUnbindLatency() throws InterruptedException {
|
||||
Thread.sleep(3000); // needed for Redis see INT-3442
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,145 +0,0 @@
|
||||
/*
|
||||
* Copyright 2013 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
|
||||
*
|
||||
* http://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.cloud.stream.binder.redis;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.cloud.stream.binder.BinderTestUtils;
|
||||
import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.listener.ChannelTopic;
|
||||
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
|
||||
import org.springframework.data.redis.listener.Topic;
|
||||
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
|
||||
import org.springframework.data.redis.serializer.GenericToStringSerializer;
|
||||
import org.springframework.data.redis.serializer.RedisSerializer;
|
||||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||||
import org.springframework.integration.redis.outbound.RedisPublishingMessageHandler;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
|
||||
/**
|
||||
* Temporary copy of SI RedisPublishingMessageHandlerTests that adds tests that publish messages with data types other
|
||||
* than String
|
||||
*
|
||||
* @author Mark Fisher
|
||||
* @author Jennifer Hickey
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public class RedisPublishingMessageHandlerTests {
|
||||
|
||||
private static final String TOPIC = "si.test.channel";
|
||||
|
||||
private static final int NUM_MESSAGES = 10;
|
||||
|
||||
private RedisConnectionFactory connectionFactory;
|
||||
|
||||
private RedisMessageListenerContainer container;
|
||||
|
||||
private CountDownLatch latch = new CountDownLatch(NUM_MESSAGES);
|
||||
|
||||
@Rule
|
||||
public RedisTestSupport redisAvailableRule = new RedisTestSupport();
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
this.connectionFactory = redisAvailableRule.getResource();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWithDefaultSerializer() throws Exception {
|
||||
setupListener(new StringRedisSerializer());
|
||||
final RedisPublishingMessageHandler handler = new RedisPublishingMessageHandler(connectionFactory);
|
||||
handler.setBeanFactory(BinderTestUtils.MOCK_BF);
|
||||
handler.setTopic(TOPIC);
|
||||
handler.afterPropertiesSet();
|
||||
for (int i = 0; i < NUM_MESSAGES; i++) {
|
||||
handler.handleMessage(MessageBuilder.withPayload("test-" + i).build());
|
||||
}
|
||||
latch.await(3, TimeUnit.SECONDS);
|
||||
assertEquals(0, latch.getCount());
|
||||
container.stop();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWithNoSerializer() throws Exception {
|
||||
setupListener(null);
|
||||
final RedisPublishingMessageHandler handler = new RedisPublishingMessageHandler(connectionFactory);
|
||||
handler.setBeanFactory(BinderTestUtils.MOCK_BF);
|
||||
handler.setTopic(TOPIC);
|
||||
handler.afterPropertiesSet();
|
||||
for (int i = 0; i < NUM_MESSAGES; i++) {
|
||||
handler.handleMessage(MessageBuilder.withPayload(new String("test-" + i).getBytes()).build());
|
||||
}
|
||||
latch.await(3, TimeUnit.SECONDS);
|
||||
assertEquals(0, latch.getCount());
|
||||
container.stop();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWithCustomSerializer() throws Exception {
|
||||
GenericToStringSerializer<Long> serializer = new GenericToStringSerializer<Long>(Long.class);
|
||||
setupListener(serializer);
|
||||
final RedisPublishingMessageHandler handler = new RedisPublishingMessageHandler(connectionFactory);
|
||||
handler.setBeanFactory(BinderTestUtils.MOCK_BF);
|
||||
handler.setTopic(TOPIC);
|
||||
handler.setSerializer(serializer);
|
||||
handler.afterPropertiesSet();
|
||||
for (long i = 0; i < NUM_MESSAGES; i++) {
|
||||
handler.handleMessage(MessageBuilder.withPayload(i).build());
|
||||
}
|
||||
latch.await(3, TimeUnit.SECONDS);
|
||||
assertEquals(0, latch.getCount());
|
||||
container.stop();
|
||||
}
|
||||
|
||||
private void setupListener(RedisSerializer<?> listenerSerializer) throws InterruptedException {
|
||||
MessageListenerAdapter listener = new MessageListenerAdapter();
|
||||
listener.setDelegate(new Listener(latch));
|
||||
listener.setSerializer(listenerSerializer);
|
||||
listener.afterPropertiesSet();
|
||||
|
||||
this.container = new RedisMessageListenerContainer();
|
||||
container.setConnectionFactory(connectionFactory);
|
||||
container.afterPropertiesSet();
|
||||
container.addMessageListener(listener, Collections.<Topic> singletonList(new ChannelTopic(TOPIC)));
|
||||
container.start();
|
||||
Thread.sleep(1000);
|
||||
}
|
||||
|
||||
private static class Listener {
|
||||
|
||||
private final CountDownLatch latch;
|
||||
|
||||
private Listener(CountDownLatch latch) {
|
||||
this.latch = latch;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void handleMessage(Object s) {
|
||||
this.latch.countDown();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,224 +0,0 @@
|
||||
/*
|
||||
* Copyright 2002-2013 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
|
||||
*
|
||||
* http://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.cloud.stream.binder.redis;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.BlockingDeque;
|
||||
import java.util.concurrent.LinkedBlockingDeque;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.cloud.stream.binder.BinderTestUtils;
|
||||
import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.data.redis.serializer.GenericToStringSerializer;
|
||||
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
|
||||
import org.springframework.data.redis.serializer.RedisSerializer;
|
||||
import org.springframework.data.redis.serializer.SerializationException;
|
||||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
/**
|
||||
* Integration test of {@link RedisQueueInboundChannelAdapter}
|
||||
*
|
||||
* @author Jennifer Hickey
|
||||
*/
|
||||
public class RedisQueueInboundChannelAdapterTests {
|
||||
|
||||
private static final String QUEUE_NAME = "inboundadaptertest";
|
||||
|
||||
private RedisConnectionFactory connectionFactory;
|
||||
|
||||
private final BlockingDeque<Object> messages = new LinkedBlockingDeque<>(99);
|
||||
|
||||
private RedisQueueMessageDrivenEndpoint adapter;
|
||||
|
||||
private double timeoutMultiplier = 1.0D;
|
||||
|
||||
@Rule
|
||||
public RedisTestSupport redisAvailableRule = new RedisTestSupport();
|
||||
|
||||
private String currentQueueName;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
messages.clear();
|
||||
this.connectionFactory = redisAvailableRule.getResource();
|
||||
DirectChannel outputChannel = new DirectChannel();
|
||||
outputChannel.setBeanFactory(BinderTestUtils.MOCK_BF);
|
||||
outputChannel.subscribe(new TestMessageHandler());
|
||||
|
||||
this.currentQueueName = QUEUE_NAME + ":" + System.nanoTime();
|
||||
|
||||
adapter = new RedisQueueMessageDrivenEndpoint(currentQueueName, connectionFactory);
|
||||
adapter.setBeanFactory(BinderTestUtils.MOCK_BF);
|
||||
adapter.setOutputChannel(outputChannel);
|
||||
|
||||
String multiplier = System.getenv("REDIS_TIMEOUT_MULTIPLIER");
|
||||
if (multiplier != null) {
|
||||
timeoutMultiplier = Double.parseDouble(multiplier);
|
||||
}
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() {
|
||||
adapter.stop();
|
||||
connectionFactory.getConnection().del(currentQueueName.getBytes());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDefaultPayloadSerializer() throws Exception {
|
||||
RedisTemplate<String, Object> template = new RedisTemplate<>();
|
||||
template.setConnectionFactory(connectionFactory);
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.afterPropertiesSet();
|
||||
adapter.start();
|
||||
|
||||
template.boundListOps(currentQueueName).rightPush("message1");
|
||||
@SuppressWarnings("unchecked")
|
||||
Message<String> message = (Message<String>) messages.poll((int)(1000 * timeoutMultiplier), TimeUnit.MILLISECONDS);
|
||||
assertNotNull(message);
|
||||
assertEquals("message1", message.getPayload());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDefaultMsgSerializer() throws Exception {
|
||||
RedisTemplate<String, Message<String>> template = new RedisTemplate<String, Message<String>>();
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.setValueSerializer(new JdkSerializationRedisSerializer());
|
||||
template.setConnectionFactory(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.setExpectMessage(true);
|
||||
adapter.afterPropertiesSet();
|
||||
adapter.start();
|
||||
|
||||
Map<String, Object> headers = new HashMap<>();
|
||||
headers.put("header1", "foo");
|
||||
template.boundListOps(currentQueueName).rightPush(new GenericMessage<>("message2", headers));
|
||||
@SuppressWarnings("unchecked")
|
||||
Message<String> message = (Message<String>) messages.poll((int)(1000 * timeoutMultiplier), TimeUnit.MILLISECONDS);
|
||||
assertEquals("message2", message.getPayload());
|
||||
assertEquals("foo", message.getHeaders().get("header1"));
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Test
|
||||
public void testNoSerializer() throws Exception {
|
||||
RedisTemplate<String, byte[]> template = new RedisTemplate<>();
|
||||
template.setEnableDefaultSerializer(false);
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.setConnectionFactory(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.setSerializer(null);
|
||||
adapter.afterPropertiesSet();
|
||||
adapter.start();
|
||||
|
||||
template.boundListOps(currentQueueName).rightPush("message3".getBytes());
|
||||
Message<byte[]> message = (Message<byte[]>) messages.poll((int)(1000 * timeoutMultiplier), TimeUnit.MILLISECONDS);
|
||||
assertEquals("message3", new String(message.getPayload()));
|
||||
}
|
||||
|
||||
@Test(expected = IllegalArgumentException.class)
|
||||
public void testNoSerializerNoExtractPayload() throws Exception {
|
||||
RedisTemplate<String, byte[]> template = new RedisTemplate<>();
|
||||
template.setEnableDefaultSerializer(false);
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.setConnectionFactory(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.setSerializer(null);
|
||||
adapter.setExpectMessage(true);
|
||||
adapter.afterPropertiesSet();
|
||||
adapter.start();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCustomPayloadSerializer() throws Exception {
|
||||
RedisTemplate<String, Long> template = new RedisTemplate<>();
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.setValueSerializer(new GenericToStringSerializer<>(Long.class));
|
||||
template.setConnectionFactory(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.setSerializer(new GenericToStringSerializer<Long>(Long.class));
|
||||
adapter.afterPropertiesSet();
|
||||
adapter.start();
|
||||
|
||||
template.boundListOps(currentQueueName).rightPush(5l);
|
||||
@SuppressWarnings("unchecked")
|
||||
Message<Long> message = (Message<Long>) messages.poll((int)(1000 * timeoutMultiplier), TimeUnit.MILLISECONDS);
|
||||
assertEquals(5L, (long) message.getPayload());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCustomMessageSerializer() throws Exception {
|
||||
RedisTemplate<String, Message<?>> template = new RedisTemplate<>();
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.setValueSerializer(new TestMessageSerializer());
|
||||
template.setConnectionFactory(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.setSerializer(new TestMessageSerializer());
|
||||
adapter.setExpectMessage(true);
|
||||
adapter.afterPropertiesSet();
|
||||
adapter.start();
|
||||
|
||||
template.boundListOps(currentQueueName).rightPush(new GenericMessage<>(10l));
|
||||
@SuppressWarnings("unchecked")
|
||||
Message<Long> message = (Message<Long>) messages.poll((int)(1000 * timeoutMultiplier), TimeUnit.MILLISECONDS);
|
||||
assertEquals(10L, (long) message.getPayload());
|
||||
}
|
||||
|
||||
private class TestMessageHandler implements MessageHandler {
|
||||
|
||||
@Override
|
||||
public void handleMessage(Message<?> message) throws MessagingException {
|
||||
messages.add(message);
|
||||
}
|
||||
}
|
||||
|
||||
private class TestMessageSerializer implements RedisSerializer<Message<?>> {
|
||||
|
||||
@Override
|
||||
public byte[] serialize(Message<?> t) throws SerializationException {
|
||||
return "Foo".getBytes();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Message<?> deserialize(byte[] bytes) throws SerializationException {
|
||||
return new GenericMessage<>(10l);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,172 +0,0 @@
|
||||
/*
|
||||
* Copyright 2002-2013 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
|
||||
*
|
||||
* http://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.cloud.stream.binder.redis;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.cloud.stream.binder.BinderTestUtils;
|
||||
import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.serializer.GenericToStringSerializer;
|
||||
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
|
||||
import org.springframework.data.redis.serializer.RedisSerializer;
|
||||
import org.springframework.data.redis.serializer.SerializationException;
|
||||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||||
import org.springframework.integration.redis.outbound.RedisQueueOutboundChannelAdapter;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
/**
|
||||
* Integration test of {@link RedisQueueOutboundChannelAdapter}
|
||||
*
|
||||
* @author Jennifer Hickey
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public class RedisQueueOutboundChannelAdapterTests {
|
||||
|
||||
private static final String QUEUE_NAME = "outboundadaptertest";
|
||||
|
||||
private RedisConnectionFactory connectionFactory;
|
||||
|
||||
private RedisQueueOutboundChannelAdapter adapter;
|
||||
|
||||
@Rule
|
||||
public RedisTestSupport redisAvailableRule = new RedisTestSupport();
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
this.connectionFactory = redisAvailableRule.getResource();
|
||||
adapter = new RedisQueueOutboundChannelAdapter(QUEUE_NAME, connectionFactory);
|
||||
adapter.setBeanFactory(BinderTestUtils.MOCK_BF);
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() {
|
||||
connectionFactory.getConnection().del(QUEUE_NAME.getBytes());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDefaultPayloadSerializer() throws Exception {
|
||||
StringRedisTemplate template = new StringRedisTemplate(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.afterPropertiesSet();
|
||||
adapter.handleMessage(new GenericMessage<String>("message1"));
|
||||
assertEquals("message1", template.boundListOps(QUEUE_NAME).rightPop());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDefaultMsgSerializer() throws Exception {
|
||||
RedisTemplate<String, Message<?>> template = new RedisTemplate<String, Message<?>>();
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.setValueSerializer(new JdkSerializationRedisSerializer());
|
||||
template.setConnectionFactory(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.setExtractPayload(false);
|
||||
adapter.afterPropertiesSet();
|
||||
|
||||
Map<String, Object> headers = new HashMap<String, Object>();
|
||||
headers.put("header1", "foo");
|
||||
adapter.handleMessage(new GenericMessage<String>("message2", headers));
|
||||
Message<?> message = template.boundListOps(QUEUE_NAME).rightPop();
|
||||
assertEquals("message2", message.getPayload());
|
||||
assertEquals("foo", message.getHeaders().get("header1"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testNoSerializer() throws Exception {
|
||||
RedisTemplate<String, byte[]> template = new RedisTemplate<String, byte[]>();
|
||||
template.setEnableDefaultSerializer(false);
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.setConnectionFactory(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.afterPropertiesSet();
|
||||
|
||||
adapter.handleMessage(new GenericMessage<byte[]>("message3".getBytes()));
|
||||
byte[] value = template.boundListOps(QUEUE_NAME).rightPop();
|
||||
assertEquals("message3", new String(value));
|
||||
}
|
||||
|
||||
@Test(expected = IllegalArgumentException.class)
|
||||
public void testNoSerializerNoExtractPayload() throws Exception {
|
||||
RedisTemplate<String, byte[]> template = new RedisTemplate<String, byte[]>();
|
||||
template.setEnableDefaultSerializer(false);
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.setConnectionFactory(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.setSerializer(null);
|
||||
adapter.setExtractPayload(false);
|
||||
adapter.afterPropertiesSet();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCustomPayloadSerializer() throws Exception {
|
||||
RedisTemplate<String, Long> template = new RedisTemplate<String, Long>();
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.setValueSerializer(new GenericToStringSerializer<Long>(Long.class));
|
||||
template.setConnectionFactory(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.setSerializer(new GenericToStringSerializer<Long>(Long.class));
|
||||
adapter.afterPropertiesSet();
|
||||
|
||||
adapter.handleMessage(new GenericMessage<Long>(5l));
|
||||
assertEquals(Long.valueOf(5), template.boundListOps(QUEUE_NAME).rightPop());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCustomMessageSerializer() throws Exception {
|
||||
RedisTemplate<String, Message<?>> template = new RedisTemplate<String, Message<?>>();
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.setValueSerializer(new TestMessageSerializer());
|
||||
template.setConnectionFactory(connectionFactory);
|
||||
template.afterPropertiesSet();
|
||||
|
||||
adapter.setSerializer(new TestMessageSerializer());
|
||||
adapter.setExtractPayload(false);
|
||||
adapter.afterPropertiesSet();
|
||||
|
||||
Message<?> message = template.boundListOps(QUEUE_NAME).rightPop();
|
||||
assertEquals(10l, message.getPayload());
|
||||
}
|
||||
|
||||
private class TestMessageSerializer implements RedisSerializer<Message<?>> {
|
||||
|
||||
@Override
|
||||
public byte[] serialize(Message<?> t) throws SerializationException {
|
||||
return "Foo".getBytes();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Message<?> deserialize(byte[] bytes) throws SerializationException {
|
||||
return new GenericMessage<Long>(10l);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,71 +0,0 @@
|
||||
/*
|
||||
* Copyright 2014-2016 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
|
||||
*
|
||||
* http://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.cloud.stream.binder.redis;
|
||||
|
||||
import org.springframework.cloud.stream.binder.AbstractTestBinder;
|
||||
import org.springframework.cloud.stream.binder.ConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.context.support.GenericApplicationContext;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.integration.channel.DefaultHeaderChannelRegistry;
|
||||
import org.springframework.integration.codec.kryo.PojoCodec;
|
||||
import org.springframework.integration.context.IntegrationContextUtils;
|
||||
import org.springframework.integration.support.DefaultMessageBuilderFactory;
|
||||
import org.springframework.integration.support.utils.IntegrationUtils;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
||||
|
||||
/**
|
||||
* Test support class for {@link RedisMessageChannelBinder}.
|
||||
*
|
||||
* @author Ilayaperumal Gopinathan
|
||||
* @author Gary Russell
|
||||
* @author David Turanski
|
||||
*/
|
||||
public class RedisTestBinder extends AbstractTestBinder<RedisMessageChannelBinder, ConsumerProperties, ProducerProperties> {
|
||||
|
||||
private StringRedisTemplate template;
|
||||
|
||||
public RedisTestBinder(RedisConnectionFactory connectionFactory) {
|
||||
RedisMessageChannelBinder binder = new RedisMessageChannelBinder(connectionFactory);
|
||||
GenericApplicationContext context = new GenericApplicationContext();
|
||||
context.getBeanFactory().registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
|
||||
new DefaultMessageBuilderFactory());
|
||||
DefaultHeaderChannelRegistry channelRegistry = new DefaultHeaderChannelRegistry();
|
||||
channelRegistry.setReaperDelay(Long.MAX_VALUE);
|
||||
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
|
||||
taskScheduler.afterPropertiesSet();
|
||||
channelRegistry.setTaskScheduler(taskScheduler);
|
||||
context.getBeanFactory().registerSingleton(
|
||||
IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME,
|
||||
channelRegistry);
|
||||
context.refresh();
|
||||
binder.setApplicationContext(context);
|
||||
binder.setCodec(new PojoCodec());
|
||||
setBinder(binder);
|
||||
template = new StringRedisTemplate(connectionFactory);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cleanup() {
|
||||
if (!queues.isEmpty()) {
|
||||
for (String queue : queues) {
|
||||
template.delete(queue);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,159 +0,0 @@
|
||||
/*
|
||||
* Copyright 2015-2016 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
|
||||
*
|
||||
* http://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.cloud.stream.binder.redis.integration;
|
||||
|
||||
import static org.hamcrest.CoreMatchers.equalTo;
|
||||
import static org.hamcrest.CoreMatchers.instanceOf;
|
||||
import static org.hamcrest.CoreMatchers.is;
|
||||
import static org.hamcrest.CoreMatchers.not;
|
||||
import static org.hamcrest.collection.IsMapContaining.hasKey;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertThat;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.actuate.health.CompositeHealthIndicator;
|
||||
import org.springframework.boot.actuate.health.HealthIndicator;
|
||||
import org.springframework.boot.actuate.health.Status;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.binder.Binder;
|
||||
import org.springframework.cloud.stream.binder.BinderFactory;
|
||||
import org.springframework.cloud.stream.binder.redis.RedisMessageChannelBinder;
|
||||
import org.springframework.cloud.stream.messaging.Processor;
|
||||
import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
|
||||
/**
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
public class RedisBinderModuleTests {
|
||||
|
||||
@ClassRule
|
||||
public static RedisTestSupport redisTestSupport = new RedisTestSupport();
|
||||
|
||||
private ConfigurableApplicationContext context = null;
|
||||
|
||||
public static final RedisConnectionFactory MOCK_CONNECTION_FACTORY = Mockito.mock(RedisConnectionFactory.class,
|
||||
Mockito.RETURNS_MOCKS);
|
||||
|
||||
@After
|
||||
public void tearDown() {
|
||||
if (context != null) {
|
||||
context.close();
|
||||
context = null;
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testParentConnectionFactoryInheritedByDefault() {
|
||||
context = SpringApplication.run(SimpleProcessor.class, "--server.port=0");
|
||||
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
|
||||
Binder binder = binderFactory.getBinder(null);
|
||||
assertThat(binder, instanceOf(RedisMessageChannelBinder.class));
|
||||
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
|
||||
RedisConnectionFactory binderConnectionFactory =
|
||||
(RedisConnectionFactory) binderFieldAccessor.getPropertyValue("connectionFactory");
|
||||
assertThat(binderConnectionFactory, instanceOf(RedisConnectionFactory.class));
|
||||
RedisConnectionFactory connectionFactory = context.getBean(RedisConnectionFactory.class);
|
||||
assertThat(binderConnectionFactory, is(connectionFactory));
|
||||
CompositeHealthIndicator bindersHealthIndicator =
|
||||
context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
|
||||
assertNotNull(bindersHealthIndicator);
|
||||
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String,HealthIndicator> healthIndicators =
|
||||
(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
|
||||
assertThat(healthIndicators, hasKey("redis"));
|
||||
assertThat(healthIndicators.get("redis").health().getStatus(), equalTo(Status.UP));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testParentConnectionFactoryInheritedIfOverridden() {
|
||||
context = new SpringApplication(SimpleProcessor.class, ConnectionFactoryConfiguration.class).run();
|
||||
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
|
||||
Binder binder = binderFactory.getBinder(null);
|
||||
assertThat(binder, instanceOf(RedisMessageChannelBinder.class));
|
||||
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
|
||||
RedisConnectionFactory binderConnectionFactory =
|
||||
(RedisConnectionFactory) binderFieldAccessor.getPropertyValue("connectionFactory");
|
||||
assertThat(binderConnectionFactory, is(MOCK_CONNECTION_FACTORY));
|
||||
RedisConnectionFactory connectionFactory = context.getBean(RedisConnectionFactory.class);
|
||||
assertThat(binderConnectionFactory, is(connectionFactory));
|
||||
CompositeHealthIndicator bindersHealthIndicator =
|
||||
context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
|
||||
assertNotNull(bindersHealthIndicator);
|
||||
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String,HealthIndicator> healthIndicators =
|
||||
(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
|
||||
assertThat(healthIndicators, hasKey("redis"));
|
||||
assertThat(healthIndicators.get("redis").health().getStatus(), equalTo(Status.UP));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testParentConnectionFactoryNotInheritedByCustomizedBinders() {
|
||||
List<String> params = new ArrayList<>();
|
||||
params.add("--spring.cloud.stream.input.binder=custom");
|
||||
params.add("--spring.cloud.stream.output.binder=custom");
|
||||
params.add("--spring.cloud.stream.binders.custom.type=redis");
|
||||
params.add("--spring.cloud.stream.binders.custom.environment.foo=bar");
|
||||
context = SpringApplication.run(SimpleProcessor.class, params.toArray(new String[]{}));
|
||||
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
|
||||
Binder binder = binderFactory.getBinder(null);
|
||||
assertThat(binder, instanceOf(RedisMessageChannelBinder.class));
|
||||
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
|
||||
RedisConnectionFactory binderConnectionFactory =
|
||||
(RedisConnectionFactory) binderFieldAccessor.getPropertyValue("connectionFactory");
|
||||
RedisConnectionFactory connectionFactory = context.getBean(RedisConnectionFactory.class);
|
||||
assertThat(binderConnectionFactory, not(is(connectionFactory)));
|
||||
CompositeHealthIndicator bindersHealthIndicator =
|
||||
context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
|
||||
assertNotNull(bindersHealthIndicator);
|
||||
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String,HealthIndicator> healthIndicators =
|
||||
(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
|
||||
assertThat(healthIndicators, hasKey("custom"));
|
||||
assertThat(healthIndicators.get("custom").health().getStatus(), equalTo(Status.UP));
|
||||
}
|
||||
|
||||
@EnableBinding(Processor.class)
|
||||
@SpringBootApplication
|
||||
public static class SimpleProcessor {
|
||||
|
||||
}
|
||||
|
||||
public static class ConnectionFactoryConfiguration {
|
||||
|
||||
@Bean
|
||||
public RedisConnectionFactory connectionFactory() {
|
||||
return MOCK_CONNECTION_FACTORY;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -14,9 +14,8 @@
|
||||
<version>1.0.0.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
<modules>
|
||||
<module>spring-cloud-starter-stream-rabbit</module>
|
||||
<module>spring-cloud-starter-stream-redis</module>
|
||||
<module>spring-cloud-starter-stream-kafka</module>
|
||||
<module>spring-cloud-starter-stream-rabbit</module>
|
||||
</modules>
|
||||
<properties>
|
||||
<spring-cloud-commons.version>1.1.0.RC1</spring-cloud-commons.version>
|
||||
|
||||
@@ -1,26 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-starters</artifactId>
|
||||
<version>1.0.0.BUILD-SNAPSHOT</version>
|
||||
<relativePath>..</relativePath>
|
||||
</parent>
|
||||
<artifactId>spring-cloud-starter-stream-redis</artifactId>
|
||||
<description>Spring Cloud Starter Stream Redis</description>
|
||||
<url>http://projects.spring.io/spring-cloud</url>
|
||||
<organization>
|
||||
<name>Pivotal Software, Inc.</name>
|
||||
<url>http://www.spring.io</url>
|
||||
</organization>
|
||||
<properties>
|
||||
<main.basedir>${basedir}/../..</main.basedir>
|
||||
</properties>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-redis</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -1 +0,0 @@
|
||||
provides: spring-cloud-stream-binder-redis
|
||||
Reference in New Issue
Block a user