DATAKV-46

+ fix some other minor bugs (pubsub support still doesn't work for RJC)

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java

@@ -57,8 +57,8 @@ public class RjcConnection implements RedisConnection {
 	public RjcConnection(org.idevlab.rjc.ds.RedisConnection connection, int dbIndex) {
 		SingleDataSource connectionDataSource = new SingleDataSource(connection);
 		session = new SessionFactoryImpl(connectionDataSource).create();
-		client = new Client(connection);
 		subscriber = new RedisNodeSubscriber(connectionDataSource);
+		client = new Client(connection);
 
 		this.dbIndex = dbIndex;
 
@@ -1731,7 +1731,7 @@ public class RjcConnection implements RedisConnection {
 				pipeline.zscore(stringKey, stringValue);
 				return null;
 			}
-			return Double.valueOf(session.zscore(stringKey, stringValue));
+			return RjcUtils.convert(session.zscore(stringKey, stringValue));
 		} catch (Exception ex) {
 			throw convertRjcAccessException(ex);
 		}
@@ -2031,7 +2031,7 @@ public class RjcConnection implements RedisConnection {
 			}
 
 			subscription = new RjcSubscription(listener, subscriber);
-			subscription.pSubscribe(channels);
+			subscription.subscribe(channels);
 
 		} catch (Exception ex) {
 			throw convertRjcAccessException(ex);

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java

@@ -216,4 +216,8 @@ public abstract class RjcUtils {
 		}
 		return result;
 	}
+
+	static Double convert(String zscore) {
+		return (zscore == null ? null : Double.valueOf(zscore));
+	}
 }

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/jredis/JRedisConnectionIntegrationTests.java

@@ -54,4 +54,24 @@ public class JRedisConnectionIntegrationTests extends AbstractConnectionIntegrat
 	@Ignore("JRedis does not support pipelining")
 	public void testNullCollections() {
 	}
-}
+
+	@Ignore
+	public void testNullKey() throws Exception {
+	}
+
+	@Ignore
+	public void testNullValue() throws Exception {
+	}
+
+	@Ignore
+	public void testHashNullKey() throws Exception {
+	}
+
+	@Ignore
+	public void testHashNullValue() throws Exception {
+	}
+
+	@Ignore
+	public void testNullSerialization() throws Exception {
+	}
+}

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTestParams.java

@@ -21,6 +21,7 @@ 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.connection.rjc.RjcConnectionFactory;
 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;
@@ -48,7 +49,20 @@ public class PubSubTestParams {
 		RedisTemplate<String, String> stringTemplate = new StringRedisTemplate(jedisConnFactory);
 		RedisTemplate<String, Person> personTemplate = new RedisTemplate<String, Person>(jedisConnFactory);
 
+		// create RJC
 
-		return Arrays.asList(new Object[][] { { stringFactory, stringTemplate }, { personFactory, personTemplate } });
+		RjcConnectionFactory rjcConnFactory = new RjcConnectionFactory();
+		rjcConnFactory.setUsePool(false);
+		rjcConnFactory.setPort(SettingsUtils.getPort());
+		rjcConnFactory.setHostName(SettingsUtils.getHost());
+		rjcConnFactory.afterPropertiesSet();
+
+		RedisTemplate<String, String> stringTemplateRJC = new StringRedisTemplate(rjcConnFactory);
+		RedisTemplate<String, Person> personTemplateRJC = new RedisTemplate<String, Person>(rjcConnFactory);
+
+
+		return Arrays.asList(new Object[][] { { stringFactory, stringTemplate }, { personFactory, personTemplate }
+		//,{ stringFactory, stringTemplateRJC }, { personFactory, personTemplateRJC } 
+		});
 	}
 }

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTests.java

@@ -32,8 +32,7 @@ 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.RedisConnectionFactory;
+import org.springframework.data.keyvalue.redis.ConnectionFactoryTracker;
 import org.springframework.data.keyvalue.redis.core.RedisTemplate;
 import org.springframework.data.keyvalue.redis.listener.adapter.MessageListenerAdapter;
 import org.springframework.data.keyvalue.redis.support.collections.ObjectFactory;
@@ -51,7 +50,6 @@ public class PubSubTests<T> {
 	protected RedisMessageListenerContainer container;
 	protected ObjectFactory<T> factory;
 	protected RedisTemplate template;
-	private static Set<RedisConnectionFactory> connFactories = new LinkedHashSet<RedisConnectionFactory>();
 
 	private final BlockingDeque<String> bag = new LinkedBlockingDeque<String>(99);
 
@@ -84,21 +82,12 @@ public class PubSubTests<T> {
 	public PubSubTests(ObjectFactory<T> factory, RedisTemplate template) {
 		this.factory = factory;
 		this.template = template;
-		connFactories.add(template.getConnectionFactory());
+		ConnectionFactoryTracker.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);
-				}
-			}
-		}
+		ConnectionFactoryTracker.cleanUp();
 	}
 
 	@Parameters
@@ -126,6 +115,8 @@ public class PubSubTests<T> {
 		set.add(bag.poll(1, TimeUnit.SECONDS));
 		set.add(bag.poll(1, TimeUnit.SECONDS));
 
+		System.out.println(set);
+
 		assertTrue(set.contains(payload1));
 		assertTrue(set.contains(payload2));
 	}