Merge remote-tracking branch 'upstream/master' into 4.0.0-WIP
Conflicts: spring-integration-core/src/main/java/org/springframework/integration/channel/registry/ChannelRegistry.java spring-integration-core/src/main/java/org/springframework/integration/channel/registry/LocalChannelRegistry.java spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonMessageParser.java spring-integration-core/src/test/java/org/springframework/integration/channel/registry/LocalChannelRegistryTests.java spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests.java spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayTests.java Resolved.
This commit is contained in:
Gary Russell
@@ -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