From 213cafb2da0e3b977fbacd32786ec8b7149c6d5e Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 31 Oct 2013 21:51:50 +0200 Subject: [PATCH] INT-3014: Introduce Redis Queue Adapters https://jira.springsource.org/browse/INT-3014 https://jira.springsource.org/browse/INT-3015 * Introduce `RedisQueueMessageDrivenEndpoint` and `RedisQueueOutboundChannelAdapter` * add Tests for them * Remove `ChannelRegistry` artifacts --- .../channel/registry/ChannelRegistry.java | 50 ---- .../registry/LocalChannelRegistry.java | 168 ------------ .../channel/registry/package-info.java | 4 - .../registry/LocalChannelRegistryTests.java | 173 ------------- .../RedisQueueMessageDrivenEndpoint.java | 245 ++++++++++++++++++ .../RedisQueueOutboundChannelAdapter.java | 112 ++++++++ .../RedisQueueMessageDrivenEndpointTests.java | 157 +++++++++++ ...RedisQueueOutboundChannelAdapterTests.java | 143 ++++++++++ .../src/test/resources/log4j.properties | 8 + 9 files changed, 665 insertions(+), 395 deletions(-) delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/registry/ChannelRegistry.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/registry/LocalChannelRegistry.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/registry/package-info.java delete mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/registry/LocalChannelRegistryTests.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java create mode 100644 spring-integration-redis/src/test/resources/log4j.properties diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/ChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/ChannelRegistry.java deleted file mode 100644 index 6cc15c015d..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/ChannelRegistry.java +++ /dev/null @@ -1,50 +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.integration.channel.registry; - -import org.springframework.integration.MessageChannel; - -/** - * A strategy interface used to bind a {@link MessageChannel} to a logical name. The name - * is intended to identify a logical consumer or producer of messages. This may be a - * queue, a channel adapter, another message channel, a Spring bean, etc. - * - * @author Mark Fisher - * @author David Turanski - * @since 3.0 - */ -public interface ChannelRegistry { - - /** - * Register a message consumer - * @param name the logical identity of the message source - * @param channel the channel bound as a consumer - */ - void inbound(String name, MessageChannel channel); - - /** - * Register a message producer - * @param name the logical identity of the message target - * @param channel the channel bound as a producer - */ - void outbound(String name, MessageChannel channel); - - /** - * Create a tap on an already registered inbound channel - * @param name the registered name - * @param channel the channel that will receive messages from the tap - */ - void tap(String name, MessageChannel channel); - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/LocalChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/LocalChannelRegistry.java deleted file mode 100644 index 35c960b6a0..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/LocalChannelRegistry.java +++ /dev/null @@ -1,168 +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.integration.channel.registry; - -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; -import org.springframework.context.support.AbstractApplicationContext; -import org.springframework.integration.MessageChannel; -import org.springframework.integration.channel.AbstractMessageChannel; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.channel.interceptor.WireTap; -import org.springframework.integration.core.SubscribableChannel; -import org.springframework.integration.handler.BridgeHandler; -import org.springframework.util.Assert; - -/** - * A simple implementation of {@link ChannelRegistry} for in-process use. For inbound and - * outbound, creates a {@link DirectChannel} and bridges the passed - * {@link MessageChannel} to the channel which is registered in the given application - * context. If that channel does not yet exist, it will be created. For tap, it adds a - * {@link WireTap} for an inbound channel whose name matches the one provided. If no such - * inbound channel exists at the time of the method invocation, it will throw an - * Exception. Otherwise the provided channel instance will receive messages from the wire - * tap on that inbound channel. - * - * @author David Turanski - * @author Mark Fisher - * @since 3.0 - */ -public class LocalChannelRegistry implements ChannelRegistry, ApplicationContextAware, InitializingBean { - - private volatile AbstractApplicationContext applicationContext; - - @Override - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - Assert.isInstanceOf(AbstractApplicationContext.class, applicationContext); - this.applicationContext = (AbstractApplicationContext) applicationContext; - } - - @Override - public void afterPropertiesSet() throws Exception { - Assert.notNull(applicationContext, "The 'applicationContext' property cannot be null"); - } - - /** - * Looks up or creates a DirectChannel with the given name and creates a bridge from - * that channel to the provided channel instance. Also registers a wire tap if the - * channel for the given name had been created. The target of the wire tap is a - * publish-subscribe channel. - */ - @Override - public void inbound(String name, MessageChannel channel) { - Assert.hasText(name, "a valid name is required to register an inbound channel"); - Assert.notNull(channel, "channel must not be null"); - DirectChannel registeredChannel = lookupOrCreateSharedChannel(name, DirectChannel.class); - bridge(registeredChannel, channel); - createSharedTapChannelIfNecessary(registeredChannel); - } - - /** - * Looks up or creates a DirectChannel with the given name and creates a bridge to - * that channel from the provided channel instance. - */ - @Override - public void outbound(String name, MessageChannel channel) { - Assert.hasText(name, "a valid name is required to register an outbound channel"); - Assert.notNull(channel, "channel must not be null"); - Assert.isTrue(channel instanceof SubscribableChannel, - "channel must be of type " + SubscribableChannel.class.getName()); - DirectChannel registeredChannel = lookupOrCreateSharedChannel(name, DirectChannel.class); - bridge((SubscribableChannel) channel, registeredChannel); - } - - /** - * Looks up a wiretap for the inbound channel with the given name and creates a - * bridge from that wiretap's output channel to the provided channel instance. - * Will throw an Exception if no such wiretap exists. - */ - @Override - public void tap(String name, MessageChannel channel) { - Assert.hasText(name, "a valid name is required to register a tap channel"); - Assert.notNull(channel, "channel must not be null"); - SubscribableChannel tapChannel = null; - String tapName = name + ".tap"; - try { - tapChannel = applicationContext.getBean(tapName, SubscribableChannel.class); - } - catch (Exception e) { - throw new IllegalArgumentException("No tap channel exists for '" + name - + "'. A tap is only valid for a registered inbound channel."); - } - bridge(tapChannel, channel); - } - - protected synchronized T lookupOrCreateSharedChannel(String name, Class requiredType) { - T channel = null; - if (applicationContext.containsBean(name)) { - try { - channel = applicationContext.getBean(name, requiredType); - } - catch (Exception e) { - throw new IllegalArgumentException("bean '" + name - + "' is already registered but does not match the required type"); - } - } - else { - channel = createSharedChannel(name, requiredType); - } - return channel; - } - - protected T createSharedChannel(String name, Class requiredType) { - try { - T channel = requiredType.newInstance(); - channel.setComponentName(name); - channel.setBeanFactory(applicationContext); - channel.setBeanName(name); - channel.afterPropertiesSet(); - applicationContext.getBeanFactory().registerSingleton(name, channel); - return channel; - } - catch (Exception e) { - throw new IllegalArgumentException("failed to create channel: " + name, e); - } - } - - private synchronized void createSharedTapChannelIfNecessary(AbstractMessageChannel channel) { - String tapName = channel.getComponentName() + ".tap"; - PublishSubscribeChannel tapChannel = null; - if (!applicationContext.containsBean(tapName)) { - tapChannel = createSharedChannel(tapName, PublishSubscribeChannel.class); - WireTap wireTap = new WireTap(tapChannel); - channel.addInterceptor(wireTap); - } - else { - try { - tapChannel = applicationContext.getBean(tapName, PublishSubscribeChannel.class); - } - catch (Exception e) { - throw new IllegalArgumentException("bean '" + tapName - + "' is already registered but does not match the required type"); - } - } - } - - protected BridgeHandler bridge(SubscribableChannel from, MessageChannel to) { - BridgeHandler handler = new BridgeHandler(); - handler.setOutputChannel(to); - handler.afterPropertiesSet(); - from.subscribe(handler); - return handler; - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/package-info.java deleted file mode 100644 index d9b84f22f3..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/package-info.java +++ /dev/null @@ -1,4 +0,0 @@ -/** - * Provides classes representing channel registries. - */ -package org.springframework.integration.channel.registry; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/LocalChannelRegistryTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/LocalChannelRegistryTests.java deleted file mode 100644 index be02be4342..0000000000 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/LocalChannelRegistryTests.java +++ /dev/null @@ -1,173 +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.integration.channel.registry; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; - -import java.util.concurrent.atomic.AtomicBoolean; - -import org.junit.Before; -import org.junit.Test; - -import org.springframework.context.ApplicationContext; -import org.springframework.context.support.AbstractApplicationContext; -import org.springframework.context.support.GenericApplicationContext; -import org.springframework.integration.Message; -import org.springframework.integration.MessageChannel; -import org.springframework.integration.MessagingException; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.core.MessageHandler; -import org.springframework.integration.core.SubscribableChannel; -import org.springframework.integration.message.GenericMessage; - -/** - * @author David Turanski - * @author Mark Fisher - * @since 3.0 - */ -public class LocalChannelRegistryTests { - - private LocalChannelRegistry registry = new LocalChannelRegistry(); - - private AbstractApplicationContext context = new GenericApplicationContext(); - - @Before - public void setUp() { - context.refresh(); - registry.setApplicationContext(context); - } - - @Test - public void testInbound() { - DirectChannel channel = new DirectChannel(); - registry.inbound("inbound", channel); - assertTrue(context.containsBean("inbound")); - - final AtomicBoolean messageReceived = new AtomicBoolean(); - channel.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - messageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - SubscribableChannel registeredChannel = context.getBean("inbound", SubscribableChannel.class); - registeredChannel.send(new GenericMessage("hello")); - assertTrue(messageReceived.get()); - } - - @Test - public void testOutbound() { - DirectChannel channel = new DirectChannel(); - registry.outbound("outbound", channel); - assertTrue(context.containsBean("outbound")); - - final AtomicBoolean messageReceived = new AtomicBoolean(); - SubscribableChannel registeredChannel = context.getBean("outbound", SubscribableChannel.class); - registeredChannel.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - messageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - channel.send(new GenericMessage("hello")); - assertTrue(messageReceived.get()); - } - - @Test(expected = IllegalArgumentException.class) - public void testOutboundTapShouldFail() { - DirectChannel channel = new DirectChannel(); - registry.outbound("outbound", channel); - DirectChannel tapChannel = new DirectChannel(); - registry.tap("outbound", tapChannel); - } - - @Test - public void testInboundTap() { - DirectChannel channel = new DirectChannel(); - registry.inbound("inbound", channel); - DirectChannel tapChannel = new DirectChannel(); - registry.tap("inbound", tapChannel); - final AtomicBoolean originalMessageReceived = new AtomicBoolean(); - final AtomicBoolean tapMessageReceived = new AtomicBoolean(); - tapChannel.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - tapMessageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - channel.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - originalMessageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - MessageChannel registeredChannel = context.getBean("inbound", MessageChannel.class); - registeredChannel.send(new GenericMessage("hello")); - assertTrue(originalMessageReceived.get()); - assertTrue(tapMessageReceived.get()); - } - - @Test - public void testFlowThroughRegisteredChannelFromOutboundToInbound() { - DirectChannel outbound = new DirectChannel(); - DirectChannel inbound = new DirectChannel(); - registry.outbound("foo", outbound); - registry.inbound("foo", inbound); - final AtomicBoolean messageReceived = new AtomicBoolean(); - inbound.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - messageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - outbound.send(new GenericMessage("hello")); - assertTrue(messageReceived.get()); - } - - @Test - public void testFlowThroughRegisteredChannelFromOutboundToInboundWithTap() { - DirectChannel outbound = new DirectChannel(); - DirectChannel inbound = new DirectChannel(); - DirectChannel tap = new DirectChannel(); - registry.outbound("foo", outbound); - registry.inbound("foo", inbound); - registry.tap("foo", tap); - final AtomicBoolean originalMessageReceived = new AtomicBoolean(); - final AtomicBoolean tapMessageReceived = new AtomicBoolean(); - inbound.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - originalMessageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - tap.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - tapMessageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - outbound.send(new GenericMessage("hello")); - assertTrue(originalMessageReceived.get()); - assertTrue(tapMessageReceived.get()); - } - -} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java new file mode 100644 index 0000000000..587d2277e3 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java @@ -0,0 +1,245 @@ +/* + * 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.integration.redis.inbound; + +import java.util.concurrent.TimeUnit; + +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.BoundListOperations; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.MessagingException; +import org.springframework.integration.channel.MessagePublishingErrorHandler; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.util.ErrorHandlingTaskExecutor; +import org.springframework.jmx.export.annotation.ManagedMetric; +import org.springframework.jmx.export.annotation.ManagedOperation; +import org.springframework.jmx.export.annotation.ManagedResource; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + * @author Gunnar Hillert + * @author Artem Bilan + * @since 3.0 + */ +@ManagedResource +public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport { + + public static final long DEFAULT_RECEIVE_TIMEOUT = 1000; + + private final BoundListOperations boundListOperations; + + private MessageChannel errorChannel; + + private volatile TaskExecutor taskExecutor; + + private volatile RedisSerializer serializer = new JdkSerializationRedisSerializer(); + + private volatile boolean expectMessage = false; + + private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; + + private volatile boolean active; + + private volatile boolean listening; + + /** + * @param queueName Must not be an empty String + * @param connectionFactory Must not be null + */ + public RedisQueueMessageDrivenEndpoint(String queueName, RedisConnectionFactory connectionFactory) { + Assert.hasText(queueName, "'queueName' is required"); + Assert.notNull(connectionFactory, "'connectionFactory' must not be null"); + RedisTemplate template = new RedisTemplate(); + template.setConnectionFactory(connectionFactory); + template.setEnableDefaultSerializer(false); + template.setKeySerializer(new StringRedisSerializer()); + template.afterPropertiesSet(); + this.boundListOperations = template.boundListOps(queueName); + } + + public void setSerializer(RedisSerializer serializer) { + this.serializer = serializer; + } + + /** + * When data is retrieved from the Redis queue, does the returned data represent + * just the payload for a Message, or does the data represent a serialized + * {@link Message}?. {@code expectMessage} defaults to false. This means + * the retrieved data will be used as the payload for a new Spring Integration + * Message. Otherwise, the data is deserialized as Spring Integration + * Message. + * + * @param expectMessage Defaults to false + */ + public void setExpectMessage(boolean expectMessage) { + this.expectMessage = expectMessage; + } + + /** + * This timeout (milliseconds) is used when retrieving elements from the queue + * specified by {@link #boundListOperations}. + *

+ * If the queue does contain elements, the data is retrieved immediately. However, + * if the queue is empty, the Redis connection is blocked until either an element + * can be retrieved from the queue or until the specified timeout passes. + *

+ * A timeout of zero can be used to block indefinitely. If not set explicitly + * the timeout value will default to {@code 1000} + *

+ * See also: http://redis.io/commands/brpop + * + * @param receiveTimeout Must be non-negative. Specified in milliseconds. + */ + public void setReceiveTimeout(long receiveTimeout) { + Assert.isTrue(receiveTimeout > 0, "'receiveTimeout' must be > 0."); + this.receiveTimeout = receiveTimeout; + } + + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + @Override + public void setErrorChannel(MessageChannel errorChannel) { + super.setErrorChannel(errorChannel); + this.errorChannel = errorChannel; + } + + @Override + protected void onInit() { + super.onInit(); + if (this.expectMessage) { + Assert.notNull(this.serializer, "'serializer' has to be provided where 'expectMessage == true'."); + } + if (this.taskExecutor == null) { + String beanName = this.getComponentName(); + this.taskExecutor = new SimpleAsyncTaskExecutor((beanName == null ? "" : beanName + "-") + this.getComponentType()); + } + if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) { + MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler(); + errorHandler.setDefaultErrorChannel(this.errorChannel); + this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler); + } + } + + @Override + public String getComponentType() { + return "int-redis:message-driven-channel-adapter"; + } + + @SuppressWarnings("unchecked") + private void popMessageAndSend() { + Message message = null; + + byte[] value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS); + + if (value != null) { + if (this.expectMessage) { + try { + message = (Message) this.serializer.deserialize(value); + } + catch (Exception e) { + throw new MessagingException("Deserialization of Message failed.", e); + } + } + else { + Object payload = value; + if (this.serializer != null) { + payload = this.serializer.deserialize(value); + } + message = MessageBuilder.withPayload(payload).build(); + } + } + + if (message != null) { + this.sendMessage(message); + } + } + + @Override + protected void doStart() { + if (!this.active) { + this.active = true; + this.restart(); + } + } + + private void restart() { + this.taskExecutor.execute(new ListenerTask()); + } + + @Override + protected void doStop() { + this.active = false; + } + + public boolean isListening() { + return listening; + } + + /** + * Returns the size of the Queue specified by {@link #boundListOperations}. The queue is + * represented by a Redis list. If the queue does not exist 0 + * is returned. See also http://redis.io/commands/llen + * + * @return Size of the queue. Never negative. + */ + @ManagedMetric + public long getQueueSize() { + return this.boundListOperations.size(); + } + + /** + * Clear the Redis Queue specified by {@link #boundListOperations}. + */ + @ManagedOperation + public void clearQueue() { + this.boundListOperations.getOperations().delete(this.boundListOperations.getKey()); + } + + + private class ListenerTask implements Runnable { + + @Override + public void run() { + RedisQueueMessageDrivenEndpoint.this.listening = true; + try { + while (RedisQueueMessageDrivenEndpoint.this.active) { + RedisQueueMessageDrivenEndpoint.this.popMessageAndSend(); + } + } + finally { + if (RedisQueueMessageDrivenEndpoint.this.active) { + RedisQueueMessageDrivenEndpoint.this.restart(); + } + else { + RedisQueueMessageDrivenEndpoint.this.listening = false; + } + } + } + + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java new file mode 100644 index 0000000000..bff2a74832 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java @@ -0,0 +1,112 @@ +/* + * 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.integration.redis.outbound; + +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.Message; +import org.springframework.integration.expression.IntegrationEvaluationContextAware; +import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + * @author Gunnar Hillert + * @author Artem Bilan + * @since 3.0 + */ +public class RedisQueueOutboundChannelAdapter extends AbstractMessageHandler implements IntegrationEvaluationContextAware { + + private final RedisSerializer stringSerializer = new StringRedisSerializer(); + + private final RedisTemplate template; + + private final Expression queueNameExpression; + + private EvaluationContext evaluationContext; + + private volatile boolean extractPayload = true; + + private volatile RedisSerializer serializer = new JdkSerializationRedisSerializer(); + + private volatile boolean serializerExplicitlySet; + + public RedisQueueOutboundChannelAdapter(String queueName, RedisConnectionFactory connectionFactory) { + this(new LiteralExpression(queueName), connectionFactory); + } + + public RedisQueueOutboundChannelAdapter(Expression queueNameExpression, RedisConnectionFactory connectionFactory) { + Assert.notNull(queueNameExpression, "'queueNameExpression' is required"); + Assert.hasText(queueNameExpression.getExpressionString(), "'queueNameExpression.getExpressionString()' is required"); + Assert.notNull(connectionFactory, "'connectionFactory' must not be null"); + this.queueNameExpression = queueNameExpression; + this.template = new RedisTemplate(); + this.template.setConnectionFactory(connectionFactory); + this.template.setEnableDefaultSerializer(false); + this.template.setKeySerializer(new StringRedisSerializer()); + this.template.afterPropertiesSet(); + } + + @Override + public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { + this.evaluationContext = evaluationContext; + } + + public void setExtractPayload(boolean extractPayload) { + this.extractPayload = extractPayload; + } + + public void setSerializer(RedisSerializer serializer) { + Assert.notNull(serializer, "'serializer' must not be null"); + this.serializer = serializer; + this.serializerExplicitlySet = true; + } + + @Override + public String getComponentType() { + return "int-redis:outbound-channel-adapter"; + } + + @Override + @SuppressWarnings("unchecked") + protected void handleMessageInternal(Message message) throws Exception { + Object value = message; + + if (this.extractPayload) { + value = message.getPayload(); + } + + if (!(value instanceof byte[])) { + if (value instanceof String && !this.serializerExplicitlySet) { + value = this.stringSerializer.serialize((String) value); + } + else { + value = ((RedisSerializer) this.serializer).serialize(value); + } + } + + String queueName = this.queueNameExpression.getValue(this.evaluationContext, message, String.class); + this.template.boundListOps(queueName).leftPush(value); + } + +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java new file mode 100644 index 0000000000..9184e0eff5 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java @@ -0,0 +1,157 @@ +/* + * 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.integration.redis.inbound; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; + +import java.util.Date; + +import org.hamcrest.Matchers; +import org.junit.Test; + +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.message.ErrorMessage; +import org.springframework.integration.redis.rules.RedisAvailable; +import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.support.MessageBuilder; + +/** + * @author Gunnar Hillert + * @author Artem Bilan + * @since 3.0 + */ +public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { + + @Test + @RedisAvailable + @SuppressWarnings("unchecked") + public void testInt3014Default() throws Exception { + + String queueName = "si.test.redisQueueInboundChannelAdapterTests"; + + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + + RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.setEnableDefaultSerializer(false); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + redisTemplate.afterPropertiesSet(); + + String payload = "testing"; + + redisTemplate.boundListOps(queueName).leftPush(payload); + + Date payload2 = new Date(); + + redisTemplate.boundListOps(queueName).leftPush(payload2); + + PollableChannel channel = new QueueChannel(); + + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, connectionFactory); + endpoint.setOutputChannel(channel); + endpoint.setReceiveTimeout(1000); + endpoint.afterPropertiesSet(); + endpoint.start(); + + Message receive = (Message) channel.receive(2000); + assertNotNull(receive); + assertEquals(payload, receive.getPayload()); + + receive = (Message) channel.receive(2000); + assertNotNull(receive); + assertEquals(payload2, receive.getPayload()); + + endpoint.stop(); + this.waitUntilListening(endpoint); + } + + @Test + @RedisAvailable + @SuppressWarnings("unchecked") + public void testInt3014ExpectMessageTrue() throws Exception { + + final String queueName = "si.test.redisQueueInboundChannelAdapterTests2"; + + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + + RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.setEnableDefaultSerializer(false); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + redisTemplate.afterPropertiesSet(); + + Message message = MessageBuilder.withPayload("testing").build(); + + redisTemplate.boundListOps(queueName).leftPush(message); + + redisTemplate.boundListOps(queueName).leftPush("test"); + + PollableChannel channel = new QueueChannel(); + + PollableChannel errorChannel = new QueueChannel(); + + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, connectionFactory); + endpoint.setExpectMessage(true); + endpoint.setOutputChannel(channel); + endpoint.setErrorChannel(errorChannel); + endpoint.setReceiveTimeout(1000); + endpoint.afterPropertiesSet(); + endpoint.start(); + + Message receive = (Message) channel.receive(2000); + assertNotNull(receive); + + assertEquals(message, receive); + + receive = (Message) errorChannel.receive(2000); + assertNotNull(receive); + assertThat(receive, Matchers.instanceOf(ErrorMessage.class)); + assertThat(receive.getPayload(), Matchers.instanceOf(MessagingException.class)); + assertThat(((Exception) receive.getPayload()).getMessage(), Matchers.containsString("Deserialization of Message failed.")); + assertThat(((Exception) receive.getPayload()).getCause(), Matchers.instanceOf(ClassCastException.class)); + assertThat(((Exception) receive.getPayload()).getCause().getMessage(), + Matchers.containsString("java.lang.String cannot be cast to org.springframework.integration.Message")); + + + endpoint.stop(); + this.waitUntilListening(endpoint); + } + + + public void waitUntilListening(RedisQueueMessageDrivenEndpoint endpoint) throws Exception { + int n = 0; + while (endpoint.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + throw new Exception("RedisQueueMessageDrivenEndpoint failed to stop."); + } + } + + } + +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java new file mode 100644 index 0000000000..47a0c8c35c --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java @@ -0,0 +1,143 @@ +/* + * 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.integration.redis.outbound; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.Arrays; +import java.util.Date; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; + +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.JacksonJsonRedisSerializer; +import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.Message; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.redis.rules.RedisAvailable; +import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.support.MessageBuilder; + +/** + * @author Gunnar Hillert + * @author Artem Bilan + * @since 3.0 + */ +public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests { + + @Test + @RedisAvailable + public void testInt3015Default() throws Exception { + + final String queueName = "si.test.testRedisQueueOutboundChannelAdapter"; + + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + + final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory); + + String payload = "testing"; + handler.handleMessage(MessageBuilder.withPayload(payload).build()); + + RedisTemplate redisTemplate = new StringRedisTemplate(); + redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.afterPropertiesSet(); + + Object result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); + assertNotNull(result); + + assertEquals(payload, result); + + Date payload2 = new Date(); + handler.handleMessage(MessageBuilder.withPayload(payload2).build()); + + RedisTemplate redisTemplate2 = new RedisTemplate(); + redisTemplate2.setConnectionFactory(connectionFactory); + redisTemplate2.setEnableDefaultSerializer(false); + redisTemplate2.setKeySerializer(new StringRedisSerializer()); + redisTemplate2.setValueSerializer(new JdkSerializationRedisSerializer()); + redisTemplate2.afterPropertiesSet(); + + Object result2 = redisTemplate2.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); + assertNotNull(result2); + + assertEquals(payload2, result2); + } + + @Test + @RedisAvailable + public void testInt3015ExtractPayloadFalse() throws Exception { + + final String queueName = "si.test.testRedisQueueOutboundChannelAdapter2"; + + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + + final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory); + handler.setExtractPayload(false); + + Message message = MessageBuilder.withPayload("testing").build(); + handler.handleMessage(message); + + RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.setEnableDefaultSerializer(false); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + redisTemplate.afterPropertiesSet(); + + Object result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); + assertNotNull(result); + + assertEquals(message, result); + + } + + @Test + @RedisAvailable + public void testInt3015ExplicitSerializer() throws Exception { + + final String queueName = "si.test.testRedisQueueOutboundChannelAdapter2"; + + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + + final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory); + handler.setSerializer(new JacksonJsonRedisSerializer(Object.class)); + + RedisTemplate redisTemplate = new StringRedisTemplate(); + redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.afterPropertiesSet(); + + handler.handleMessage(new GenericMessage(Arrays.asList("foo", "bar", "baz"))); + + Object result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); + assertNotNull(result); + + assertEquals("[\"foo\",\"bar\",\"baz\"]", result); + + handler.handleMessage(new GenericMessage("test")); + + result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); + assertNotNull(result); + + assertEquals("\"test\"", result); + } + +} diff --git a/spring-integration-redis/src/test/resources/log4j.properties b/spring-integration-redis/src/test/resources/log4j.properties new file mode 100644 index 0000000000..7ebfb6af9b --- /dev/null +++ b/spring-integration-redis/src/test/resources/log4j.properties @@ -0,0 +1,8 @@ +log4j.rootCategory=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - <%m>%n + +log4j.category.org.springframework.integration=WARN +log4j.category.org.springframework.integration.redis=DEBUG