diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultMessage.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultMessage.java index 84d0a7d8d..901aaf302 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultMessage.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultMessage.java @@ -25,6 +25,7 @@ public class DefaultMessage implements Message { private final byte[] payload; private final byte[] channel; + private String toString; public DefaultMessage(byte[] payload, byte[] channel) { this.payload = payload; @@ -40,4 +41,12 @@ public class DefaultMessage implements Message { public byte[] getPayload() { return (payload != null ? payload.clone() : null); } + + @Override + public String toString() { + if (toString == null){ + toString = new String(payload); + } + return toString; + } } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisOperations.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisOperations.java index ac1a9349d..06ff3d420 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisOperations.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisOperations.java @@ -81,11 +81,15 @@ public interface RedisOperations { void discard(); Object exec(); - + List sort(K key, SortParameters params); Long sort(K key, SortParameters params, K destination); + // pubsub functionality on the template + void convertAndSend(String destination, Object message); + + // operation types /** * Returns the operations performed on simple values (or Strings in Redis terminology). diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java index ac88b1bca..7fecbf3f1 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java @@ -269,7 +269,7 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation } @SuppressWarnings("unchecked") - private byte[] rawValue(T value) { + private byte[] rawValue(Object value) { return (value != null ? valueSerializer.serialize(value) : null); } @@ -509,6 +509,22 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation }, true); } + @Override + public void convertAndSend(String channel, Object message) { + Assert.hasText(channel, "a non-empty channel is required"); + + final byte[] rawChannel = rawString(channel); + final byte[] rawMessage = rawValue(message); + + execute(new RedisCallback() { + @Override + public Object doInRedis(RedisConnection connection) { + connection.publish(rawMessage, rawChannel); + return null; + } + }, true); + } + // // Value operations @@ -889,7 +905,6 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation - // // List operations // @@ -1736,7 +1751,7 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation return connection.hGetAll(rawKey); } }, true); - + return deserializeHashMap(entries); } } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/RedisListeningContainer.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/RedisListenerContainer.java similarity index 95% rename from spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/RedisListeningContainer.java rename to spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/RedisListenerContainer.java index 41205a98a..7c3071370 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/RedisListeningContainer.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/RedisListenerContainer.java @@ -56,15 +56,14 @@ import org.springframework.util.CollectionUtils; * * @author Costin Leau */ -public class RedisListeningContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle { +public class RedisListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle { - private static final Log log = LogFactory.getLog(RedisListeningContainer.class); + private static final Log log = LogFactory.getLog(RedisListenerContainer.class); /** * Default thread name prefix: "RedisListeningContainer-". */ - public static final String DEFAULT_THREAD_NAME_PREFIX = ClassUtils.getShortName(RedisListeningContainer.class) - + "-"; + public static final String DEFAULT_THREAD_NAME_PREFIX = ClassUtils.getShortName(RedisListenerContainer.class) + "-"; private Executor subscriptionExecutor; @@ -137,6 +136,10 @@ public class RedisListeningContainer implements InitializingBean, DisposableBean if (manageExecutor) { if (taskExecutor instanceof DisposableBean) { ((DisposableBean) taskExecutor).destroy(); + + if (log.isDebugEnabled()) { + log.debug("Stopped internally-managed task executor"); + } } } } @@ -168,6 +171,9 @@ public class RedisListeningContainer implements InitializingBean, DisposableBean if (!running) { running = true; lazyListen(); + if (log.isDebugEnabled()) { + log.debug("Started RedisListenerContainer"); + } } } @@ -175,6 +181,10 @@ public class RedisListeningContainer implements InitializingBean, DisposableBean public void stop() { running = false; subscriptionTask.cancel(); + + if (log.isDebugEnabled()) { + log.debug("Stopped RedisListenerContainer"); + } } /** @@ -318,6 +328,8 @@ public class RedisListeningContainer implements InitializingBean, DisposableBean List channels = new ArrayList(topics.size()); List patterns = new ArrayList(topics.size()); + boolean trace = log.isTraceEnabled(); + for (Topic topic : topics) { ArrayHolder holder = new ArrayHolder(serializer.serialize(topic.getTopic())); @@ -330,6 +342,9 @@ public class RedisListeningContainer implements InitializingBean, DisposableBean } collection.add(listener); channels.add(holder.array); + + if (trace) + log.trace("Adding listener '" + listener + "' on channel '" + topic.getTopic() + "'"); } else if (topic instanceof PatternTopic) { @@ -340,6 +355,9 @@ public class RedisListeningContainer implements InitializingBean, DisposableBean } collection.add(listener); patterns.add(holder.array); + + if (trace) + log.trace("Adding listener '" + listener + "' for pattern '" + topic.getTopic() + "'"); } else { @@ -430,7 +448,9 @@ public class RedisListeningContainer implements InitializingBean, DisposableBean // and schedule the rest if (!channelMapping.isEmpty()) { // schedule the rest of the subscription - subscriptionExecutor.execute(new PatternSubscriptionTask()); + if (!patternMapping.isEmpty()) { + subscriptionExecutor.execute(new PatternSubscriptionTask()); + } connection.subscribe(new DispatchMessageListener(), unwrap(channelMapping.keySet())); } else { diff --git a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTestParams.java b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTestParams.java new file mode 100644 index 000000000..750487509 --- /dev/null +++ b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTestParams.java @@ -0,0 +1,52 @@ +/* + * Copyright 2011 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.data.keyvalue.redis.listener; + +import java.util.Arrays; +import java.util.Collection; + +import org.springframework.data.keyvalue.redis.Person; +import org.springframework.data.keyvalue.redis.SettingsUtils; +import org.springframework.data.keyvalue.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.data.keyvalue.redis.core.RedisTemplate; +import org.springframework.data.keyvalue.redis.core.StringRedisTemplate; +import org.springframework.data.keyvalue.redis.support.collections.ObjectFactory; +import org.springframework.data.keyvalue.redis.support.collections.PersonObjectFactory; +import org.springframework.data.keyvalue.redis.support.collections.StringObjectFactory; + +/** + * @author Costin Leau + */ +public class PubSubTestParams { + + public static Collection testParams() { + // create Jedis Factory + ObjectFactory stringFactory = new StringObjectFactory(); + ObjectFactory personFactory = new PersonObjectFactory(); + + JedisConnectionFactory jedisConnFactory = new JedisConnectionFactory(); + jedisConnFactory.setUsePool(true); + jedisConnFactory.setPort(SettingsUtils.getPort()); + jedisConnFactory.setHostName(SettingsUtils.getHost()); + + jedisConnFactory.afterPropertiesSet(); + + RedisTemplate stringTemplate = new StringRedisTemplate(jedisConnFactory); + RedisTemplate personTemplate = new RedisTemplate(jedisConnFactory); + + return Arrays.asList(new Object[][] { { stringFactory, stringTemplate }, { personFactory, personTemplate } }); + } +} diff --git a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTests.java b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTests.java new file mode 100644 index 000000000..0f359d326 --- /dev/null +++ b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTests.java @@ -0,0 +1,122 @@ +/* + * Copyright 2011 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.data.keyvalue.redis.listener; + +import java.util.Arrays; +import java.util.Collection; +import java.util.LinkedHashSet; +import java.util.Set; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.data.keyvalue.redis.connection.Message; +import org.springframework.data.keyvalue.redis.connection.MessageListener; +import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory; +import org.springframework.data.keyvalue.redis.core.RedisTemplate; +import org.springframework.data.keyvalue.redis.support.collections.ObjectFactory; + +/** + * Base test class for PubSub integration tests + * + * @author Costin Leau + */ +@RunWith(Parameterized.class) +public class PubSubTests { + + private static final String CHANNEL = "pubsub::test"; + + protected RedisListenerContainer container; + protected ObjectFactory factory; + protected RedisTemplate template; + private static Set connFactories = new LinkedHashSet(); + + private MessageListener testListener; + + @Before + public void setUp() throws Exception { + container = new RedisListenerContainer(); + container.setConnectionFactory(template.getConnectionFactory()); + container.setBeanName("container"); + container.afterPropertiesSet(); + } + + @After + public void tearDown() throws Exception { + container.destroy(); + } + + public PubSubTests(ObjectFactory factory, RedisTemplate template) { + this.factory = factory; + this.template = template; + connFactories.add(template.getConnectionFactory()); + } + + @AfterClass + public static void cleanUp() { + if (connFactories != null) { + for (RedisConnectionFactory connectionFactory : connFactories) { + try { + ((DisposableBean) connectionFactory).destroy(); + System.out.println("Succesfully cleaned up factory " + connectionFactory); + } catch (Exception ex) { + System.err.println("Cannot clean factory " + connectionFactory + ex); + } + } + } + } + + @Parameters + public static Collection testParams() { + return PubSubTestParams.testParams(); + } + + /** + * Return a new instance of T + * @return + */ + protected T getT() { + return factory.instance(); + } + + @Test + public void testContainerSubscribe() throws Exception { + final BlockingQueue bag = new ArrayBlockingQueue(4); + + container.addMessageListener(new MessageListener() { + + @Override + public void onMessage(Message message, byte[] pattern) { + System.out.println("Received message " + message + " and pattern=" + pattern); + bag.add(message); + } + }, Arrays.asList(new ChannelTopic(CHANNEL))); + + Thread.sleep(500); + template.convertAndSend(CHANNEL, "bar"); + template.convertAndSend(CHANNEL, "bar1"); + System.out.println("Found in bag " + bag.poll(1, TimeUnit.SECONDS)); + System.out.println("Found in bag " + bag.poll(1, TimeUnit.SECONDS)); + } +} \ No newline at end of file diff --git a/spring-data-redis/src/test/resources/log4j.properties b/spring-data-redis/src/test/resources/log4j.properties index 6d5422d74..945449482 100644 --- a/spring-data-redis/src/test/resources/log4j.properties +++ b/spring-data-redis/src/test/resources/log4j.properties @@ -4,10 +4,7 @@ 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.apache.activemq=ERROR -log4j.category.org.springframework.batch=DEBUG -log4j.category.org.springframework.transaction=INFO +log4j.category.org.springframework.data.keyvalue.redis.listener=TRACE -log4j.category.org.hibernate.SQL=DEBUG # for debugging datasource initialization # log4j.category.test.jdbc=DEBUG diff --git a/spring-data-redis/template.mf b/spring-data-redis/template.mf index f3c3bd9eb..a37ef7cbb 100644 --- a/spring-data-redis/template.mf +++ b/spring-data-redis/template.mf @@ -6,8 +6,10 @@ Import-Package: sun.reflect;version="0";resolution:=optional Import-Template: org.springframework.beans.*;version="[3.0.0, 4.0.0)", + org.springframework.context.*;version="[3.0.0, 4.0.0)", org.springframework.core.*;version="[3.0.0, 4.0.0)", org.springframework.dao.*;version="[3.0.0, 4.0.0)", + org.springframework.scheduling.*;version="[3.0.0, 4.0.0)", org.springframework.util.*;version="[3.0.0, 4.0.0)", org.springframework.data.core.*;version="[1.0.0, 2.0.0)", org.springframework.data.*;version="[1.0.0, 2.0.0)",