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.
This commit is contained in:
committed by
Oliver Gierke
parent
3d8aeb3219
commit
2c59c3b678
@@ -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<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 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;
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user