diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 773d337d7..a00497b3d 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/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 class LettuceConnection extends AbstractRedisConnection { */ public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection 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 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 class LettuceConnection extends AbstractRedisConnection { } subscription = null; } + + this.dbIndex = defaultDbIndex; } public boolean isClosed() { @@ -1075,6 +1098,7 @@ public class LettuceConnection extends AbstractRedisConnection { } 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 class LettuceConnection extends AbstractRedisConnection { throw new UnsupportedOperationException("Lettuce blocks for #select"); } try { + + this.dbIndex = dbIndex; if (isQueueing()) { transaction(new LettuceTxStatusResult(getConnection().select(dbIndex))); return; @@ -3412,7 +3438,9 @@ public class LettuceConnection extends AbstractRedisConnection { if (this.pool != null) { this.asyncDedicatedConn = pool.getResource(); } else { + this.asyncDedicatedConn = client.connectAsync(CODEC); + this.asyncDedicatedConn.select(dbIndex); } } return asyncDedicatedConn; diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java index 4296d6b78..48640226b 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java @@ -118,8 +118,13 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea 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; } diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index 4ee5799af..4bf7dca8d 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -111,7 +111,9 @@ public abstract class AbstractConnectionIntegrationTests { @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 diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionIntegrationTests.java index 01d2e01b4..64615097c 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionIntegrationTests.java +++ b/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 diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineIntegrationTests.java index e81eee9ff..cc3b99c27 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineIntegrationTests.java +++ b/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 class JedisConnectionPipelineIntegrationTests extends AbstractConnectionP public void testEvalShaArrayStrings() { super.testEvalShaArrayStrings(); } - + @Test(expected = UnsupportedOperationException.class) @IfProfileValue(name = "redisVersion", value = "2.6+") public void testEvalShaArrayBytes() { diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionTransactionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionTransactionIntegrationTests.java index 32f3e14eb..dfb4ab14b 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionTransactionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionTransactionIntegrationTests.java @@ -19,7 +19,6 @@ import org.junit.After; 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 class JedisConnectionTransactionIntegrationTests extends AbstractConnecti public void testEvalShaArrayStrings() { super.testEvalShaArrayStrings(); } - + @Test(expected = UnsupportedOperationException.class) @IfProfileValue(name = "redisVersion", value = "2.6+") public void testEvalShaArrayBytes() { diff --git a/src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java index 91c328407..7dd209d02 100644 --- a/src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java +++ b/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 class JRedisConnectionIntegrationTests extends AbstractConnectionIntegrat public void testEvalShaArrayStrings() { super.testEvalShaArrayStrings(); } - + @Test(expected = UnsupportedOperationException.class) @IfProfileValue(name = "redisVersion", value = "2.6+") public void testEvalShaArrayBytes() { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java index b28660182..4c3fbdbf2 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java +++ b/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,12 +15,15 @@ */ 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; @@ -28,13 +31,15 @@ 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 class LettuceConnectionFactoryTests { } } + @AfterClass + public static void cleanUp() { + ConnectionFactoryTracker.cleanUp(); + } + @SuppressWarnings("rawtypes") @Test public void testGetNewConnectionOnError() throws Exception { @@ -109,10 +120,14 @@ public class LettuceConnectionFactoryTests { @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 class LettuceConnectionFactoryTests { 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 class LettuceConnectionFactoryTests { 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 class LettuceConnectionFactoryTests { 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(); + } + } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java index 5719a18e7..6025c23a4 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java @@ -46,12 +46,13 @@ public class LettuceConnectionUnitTestSuite { public static class LettuceConnectionUnitTests extends AbstractConnectionUnitTestBase { 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 class LettuceConnectionUnitTestSuite { 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 { diff --git a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java index ec21dbd0e..681ba1497 100644 --- a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java +++ b/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