DATAKV-40

+ add option to select db on each connection factory

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java

@@ -72,6 +72,7 @@ public class JedisConnection implements RedisConnection {
 
 	private volatile JedisSubscription subscription;
 	private volatile Pipeline pipeline;
+	private final int dbIndex;
 
 	/**
 	 * Constructs a new <code>JedisConnection</code> instance.
@@ -79,7 +80,7 @@ public class JedisConnection implements RedisConnection {
 	 * @param jedis Jedis entity
 	 */
 	public JedisConnection(Jedis jedis) {
-		this(jedis, null);
+		this(jedis, null, 0);
 	}
 
 	/**
@@ -89,13 +90,20 @@ public class JedisConnection implements RedisConnection {
 	 * @param jedis
 	 * @param pool can be null, if no pool is used
 	 */
-	public JedisConnection(Jedis jedis, Pool<Jedis> pool) {
+	public JedisConnection(Jedis jedis, Pool<Jedis> pool, int dbIndex) {
 		this.jedis = jedis;
 		// extract underlying connection for batch operations
 		client = (Client) ReflectionUtils.getField(CLIENT_FIELD, jedis);
 		transaction = new Transaction(client);
 
 		this.pool = pool;
+
+		this.dbIndex = dbIndex;
+
+		// select the db
+		if (dbIndex > 0) {
+			select(dbIndex);
+		}
 	}
 
 	protected DataAccessException convertJedisAccessException(Exception ex) {

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnectionFactory.java

@@ -24,6 +24,7 @@ import org.springframework.dao.DataAccessException;
 import org.springframework.dao.DataAccessResourceFailureException;
 import org.springframework.data.keyvalue.redis.connection.RedisConnection;
 import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory;
+import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 import redis.clients.jedis.Jedis;
@@ -51,6 +52,8 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 	private JedisPool pool = null;
 	private JedisPoolConfig poolConfig = new JedisPoolConfig();
 
+	private int dbIndex = 0;
+
 	/**
 	 * Constructs a new <code>JedisConnectionFactory</code> instance
 	 * with default settings (default connection pooling, no shard information).
@@ -99,6 +102,18 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 		}
 	}
 
+	/**
+	 * Post process a newly retrieved connection. Useful for decorating or executing
+	 * initialization commands on a new connection.
+	 * This implementation simply returns the connection.
+	 * 
+	 * @param connection
+	 * @return processed connection
+	 */
+	protected JedisConnection postProcessConnection(JedisConnection connection) {
+		return connection;
+	}
+
 	public void afterPropertiesSet() {
 		if (shardInfo == null) {
 			shardInfo = new JedisShardInfo(hostName, port);
@@ -113,8 +128,8 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 		}
 
 		if (usePool) {
-			pool = new JedisPool(poolConfig, shardInfo.getHost(), shardInfo.getPort(),
-					shardInfo.getTimeout(), shardInfo.getPassword());
+			pool = new JedisPool(poolConfig, shardInfo.getHost(), shardInfo.getPort(), shardInfo.getTimeout(),
+					shardInfo.getPassword());
 		}
 	}
 
@@ -131,7 +146,8 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 
 	public JedisConnection getConnection() {
 		Jedis jedis = fetchJedisConnector();
-		return (usePool ? new JedisConnection(jedis, pool) : new JedisConnection(jedis));
+		return postProcessConnection((usePool ? new JedisConnection(jedis, pool, dbIndex) : new JedisConnection(jedis,
+				null, dbIndex)));
 	}
 
 	@Override
@@ -263,4 +279,15 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 	public void setPoolConfig(JedisPoolConfig poolConfig) {
 		this.poolConfig = poolConfig;
 	}
+
+	/**
+	 * Sets the index of the database used by this connection factory.
+	 * Can be between 0 (default) and 15.
+	 * 
+	 * @param index database index
+	 */
+	public void setDatabase(int index) {
+		Assert.isTrue(index >= 0 && index < 16, "invalid DB index (needs to be between 0 and 15)");
+		this.dbIndex = index;
+	}
 }

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnectionFactory.java

@@ -45,6 +45,7 @@ public class JredisConnectionFactory implements InitializingBean, DisposableBean
 	private int timeout;
 
 	private boolean usePool = true;
+	private int dbIndex = DEFAULT_REDIS_DB;
 
 	private JRedisService pool = null;
 	// taken from JRedis code
@@ -75,7 +76,7 @@ public class JredisConnectionFactory implements InitializingBean, DisposableBean
 	public void afterPropertiesSet() {
 		if (connectionSpec == null) {
 			Assert.hasText(hostName);
-			connectionSpec = DefaultConnectionSpec.newSpec(hostName, port, DEFAULT_REDIS_DB, DEFAULT_REDIS_PASSWORD);
+			connectionSpec = DefaultConnectionSpec.newSpec(hostName, port, dbIndex, DEFAULT_REDIS_PASSWORD);
 			connectionSpec.setConnectionFlag(Connection.Flag.RELIABLE, false);
 
 			if (StringUtils.hasLength(password)) {
@@ -105,10 +106,22 @@ public class JredisConnectionFactory implements InitializingBean, DisposableBean
 
 	@Override
 	public RedisConnection getConnection() {
-		return new JredisConnection((usePool ? pool : new JRedisClient(connectionSpec)));
+		return postProcessConnection(new JredisConnection((usePool ? pool : new JRedisClient(connectionSpec))));
 	}
 
 
+	/**
+	 * Post process a newly retrieved connection. Useful for decorating or executing
+	 * initialization commands on a new connection.
+	 * This implementation simply returns the connection.
+	 * 
+	 * @param connection
+	 * @return processed connection
+	 */
+	protected RedisConnection postProcessConnection(JredisConnection connection) {
+		return connection;
+	}
+
 	@Override
 	public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
 		if (ex instanceof ClientRuntimeException) {
@@ -210,4 +223,15 @@ public class JredisConnectionFactory implements InitializingBean, DisposableBean
 		this.poolSize = poolSize;
 		usePool = true;
 	}
+
+	/**
+	 * Sets the index of the database used by this connection factory.
+	 * Can be between 0 (default) and 15.
+	 * 
+	 * @param index database index
+	 */
+	public void setDatabase(int index) {
+		Assert.isTrue(index >= 0 && index < 16, "invalid DB index (needs to be between 0 and 15)");
+		this.dbIndex = index;
+	}
 }

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

@@ -38,9 +38,10 @@ public class PubSubTestParams {
 		ObjectFactory<Person> personFactory = new PersonObjectFactory();
 
 		JedisConnectionFactory jedisConnFactory = new JedisConnectionFactory();
-		jedisConnFactory.setUsePool(false);
+		jedisConnFactory.setUsePool(true);
 		jedisConnFactory.setPort(SettingsUtils.getPort());
 		jedisConnFactory.setHostName(SettingsUtils.getHost());
+		jedisConnFactory.setDatabase(2);
 
 		jedisConnFactory.afterPropertiesSet();
 

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

@@ -57,7 +57,6 @@ public class PubSubTests<T> {
 
 	private final Object handler = new Object() {
 		void handleMessage(String message) {
-			System.out.println("Received message " + message);
 			bag.add(message);
 		}
 	};