From 2b397b45a7ae2291aa37b24eb0c1eccae43f8e04 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 17 Mar 2016 08:53:14 -0400 Subject: [PATCH] Moving the Redis Binder out to its own repo removed redis starter --- spring-cloud-stream-binders/pom.xml | 9 +- .../spring-cloud-stream-binder-redis/pom.xml | 64 --- .../redis/RedisMessageChannelBinder.java | 408 ------------------ ...edisMessageChannelBinderConfiguration.java | 63 --- .../config/RedisServiceAutoConfiguration.java | 77 ---- .../stream/binder/redis/package-info.java | 5 - .../redis-binder.properties | 5 - .../main/resources/META-INF/spring.binders | 2 - .../redis/AbstractRedisSerializerTests.java | 156 ------- .../stream/binder/redis/RedisBinderTests.java | 259 ----------- .../RedisPublishingMessageHandlerTests.java | 145 ------- .../RedisQueueInboundChannelAdapterTests.java | 224 ---------- ...RedisQueueOutboundChannelAdapterTests.java | 172 -------- .../stream/binder/redis/RedisTestBinder.java | 71 --- .../integration/RedisBinderModuleTests.java | 159 ------- spring-cloud-stream-starters/pom.xml | 3 +- .../spring-cloud-starter-stream-redis/pom.xml | 26 -- .../main/resources/META-INF/spring.provides | 1 - 18 files changed, 5 insertions(+), 1844 deletions(-) delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/pom.xml delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisMessageChannelBinderConfiguration.java delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisServiceAutoConfiguration.java delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/package-info.java delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring-cloud-stream/redis-binder.properties delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.binders delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/AbstractRedisSerializerTests.java delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisPublishingMessageHandlerTests.java delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisQueueInboundChannelAdapterTests.java delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisQueueOutboundChannelAdapterTests.java delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisTestBinder.java delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/integration/RedisBinderModuleTests.java delete mode 100644 spring-cloud-stream-starters/spring-cloud-starter-stream-redis/pom.xml delete mode 100644 spring-cloud-stream-starters/spring-cloud-starter-stream-redis/src/main/resources/META-INF/spring.provides diff --git a/spring-cloud-stream-binders/pom.xml b/spring-cloud-stream-binders/pom.xml index 25969697f..89434b9f1 100644 --- a/spring-cloud-stream-binders/pom.xml +++ b/spring-cloud-stream-binders/pom.xml @@ -1,7 +1,6 @@ 4.0.0 - spring-cloud-stream-binders-parent pom https://github.com/spring-cloud/spring-cloud-stream @@ -9,6 +8,7 @@ Pivotal Software, Inc. http://www.spring.io + org.springframework.cloud spring-cloud-stream-parent @@ -16,12 +16,11 @@ - spring-cloud-stream-binder-test - spring-cloud-stream-binder-rabbit - spring-cloud-stream-binder-redis spring-cloud-stream-binder-kafka + spring-cloud-stream-binder-rabbit + spring-cloud-stream-binder-test - + org.springframework.boot diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/pom.xml b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/pom.xml deleted file mode 100644 index e10c25f54..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/pom.xml +++ /dev/null @@ -1,64 +0,0 @@ - - - 4.0.0 - - spring-cloud-stream-binder-redis - jar - spring-cloud-stream-binder-redis - Redis binder implementation - - - org.springframework.cloud - spring-cloud-stream-binders-parent - 1.0.0.BUILD-SNAPSHOT - - - - - org.springframework.boot - spring-boot-configuration-processor - true - - - org.springframework.cloud - spring-cloud-stream - - - org.springframework.cloud - spring-cloud-stream-codec - - - org.springframework.boot - spring-boot-autoconfigure - true - - - org.springframework.cloud - spring-cloud-core - true - - - org.springframework.cloud - spring-cloud-stream-binder-test - test - - - org.springframework.cloud - spring-cloud-stream-test-support-internal - test - - - org.springframework.boot - spring-boot-starter-redis - - - org.springframework.integration - spring-integration-redis - - - joda-time - joda-time - test - - - diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java deleted file mode 100644 index c1d12fda5..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java +++ /dev/null @@ -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 { - - 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 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 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 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 consumerBinding = new DefaultBinding(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() { - - @Override - public Boolean doWithRetry(RetryContext context) throws Exception { - return bridgeToModuleChannel.send(message, timeout); - } - - }, new RecoveryCallback() { - - /** - * 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 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 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 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 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 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) 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 consumers = new - ArrayList(); - - 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); - } - } - - } - -} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisMessageChannelBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisMessageChannelBinderConfiguration.java deleted file mode 100644 index 4410ebe0d..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisMessageChannelBinderConfiguration.java +++ /dev/null @@ -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; - } -} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisServiceAutoConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisServiceAutoConfiguration.java deleted file mode 100644 index 56ca0d27a..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisServiceAutoConfiguration.java +++ /dev/null @@ -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); - } -} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/package-info.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/package-info.java deleted file mode 100644 index 5c34fc202..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/package-info.java +++ /dev/null @@ -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; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring-cloud-stream/redis-binder.properties b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring-cloud-stream/redis-binder.properties deleted file mode 100644 index dd78f8e81..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring-cloud-stream/redis-binder.properties +++ /dev/null @@ -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 diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.binders b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.binders deleted file mode 100644 index 12ae7ef61..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.binders +++ /dev/null @@ -1,2 +0,0 @@ -redis:\ -org.springframework.cloud.stream.binder.redis.config.RedisServiceAutoConfiguration diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/AbstractRedisSerializerTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/AbstractRedisSerializerTests.java deleted file mode 100644 index deb396a43..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/AbstractRedisSerializerTests.java +++ /dev/null @@ -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 serializer; - - protected abstract RedisSerializer 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 map = new HashMap(); - 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 list = new LinkedList(); - list.add("foo"); - byte[] bytes = serializer.serialize(list); - List obj = (List) serializer.deserialize(bytes); - assertEquals("foo", obj.get(0)); - } - - @Test - public void testSetSerialization() { - Set set = new TreeSet(); - 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; - } - -} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java deleted file mode 100644 index 5d8f331e8..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java +++ /dev/null @@ -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 { - - private final String CLASS_UNDER_TEST_NAME = RedisMessageChannelBinder.class.getSimpleName(); - - @Rule - public RedisTestSupport redisAvailableRule = new RedisTestSupport(); - - private RedisTemplate 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 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 consumerBinding = binder.bindConsumer("props.0", "test", new DirectChannel(), createConsumerProperties()); - Binding producerBinding = binder.bindProducer("props.0", new DirectChannel(), createProducerProperties()); - AbstractEndpoint producerEndpoint = extractEndpoint(producerBinding); - @SuppressWarnings("unchecked") - Map 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 consumerBinding = binder.bindConsumer("retry.0", "test", new DirectChannel(), consumerProperties); // no subscriber - channel.send(new GenericMessage<>("foo")); - RedisTemplate 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 headers = Arrays.asList(TestUtils.getPropertyValue(binder, "headersToMap", String[].class)); - assertEquals(7, headers.size()); - assertTrue(headers.contains("foo")); - assertTrue(headers.contains("bar")); - } - - private RedisTemplate createTemplate() { - if (this.redisTemplate != null) { - return this.redisTemplate; - } - RedisTemplate template = new RedisTemplate(); - 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 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 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(bytes), false).getPayload(); - return new String(bytes, "UTF-8"); - } - - }; - } - - @Override - protected void binderBindUnbindLatency() throws InterruptedException { - Thread.sleep(3000); // needed for Redis see INT-3442 - } - -} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisPublishingMessageHandlerTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisPublishingMessageHandlerTests.java deleted file mode 100644 index 898918522..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisPublishingMessageHandlerTests.java +++ /dev/null @@ -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 serializer = new GenericToStringSerializer(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. 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(); - } - } - -} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisQueueInboundChannelAdapterTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisQueueInboundChannelAdapterTests.java deleted file mode 100644 index 04a595f6c..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisQueueInboundChannelAdapterTests.java +++ /dev/null @@ -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 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 template = new RedisTemplate<>(); - template.setConnectionFactory(connectionFactory); - template.setKeySerializer(new StringRedisSerializer()); - template.afterPropertiesSet(); - - adapter.afterPropertiesSet(); - adapter.start(); - - template.boundListOps(currentQueueName).rightPush("message1"); - @SuppressWarnings("unchecked") - Message message = (Message) messages.poll((int)(1000 * timeoutMultiplier), TimeUnit.MILLISECONDS); - assertNotNull(message); - assertEquals("message1", message.getPayload()); - } - - @Test - public void testDefaultMsgSerializer() throws Exception { - RedisTemplate> template = new RedisTemplate>(); - template.setKeySerializer(new StringRedisSerializer()); - template.setValueSerializer(new JdkSerializationRedisSerializer()); - template.setConnectionFactory(connectionFactory); - template.afterPropertiesSet(); - - adapter.setExpectMessage(true); - adapter.afterPropertiesSet(); - adapter.start(); - - Map headers = new HashMap<>(); - headers.put("header1", "foo"); - template.boundListOps(currentQueueName).rightPush(new GenericMessage<>("message2", headers)); - @SuppressWarnings("unchecked") - Message message = (Message) 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 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 message = (Message) messages.poll((int)(1000 * timeoutMultiplier), TimeUnit.MILLISECONDS); - assertEquals("message3", new String(message.getPayload())); - } - - @Test(expected = IllegalArgumentException.class) - public void testNoSerializerNoExtractPayload() throws Exception { - RedisTemplate 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 template = new RedisTemplate<>(); - template.setKeySerializer(new StringRedisSerializer()); - template.setValueSerializer(new GenericToStringSerializer<>(Long.class)); - template.setConnectionFactory(connectionFactory); - template.afterPropertiesSet(); - - adapter.setSerializer(new GenericToStringSerializer(Long.class)); - adapter.afterPropertiesSet(); - adapter.start(); - - template.boundListOps(currentQueueName).rightPush(5l); - @SuppressWarnings("unchecked") - Message message = (Message) messages.poll((int)(1000 * timeoutMultiplier), TimeUnit.MILLISECONDS); - assertEquals(5L, (long) message.getPayload()); - } - - @Test - public void testCustomMessageSerializer() throws Exception { - RedisTemplate> 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 message = (Message) 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> { - - @Override - public byte[] serialize(Message t) throws SerializationException { - return "Foo".getBytes(); - } - - @Override - public Message deserialize(byte[] bytes) throws SerializationException { - return new GenericMessage<>(10l); - } - } -} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisQueueOutboundChannelAdapterTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisQueueOutboundChannelAdapterTests.java deleted file mode 100644 index f43402561..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisQueueOutboundChannelAdapterTests.java +++ /dev/null @@ -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("message1")); - assertEquals("message1", template.boundListOps(QUEUE_NAME).rightPop()); - } - - @Test - public void testDefaultMsgSerializer() throws Exception { - RedisTemplate> template = new RedisTemplate>(); - template.setKeySerializer(new StringRedisSerializer()); - template.setValueSerializer(new JdkSerializationRedisSerializer()); - template.setConnectionFactory(connectionFactory); - template.afterPropertiesSet(); - - adapter.setExtractPayload(false); - adapter.afterPropertiesSet(); - - Map headers = new HashMap(); - headers.put("header1", "foo"); - adapter.handleMessage(new GenericMessage("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 template = new RedisTemplate(); - template.setEnableDefaultSerializer(false); - template.setKeySerializer(new StringRedisSerializer()); - template.setConnectionFactory(connectionFactory); - template.afterPropertiesSet(); - - adapter.afterPropertiesSet(); - - adapter.handleMessage(new GenericMessage("message3".getBytes())); - byte[] value = template.boundListOps(QUEUE_NAME).rightPop(); - assertEquals("message3", new String(value)); - } - - @Test(expected = IllegalArgumentException.class) - public void testNoSerializerNoExtractPayload() throws Exception { - RedisTemplate template = new RedisTemplate(); - 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 template = new RedisTemplate(); - template.setKeySerializer(new StringRedisSerializer()); - template.setValueSerializer(new GenericToStringSerializer(Long.class)); - template.setConnectionFactory(connectionFactory); - template.afterPropertiesSet(); - - adapter.setSerializer(new GenericToStringSerializer(Long.class)); - adapter.afterPropertiesSet(); - - adapter.handleMessage(new GenericMessage(5l)); - assertEquals(Long.valueOf(5), template.boundListOps(QUEUE_NAME).rightPop()); - } - - @Test - public void testCustomMessageSerializer() throws Exception { - RedisTemplate> template = new RedisTemplate>(); - 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> { - - @Override - public byte[] serialize(Message t) throws SerializationException { - return "Foo".getBytes(); - } - - @Override - public Message deserialize(byte[] bytes) throws SerializationException { - return new GenericMessage(10l); - } - } -} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisTestBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisTestBinder.java deleted file mode 100644 index 98233b5bd..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisTestBinder.java +++ /dev/null @@ -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 { - - 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); - } - } - } -} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/integration/RedisBinderModuleTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/integration/RedisBinderModuleTests.java deleted file mode 100644 index 8bed037f3..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/integration/RedisBinderModuleTests.java +++ /dev/null @@ -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 healthIndicators = - (Map) 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 healthIndicators = - (Map) directFieldAccessor.getPropertyValue("indicators"); - assertThat(healthIndicators, hasKey("redis")); - assertThat(healthIndicators.get("redis").health().getStatus(), equalTo(Status.UP)); - } - - @Test - public void testParentConnectionFactoryNotInheritedByCustomizedBinders() { - List 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 healthIndicators = - (Map) 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; - } - } -} diff --git a/spring-cloud-stream-starters/pom.xml b/spring-cloud-stream-starters/pom.xml index 1a3f78639..c99adaac3 100644 --- a/spring-cloud-stream-starters/pom.xml +++ b/spring-cloud-stream-starters/pom.xml @@ -14,9 +14,8 @@ 1.0.0.BUILD-SNAPSHOT - spring-cloud-starter-stream-rabbit - spring-cloud-starter-stream-redis spring-cloud-starter-stream-kafka + spring-cloud-starter-stream-rabbit 1.1.0.RC1 diff --git a/spring-cloud-stream-starters/spring-cloud-starter-stream-redis/pom.xml b/spring-cloud-stream-starters/spring-cloud-starter-stream-redis/pom.xml deleted file mode 100644 index 8e60f2e9e..000000000 --- a/spring-cloud-stream-starters/spring-cloud-starter-stream-redis/pom.xml +++ /dev/null @@ -1,26 +0,0 @@ - - - 4.0.0 - - org.springframework.cloud - spring-cloud-stream-starters - 1.0.0.BUILD-SNAPSHOT - .. - - spring-cloud-starter-stream-redis - Spring Cloud Starter Stream Redis - http://projects.spring.io/spring-cloud - - Pivotal Software, Inc. - http://www.spring.io - - - ${basedir}/../.. - - - - org.springframework.cloud - spring-cloud-stream-binder-redis - - - diff --git a/spring-cloud-stream-starters/spring-cloud-starter-stream-redis/src/main/resources/META-INF/spring.provides b/spring-cloud-stream-starters/spring-cloud-starter-stream-redis/src/main/resources/META-INF/spring.provides deleted file mode 100644 index 6b5eff469..000000000 --- a/spring-cloud-stream-starters/spring-cloud-starter-stream-redis/src/main/resources/META-INF/spring.provides +++ /dev/null @@ -1 +0,0 @@ -provides: spring-cloud-stream-binder-redis \ No newline at end of file