DATAKV-22

+ add initial integration test
+ add convertAndSend to RedisTemplate (following the JMS naming patterns)
+ update OSGi manifest template
+ update log4j config files

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;
+	}
 }

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisOperations.java

@@ -81,11 +81,15 @@ public interface RedisOperations<K, V> {
 	void discard();
 
 	Object exec();
-
+	
 	List<V> 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).

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java

@@ -269,7 +269,7 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 	}
 
 	@SuppressWarnings("unchecked")
-	private <T> byte[] rawValue(T value) {
+	private byte[] rawValue(Object value) {
 		return (value != null ? valueSerializer.serialize(value) : null);
 	}
 
@@ -509,6 +509,22 @@ public class RedisTemplate<K, V> 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<Object>() {
+			@Override
+			public Object doInRedis(RedisConnection connection) {
+				connection.publish(rawMessage, rawChannel);
+				return null;
+			}
+		}, true);
+	}
+
 
 	//
 	// Value operations
@@ -889,7 +905,6 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 
 
 
-
 	//
 	// List operations
 	//
@@ -1736,7 +1751,7 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 					return connection.hGetAll(rawKey);
 				}
 			}, true);
-			
+
 			return deserializeHashMap(entries);
 		}
 	}

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<byte[]> channels = new ArrayList<byte[]>(topics.size());
 		List<byte[]> patterns = new ArrayList<byte[]>(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 {

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<Object[]> testParams() {
+		// create Jedis Factory
+		ObjectFactory<String> stringFactory = new StringObjectFactory();
+		ObjectFactory<Person> personFactory = new PersonObjectFactory();
+
+		JedisConnectionFactory jedisConnFactory = new JedisConnectionFactory();
+		jedisConnFactory.setUsePool(true);
+		jedisConnFactory.setPort(SettingsUtils.getPort());
+		jedisConnFactory.setHostName(SettingsUtils.getHost());
+
+		jedisConnFactory.afterPropertiesSet();
+
+		RedisTemplate<String, String> stringTemplate = new StringRedisTemplate(jedisConnFactory);
+		RedisTemplate<String, Person> personTemplate = new RedisTemplate<String, Person>(jedisConnFactory);
+
+		return Arrays.asList(new Object[][] { { stringFactory, stringTemplate }, { personFactory, personTemplate } });
+	}
+}

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<T> {
+
+	private static final String CHANNEL = "pubsub::test";
+
+	protected RedisListenerContainer container;
+	protected ObjectFactory<T> factory;
+	protected RedisTemplate template;
+	private static Set<RedisConnectionFactory> connFactories = new LinkedHashSet<RedisConnectionFactory>();
+
+	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<T> 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<Object[]> 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<Message> bag = new ArrayBlockingQueue<Message>(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));
+	}
+}

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

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)",