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
This commit is contained in:
Artem Bilan
committed by
Gary Russell
Gary Russell
parent
9849e2087d
commit
213cafb2da
@@ -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);
|
||||
|
||||
}
|
||||
@@ -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 extends AbstractMessageChannel> T lookupOrCreateSharedChannel(String name, Class<T> 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 extends AbstractMessageChannel> T createSharedChannel(String name, Class<T> 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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,4 +0,0 @@
|
||||
/**
|
||||
* Provides classes representing channel registries.
|
||||
*/
|
||||
package org.springframework.integration.channel.registry;
|
||||
@@ -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<String>("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<String>("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<String>("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<String>("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<String>("hello"));
|
||||
assertTrue(originalMessageReceived.get());
|
||||
assertTrue(tapMessageReceived.get());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String, byte[]> 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<String, byte[]> template = new RedisTemplate<String, byte[]>();
|
||||
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}.
|
||||
* <p/>
|
||||
* 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.
|
||||
* <p/>
|
||||
* A timeout of zero can be used to block indefinitely. If not set explicitly
|
||||
* the timeout value will default to {@code 1000}
|
||||
* <p/>
|
||||
* 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<Object> message = null;
|
||||
|
||||
byte[] value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
|
||||
|
||||
if (value != null) {
|
||||
if (this.expectMessage) {
|
||||
try {
|
||||
message = (Message<Object>) 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 <code>0</code>
|
||||
* 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String> stringSerializer = new StringRedisSerializer();
|
||||
|
||||
private final RedisTemplate<String, Object> 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<String, Object>();
|
||||
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<Object>) this.serializer).serialize(value);
|
||||
}
|
||||
}
|
||||
|
||||
String queueName = this.queueNameExpression.getValue(this.evaluationContext, message, String.class);
|
||||
this.template.boundListOps(queueName).leftPush(value);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String, Object> redisTemplate = new RedisTemplate<String, Object>();
|
||||
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<Object> receive = (Message<Object>) channel.receive(2000);
|
||||
assertNotNull(receive);
|
||||
assertEquals(payload, receive.getPayload());
|
||||
|
||||
receive = (Message<Object>) 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<String, Object> redisTemplate = new RedisTemplate<String, Object>();
|
||||
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<Object> receive = (Message<Object>) channel.receive(2000);
|
||||
assertNotNull(receive);
|
||||
|
||||
assertEquals(message, receive);
|
||||
|
||||
receive = (Message<Object>) 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.");
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String, ?> 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<String, ?> redisTemplate2 = new RedisTemplate<String, Object>();
|
||||
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<String> message = MessageBuilder.withPayload("testing").build();
|
||||
handler.handleMessage(message);
|
||||
|
||||
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
|
||||
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>(Object.class));
|
||||
|
||||
RedisTemplate<String, ?> redisTemplate = new StringRedisTemplate();
|
||||
redisTemplate.setConnectionFactory(connectionFactory);
|
||||
redisTemplate.afterPropertiesSet();
|
||||
|
||||
handler.handleMessage(new GenericMessage<Object>(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<Object>("test"));
|
||||
|
||||
result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS);
|
||||
assertNotNull(result);
|
||||
|
||||
assertEquals("\"test\"", result);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user