DATAREDIS-431 - Autoselect database configured in LettuceConnectionFactory.

We now actively select the predefined db set via the dbIndex on LettuceConnectionFactory. Fixed some spelling issues along the way.

Original pull request: #157.

Author: christophstrobl

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java

@@ -127,6 +127,9 @@ public class LettuceConnection extends AbstractRedisConnection {
 			LettuceConverters.exceptionConverter());
 	private static final TypeHints typeHints = new TypeHints();
 
+	private final int defaultDbIndex;
+	private int dbIndex;
+
 	static {
 		SYNC_HANDLER = ReflectionUtils.findMethod(AbstractRedisClient.class, "syncHandler", RedisChannelHandler.class,
 				Class[].class);
@@ -293,11 +296,29 @@ public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection<byte[], byte
 	 */
 	public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> sharedConnection, long timeout,
 			RedisClient client, LettucePool pool) {
+
+		this(sharedConnection, timeout, client, pool, 0);
+	}
+
+	/**
+	 * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be
+	 *          used for transactions or blocking operations
+	 * @param timeout The connection timeout (in milliseconds)
+	 * @param client The {@link RedisClient} to use when making pub/sub connections.
+	 * @param pool The connection pool to use for blocking and tx operations.
+	 * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection.
+	 * @since 1.7
+	 */
+	public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> sharedConnection, long timeout,
+			RedisClient client, LettucePool pool, int defaultDbIndex) {
+
 		this.asyncSharedConn = sharedConnection;
 		this.timeout = timeout;
 		this.client = client;
 		this.sharedConn = sharedConnection != null ? syncConnection(asyncSharedConn) : null;
 		this.pool = pool;
+		this.defaultDbIndex = defaultDbIndex;
+		this.dbIndex = this.defaultDbIndex;
 	}
 
 	protected DataAccessException convertLettuceAccessException(Exception ex) {
@@ -404,6 +425,8 @@ public void close() throws DataAccessException {
 			}
 			subscription = null;
 		}
+
+		this.dbIndex = defaultDbIndex;
 	}
 
 	public boolean isClosed() {
@@ -1075,6 +1098,7 @@ public Boolean renameNX(byte[] oldName, byte[] newName) {
 	}
 
 	public void select(int dbIndex) {
+
 		if (asyncSharedConn != null) {
 			throw new UnsupportedOperationException("Selecting a new database not supported due to shared connection. "
 					+ "Use separate ConnectionFactorys to work with multiple databases");
@@ -1083,6 +1107,8 @@ public void select(int dbIndex) {
 			throw new UnsupportedOperationException("Lettuce blocks for #select");
 		}
 		try {
+
+			this.dbIndex = dbIndex;
 			if (isQueueing()) {
 				transaction(new LettuceTxStatusResult(getConnection().select(dbIndex)));
 				return;
@@ -3412,7 +3438,9 @@ private RedisAsyncConnection<byte[], byte[]> getAsyncDedicatedConnection() {
 			if (this.pool != null) {
 				this.asyncDedicatedConn = pool.getResource();
 			} else {
+
 				this.asyncDedicatedConn = client.connectAsync(CODEC);
+				this.asyncDedicatedConn.select(dbIndex);
 			}
 		}
 		return asyncDedicatedConn;

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java

@@ -118,8 +118,13 @@ public void destroy() {
 		client.shutdown(shutdownTimeout, shutdownTimeout, TimeUnit.MILLISECONDS);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisConnectionFactory#getConnection()
+	 */
 	public RedisConnection getConnection() {
-		LettuceConnection connection = new LettuceConnection(getSharedConnection(), timeout, client, pool);
+
+		LettuceConnection connection = new LettuceConnection(getSharedConnection(), timeout, client, pool, dbIndex);
 		connection.setConvertPipelineAndTxResults(convertPipelineAndTxResults);
 		return connection;
 	}

src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java

@@ -111,7 +111,9 @@ public void setUp() {
 	@After
 	public void tearDown() {
 		try {
-			connection.flushDb();
+
+			// since we use more than one db we're required to flush them all
+			connection.flushAll();
 		} catch (Exception e) {
 			// Connection may be closed in certain cases, like after pub/sub
 			// tests

src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-2015 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.
@@ -72,7 +72,7 @@ public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrati
 	@After
 	public void tearDown() {
 		try {
-			connection.flushDb();
+			connection.flushAll();
 		} catch (Exception e) {
 			// Jedis leaves some incomplete data in OutputStream on NPE caused
 			// by null key/value tests

src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineIntegrationTests.java

@@ -49,7 +49,7 @@ public class JedisConnectionPipelineIntegrationTests extends AbstractConnectionP
 	@After
 	public void tearDown() {
 		try {
-			connection.flushDb();
+			connection.flushAll();
 			connection.close();
 		} catch (Exception e) {
 			// Jedis leaves some incomplete data in OutputStream on NPE caused
@@ -135,7 +135,7 @@ public void testScriptLoadEvalSha() {
 	public void testEvalShaArrayStrings() {
 		super.testEvalShaArrayStrings();
 	}
-	
+
 	@Test(expected = UnsupportedOperationException.class)
 	@IfProfileValue(name = "redisVersion", value = "2.6+")
 	public void testEvalShaArrayBytes() {

src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionTransactionIntegrationTests.java

@@ -19,7 +19,6 @@
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.AbstractConnectionTransactionIntegrationTests;
 import org.springframework.data.redis.test.util.RelaxedJUnit4ClassRunner;
@@ -41,7 +40,7 @@ public class JedisConnectionTransactionIntegrationTests extends AbstractConnecti
 	@After
 	public void tearDown() {
 		try {
-			connection.flushDb();
+			connection.flushAll();
 			connection.close();
 		} catch (Exception e) {
 			// Jedis leaves some incomplete data in OutputStream on NPE caused
@@ -67,7 +66,7 @@ public void testScriptLoadEvalSha() {
 	public void testEvalShaArrayStrings() {
 		super.testEvalShaArrayStrings();
 	}
-	
+
 	@Test(expected = UnsupportedOperationException.class)
 	@IfProfileValue(name = "redisVersion", value = "2.6+")
 	public void testEvalShaArrayBytes() {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-2015 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.
@@ -60,7 +60,7 @@ public class JRedisConnectionIntegrationTests extends AbstractConnectionIntegrat
 	@After
 	public void tearDown() {
 		try {
-			connection.flushDb();
+			connection.flushAll();
 			connection.close();
 		} catch (DataAccessException e) {
 			// Jredis closes a connection on Exception (which some tests
@@ -435,7 +435,7 @@ public void testScriptLoadEvalSha() {
 	public void testEvalShaArrayStrings() {
 		super.testEvalShaArrayStrings();
 	}
-	
+
 	@Test(expected = UnsupportedOperationException.class)
 	@IfProfileValue(name = "redisVersion", value = "2.6+")
 	public void testEvalShaArrayBytes() {

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2013 the original author or authors.
+ * Copyright 2011-2015 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.
@@ -15,26 +15,31 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
-import com.lambdaworks.redis.RedisAsyncConnection;
-import com.lambdaworks.redis.RedisException;
+import static org.hamcrest.core.IsNull.*;
+import static org.junit.Assert.*;
+
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
+import org.springframework.data.redis.ConnectionFactoryTracker;
 import org.springframework.data.redis.RedisConnectionFailureException;
 import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.SettingsUtils;
 import org.springframework.data.redis.connection.DefaultStringRedisConnection;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.StringRedisConnection;
 
-import static org.junit.Assert.*;
+import com.lambdaworks.redis.RedisAsyncConnection;
+import com.lambdaworks.redis.RedisException;
 
 /**
  * Integration test of {@link LettuceConnectionFactory}
  * 
  * @author Jennifer Hickey
  * @author Thomas Darimont
+ * @author Christoph Strobl
  */
 public class LettuceConnectionFactoryTests {
 
@@ -44,6 +49,7 @@ public class LettuceConnectionFactoryTests {
 
 	@Before
 	public void setUp() {
+
 		factory = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort());
 		factory.afterPropertiesSet();
 		factory.setShutdownTimeout(0);
@@ -59,6 +65,11 @@ public void tearDown() {
 		}
 	}
 
+	@AfterClass
+	public static void cleanUp() {
+		ConnectionFactoryTracker.cleanUp();
+	}
+
 	@SuppressWarnings("rawtypes")
 	@Test
 	public void testGetNewConnectionOnError() throws Exception {
@@ -109,10 +120,14 @@ public void testValidateNoError() {
 
 	@Test
 	public void testSelectDb() {
+
 		LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort());
 		factory2.setShutdownTimeout(0);
 		factory2.setDatabase(1);
 		factory2.afterPropertiesSet();
+
+		ConnectionFactoryTracker.add(factory2);
+
 		StringRedisConnection connection2 = new DefaultStringRedisConnection(factory2.getConnection());
 		connection2.flushDb();
 		// put an item in database 0
@@ -213,6 +228,9 @@ public void testCreateFactoryWithPool() {
 		LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool);
 		factory2.setShutdownTimeout(0);
 		factory2.afterPropertiesSet();
+
+		ConnectionFactoryTracker.add(factory2);
+
 		RedisConnection conn2 = factory2.getConnection();
 		conn2.close();
 		factory2.destroy();
@@ -228,6 +246,9 @@ public void testLotsOfConnections() throws InterruptedException {
 		pool.afterPropertiesSet();
 		final LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool);
 		factory2.afterPropertiesSet();
+
+		ConnectionFactoryTracker.add(factory2);
+
 		for (int i = 1; i < 1000; i++) {
 			Thread th = new Thread(new Runnable() {
 				public void run() {
@@ -250,4 +271,30 @@ public void testConnectWithPassword() {
 		conn.bLPop(1, "key".getBytes());
 		conn.close();
 	}
+
+	/**
+	 * @see DATAREDIS-431
+	 */
+	@Test
+	public void dbIndexShouldBePropagatedCorrectly() {
+
+		LettuceConnectionFactory factory = new LettuceConnectionFactory();
+		factory.setDatabase(2);
+		factory.afterPropertiesSet();
+
+		ConnectionFactoryTracker.add(factory);
+
+		StringRedisConnection connectionToDbIndex2 = new DefaultStringRedisConnection(factory.getConnection());
+
+		try {
+
+			String key = "key-in-db-2";
+			connectionToDbIndex2.set(key, "the wheel of time");
+
+			assertThat(connection.get(key), nullValue());
+			assertThat(connectionToDbIndex2.get(key), notNullValue());
+		} finally {
+			connectionToDbIndex2.close();
+		}
+	}
 }

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java

@@ -46,12 +46,13 @@ public class LettuceConnectionUnitTestSuite {
 	public static class LettuceConnectionUnitTests extends AbstractConnectionUnitTestBase<RedisAsyncConnectionImpl> {
 
 		protected LettuceConnection connection;
+		private RedisClient clientMock;
 
 		@SuppressWarnings({ "unchecked" })
 		@Before
 		public void setUp() throws InvocationTargetException, IllegalAccessException {
 
-			RedisClient clientMock = mock(RedisClient.class);
+			clientMock = mock(RedisClient.class);
 			when(clientMock.connectAsync((RedisCodec) any())).thenReturn(getNativeRedisConnectionMock());
 			connection = new LettuceConnection(0, clientMock);
 		}
@@ -142,6 +143,18 @@ public void slaveOfNoOneShouldBeSentCorrectly() {
 		public void shouldThrowExceptionWhenAccessingRedisSentinelsCommandsWhenNoSentinelsConfigured() {
 			connection.getSentinelConnection();
 		}
+
+		/**
+		 * @see DATAREDIS-431
+		 */
+		@Test
+		public void dbIndexShouldBeSetWhenOptainingConnection() {
+
+			connection = new LettuceConnection(null, 0, clientMock, null, 1);
+			connection.getNativeConnection();
+
+			verify(getNativeRedisConnectionMock(), times(1)).select(1);
+		}
 	}
 
 	public static class LettucePipelineConnectionUnitTests extends LettuceConnectionUnitTests {

src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-2015 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.
@@ -51,7 +51,7 @@ public class SrpConnectionIntegrationTests extends AbstractConnectionIntegration
 	@After
 	public void tearDown() {
 		try {
-			connection.flushDb();
+			connection.flushAll();
 		} catch (Exception e) {
 			// SRP doesn't allow other commands to be executed once subscribed,
 			// so