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 f49476c89..4667417b5 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 @@ -36,6 +36,7 @@ import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.DataType; import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.connection.Pool; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisPipelineException; import org.springframework.data.redis.connection.RedisSubscribedConnectionException; @@ -47,6 +48,7 @@ import org.springframework.util.ObjectUtils; import com.lambdaworks.redis.RedisAsyncConnection; import com.lambdaworks.redis.RedisClient; +import com.lambdaworks.redis.RedisCommandInterruptedException; import com.lambdaworks.redis.RedisException; import com.lambdaworks.redis.SortArgs; import com.lambdaworks.redis.ZStoreArgs; @@ -65,16 +67,10 @@ import com.lambdaworks.redis.pubsub.RedisPubSubConnection; */ public class LettuceConnection implements RedisConnection { - private final com.lambdaworks.redis.RedisAsyncConnection asyncConn; - private final com.lambdaworks.redis.RedisConnection con; - - // Connection to do tx operations - not shared among threads like asyncConn - private com.lambdaworks.redis.RedisAsyncConnection asyncTxConn; - private com.lambdaworks.redis.RedisConnection txConn; - - // Connection to do blocking operations - not shared among threads like asyncConn - private com.lambdaworks.redis.RedisAsyncConnection asyncBlockingConn; - private com.lambdaworks.redis.RedisConnection blockingConn; + private final com.lambdaworks.redis.RedisAsyncConnection asyncSharedConn; + private final com.lambdaworks.redis.RedisConnection sharedConn; + private com.lambdaworks.redis.RedisAsyncConnection asyncDedicatedConn; + private com.lambdaworks.redis.RedisConnection dedicatedConn; private final RedisCodec codec = LettuceUtils.CODEC; private final long timeout; @@ -83,10 +79,22 @@ public class LettuceConnection implements RedisConnection { private boolean isClosed = false; private boolean isMulti = false; private boolean isPipelined = false; - private boolean closeNativeConnection = true; private List> ppline; private RedisClient client; private volatile LettuceSubscription subscription; + private Pool> pool; + /** flag indicating whether the connection needs to be dropped or not */ + private boolean broken = false; + + public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection dedicatedConnection, long timeout, + RedisClient client) { + this(null, dedicatedConnection, timeout, client, null); + } + + public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection dedicatedConnection, long timeout, + RedisClient client, Pool> pool) { + this(null, dedicatedConnection, timeout, client, pool); + } /** * Instantiates a new lettuce connection. @@ -95,9 +103,10 @@ public class LettuceConnection implements RedisConnection { * @param timeout the connection timeout (in milliseconds) * @param client The Lettuce client */ - public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection connection, long timeout, + public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection sharedConnection, + com.lambdaworks.redis.RedisAsyncConnection dedicatedConnection, long timeout, RedisClient client) { - this(connection, timeout, client, true); + this(sharedConnection, dedicatedConnection, timeout, client, null); } /** @@ -109,22 +118,30 @@ public class LettuceConnection implements RedisConnection { * @param closeNativeConnection True if native connection should be called on {@link #close()}. This should * be set to false if the native connection is shared. */ - public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection connection, long timeout, - RedisClient client, boolean closeNativeConnection) { - Assert.notNull(connection, "a valid connection is required"); - - this.asyncConn = connection; + public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection sharedConnection, + com.lambdaworks.redis.RedisAsyncConnection dedicatedConnection, long timeout, + RedisClient client, Pool> pool) { + Assert.notNull(dedicatedConnection, "a valid dedicated connection is required"); + this.asyncSharedConn = sharedConnection; + this.asyncDedicatedConn = dedicatedConnection; this.timeout = timeout; - this.con = new com.lambdaworks.redis.RedisConnection(asyncConn); + this.sharedConn = sharedConnection != null ? + new com.lambdaworks.redis.RedisConnection(asyncSharedConn) : null; + this.dedicatedConn = new com.lambdaworks.redis.RedisConnection(dedicatedConnection); this.client = client; - this.closeNativeConnection = closeNativeConnection; + this.pool = pool; } protected DataAccessException convertLettuceAccessException(Exception ex) { if (ex instanceof RedisException) { + if(!(ex instanceof RedisCommandInterruptedException)) { + // Some connection issues (i.e. connection is closed) are thrown as RedisExceptions + broken = true; + } return LettuceUtils.convertRedisAccessException((RedisException) ex); } if (ex instanceof ChannelException) { + broken = true; return new RedisConnectionFailureException("Redis connection failed", ex); } if (ex instanceof TimeoutException) { @@ -167,24 +184,19 @@ public class LettuceConnection implements RedisConnection { public void close() throws DataAccessException { isClosed = true; - if(asyncBlockingConn != null) { - asyncBlockingConn.close(); - asyncBlockingConn = null; - blockingConn = null; - } - - if(asyncTxConn != null) { - asyncTxConn.close(); - asyncTxConn = null; - txConn = null; - } - - if(closeNativeConnection) { - try { - asyncConn.close(); - } catch (RuntimeException ex) { - throw convertLettuceAccessException(ex); + if(pool != null) { + if (!broken) { + pool.returnResource(asyncDedicatedConn); + }else { + pool.returnBrokenResource(asyncDedicatedConn); } + return; + } + + try { + asyncDedicatedConn.close(); + } catch (RuntimeException ex) { + throw convertLettuceAccessException(ex); } if (subscription != null) { @@ -712,8 +724,18 @@ public class LettuceConnection implements RedisConnection { public void select(int dbIndex) { - throw new UnsupportedOperationException("Selecting a new database not supported due to shared connection. " + + if(asyncSharedConn != null) { + throw new UnsupportedOperationException("Selecting a new database not supported due to shared connection. " + "Use separate ConnectionFactorys to work with multiple databases"); + } + try { + if (isPipelined()) { + throw new UnsupportedOperationException("Lettuce blocks for #select"); + } + getConnection().select(dbIndex); + } catch (Exception ex) { + throw convertLettuceAccessException(ex); + } } @@ -2133,48 +2155,37 @@ public class LettuceConnection implements RedisConnection { if(isQueueing()) { return getAsyncTxConnection(); } - return asyncConn; + if(asyncSharedConn != null) { + return asyncSharedConn; + } + return asyncDedicatedConn; } private com.lambdaworks.redis.RedisConnection getConnection() { if(isQueueing()) { return getTxConnection(); } - return con; + if(sharedConn != null) { + return sharedConn; + } + return dedicatedConn; } private RedisAsyncConnection getAsyncBlockingConnection() { - if(isQueueing()) { - return getAsyncTxConnection(); - } - if(asyncBlockingConn == null) { - asyncBlockingConn = client.connectAsync(LettuceUtils.CODEC); - } - return asyncBlockingConn; + return asyncDedicatedConn; } private com.lambdaworks.redis.RedisConnection getBlockingConnection() { - if(isQueueing()) { - return getTxConnection(); - } - if(blockingConn == null) { - blockingConn = new com.lambdaworks.redis.RedisConnection(getAsyncBlockingConnection()); - } - return blockingConn; + return dedicatedConn; } private RedisAsyncConnection getAsyncTxConnection() { - if(asyncTxConn == null) { - asyncTxConn = client.connectAsync(LettuceUtils.CODEC); - } - return asyncTxConn; + return asyncDedicatedConn; } private com.lambdaworks.redis.RedisConnection getTxConnection() { - if(txConn == null) { - txConn = new com.lambdaworks.redis.RedisConnection(getAsyncTxConnection()); - } - return txConn; + return dedicatedConn; + } private Future asyncBitOp(BitOperation op, byte[] destination, byte[]... keys) { 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 19def0d51..e3b1b8b91 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 @@ -24,6 +24,7 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.dao.DataAccessException; import org.springframework.data.redis.RedisConnectionFailureException; +import org.springframework.data.redis.connection.Pool; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.util.Assert; @@ -45,6 +46,11 @@ import com.lambdaworks.redis.RedisException; * therefore it is not validated by default on {@link #getConnection()}. Use * {@link #setValidateConnection(boolean)} to change this behavior if necessary. * + * Inject a {@link Pool} to pool dedicated connections. If shareNativeConnection is + * true, the pool will be used to select a connection for blocking and tx operations only, + * which should not share a connection. If native connection sharing is disabled, + * the selected connection will be used for all operations. + * * @author Costin Leau * @author Jennifer Hickey */ @@ -60,6 +66,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea private boolean validateConnection = false; private boolean shareNativeConnection = true; private RedisAsyncConnection connection; + private Pool> pool; private int dbIndex = 0; /** Synchronization monitor for the shared Connection */ private final Object connectionMonitor = new Object(); @@ -80,6 +87,12 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea this.port = port; } + public LettuceConnectionFactory(String host, int port, Pool> pool) { + this.hostName = host; + this.port = port; + this.pool = pool; + } + public void afterPropertiesSet() { client = new RedisClient(hostName, port); client.setDefaultTimeout(timeout, TimeUnit.MILLISECONDS); @@ -95,11 +108,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea } public RedisConnection getConnection() { - RedisAsyncConnection nativeConnection = getNativeConnection(); - if (dbIndex > 0) { - nativeConnection.select(dbIndex); - } - return new LettuceConnection(nativeConnection, timeout, client, !shareNativeConnection); + return new LettuceConnection(getSharedConnection(), createLettuceConnector(false), timeout, client, pool); } public void initConnection() { @@ -107,7 +116,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea if (this.connection != null) { resetConnection(); } - this.connection = createLettuceConnector(); + this.connection = createLettuceConnector(true); } } @@ -272,7 +281,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea this.dbIndex = index; } - protected RedisAsyncConnection getNativeConnection() { + protected RedisAsyncConnection getSharedConnection() { if (shareNativeConnection) { synchronized (this.connectionMonitor) { if (this.connection == null) { @@ -284,13 +293,20 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea return this.connection; } } else { - return createLettuceConnector(); + return null; } } - private RedisAsyncConnection createLettuceConnector() { + protected RedisAsyncConnection createLettuceConnector(boolean shared) { + if(pool != null && !shared) { + return pool.getResource(); + } try { - return client.connectAsync(LettuceUtils.CODEC); + RedisAsyncConnection dedicatedConnection = client.connectAsync(LettuceUtils.CODEC); + if(dbIndex > 0) { + dedicatedConnection.select(dbIndex); + } + return dedicatedConnection; } catch (RedisException e) { throw new RedisConnectionFailureException("Unable to connect to Redis on " + getHostName() + ":" + getPort(), e); 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 a8bbdf7e3..433ee0d61 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 @@ -20,13 +20,18 @@ import static org.junit.Assert.fail; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; + +import org.apache.commons.pool.impl.GenericObjectPool.Config; import org.junit.After; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; 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.PoolException; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.StringRedisConnection; @@ -194,20 +199,57 @@ public class LettuceConnectionFactoryTests { factory.getConnection(); } - @Test(expected=RedisConnectionFailureException.class) - public void testGetNativeConnectionNotSharedException() { + @Test + public void testGetSharedConnectionNotShared() { factory.setShareNativeConnection(false); factory.setHostName("fakeHost"); factory.afterPropertiesSet(); - factory.getNativeConnection(); + assertNull(factory.getSharedConnection()); } @Test(expected=RedisConnectionFailureException.class) - public void testGetNativeConnectionSharedException() { + public void testGetSharedConnectionSharedException() { factory.setShareNativeConnection(false); factory.setHostName("fakeHost"); factory.afterPropertiesSet(); factory.setShareNativeConnection(true); - factory.getNativeConnection(); + factory.getSharedConnection(); + } + + @Test + public void testCreateLettuceConnectorWithPool() { + Config poolConfig = new Config(); + poolConfig.maxActive = 1; + poolConfig.maxWait = 1; + LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), + SettingsUtils.getPort(), new LettucePool(SettingsUtils.getHost(), SettingsUtils.getPort(), + poolConfig)); + factory2.afterPropertiesSet(); + factory2.createLettuceConnector(false); + try { + // We know we are using the pool if we try to get another connection + // when maxActive is set to 1 + factory2.createLettuceConnector(false); + fail("Expected Exception retrieving connection from pool"); + } catch (PoolException e) { + } + } + + @Ignore("Uncomment this test to manually check connection reuse in a pool scenario") + @Test + public void testLotsOfConnections() throws InterruptedException { + // Running a netstat here should show only the 8 conns from the pool (plus 2 from setUp and 1 from factory2 afterPropertiesSet for shared conn) + final LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), + SettingsUtils.getPort(), new LettucePool(SettingsUtils.getHost(), SettingsUtils.getPort())); + factory2.afterPropertiesSet(); + for(int i=1;i< 1000;i++) { + Thread th = new Thread(new Runnable() { + public void run() { + factory2.getConnection().bRPop(50000, "foo".getBytes()); + } + }); + th.start(); + } + Thread.sleep(234234234); } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java index 96e96fd85..a9635ca3a 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java @@ -16,13 +16,12 @@ package org.springframework.data.redis.connection.lettuce; -import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.junit.Assume.assumeTrue; import static org.springframework.data.redis.SpinBarrier.waitFor; -import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Ignore; @@ -35,12 +34,15 @@ import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.TestCondition; import org.springframework.data.redis.connection.AbstractConnectionIntegrationTests; import org.springframework.data.redis.connection.DefaultStringRedisConnection; -import org.springframework.data.redis.connection.ReturnType; +import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisStringCommands.BitOperation; +import org.springframework.data.redis.connection.ReturnType; import org.springframework.test.annotation.IfProfileValue; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import com.lambdaworks.redis.RedisAsyncConnection; + /** * Integration test of {@link LettuceConnection} * @@ -104,14 +106,6 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra } catch (RedisSystemException e) { // expected, can't resume tx after closing conn } - // can do normal ops after closing - connection.get("txclose"); - - // can complete a new tx after closing - connection.multi(); - connection.set("txclose", "bar"); - List results = connection.exec(); - assertEquals("OK", results.get(0)); } @Test @@ -120,15 +114,85 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra connection.bLPop(1, "what".getBytes()); connection.close(); - // can do blocking ops after closing + // can still operate on shared conn connection.lPush("what", "boo"); - connection.bLPop(1, "what".getBytes()); - // we can do regular ops - connection.get("somekey"); + try { + // can't do blocking ops after closing + connection.bLPop(1, "what".getBytes()); + fail("Expected exception using a closed conn for dedicated ops"); + }catch(RedisSystemException e) { + } + } - // we can start a tx + @Test + public void testClosePooledConnectionWithShared() { + LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort(), + new LettucePool(SettingsUtils.getHost(), SettingsUtils.getPort())); + factory2.afterPropertiesSet(); + RedisConnection connection = factory2.getConnection(); + // Use the connection to make sure the channel is initialized, else nothing happens on close + connection.ping(); + connection.close(); + // The shared connection should not be closed + connection.ping(); + + // The dedicated connection should not be closed b/c it's part of a pool connection.multi(); + factory2.destroy(); + } + + @Test + public void testClosePooledConnectionNotShared() { + LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort(), + new LettucePool(SettingsUtils.getHost(), SettingsUtils.getPort())); + factory2.setShareNativeConnection(false); + factory2.afterPropertiesSet(); + RedisConnection connection = factory2.getConnection(); + // Use the connection to make sure the channel is initialized, else nothing happens on close + connection.ping(); + connection.close(); + // The dedicated connection should not be closed + connection.ping(); + factory2.destroy(); + } + + @Test + public void testCloseNonPooledConnectionNotShared() { + LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort()); + factory2.setShareNativeConnection(false); + factory2.afterPropertiesSet(); + RedisConnection connection = factory2.getConnection(); + // Use the connection to make sure the channel is initialized, else nothing happens on close + connection.ping(); + connection.close(); + // The dedicated connection should be closed + try { + connection.set("foo".getBytes(), "bar".getBytes()); + fail("Exception should be thrown trying to use a closed connection"); + }catch(RedisSystemException e) { + } + factory2.destroy(); + } + + @SuppressWarnings("rawtypes") + @Test + public void testCloseReturnBrokenResourceToPool() { + LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort(), + new LettucePool(SettingsUtils.getHost(), SettingsUtils.getPort())); + factory2.setShareNativeConnection(false); + factory2.afterPropertiesSet(); + RedisConnection connection = factory2.getConnection(); + // Use the connection to make sure the channel is initialized, else nothing happens on close + connection.ping(); + ((RedisAsyncConnection)connection.getNativeConnection()).close(); + try { + connection.ping(); + fail("Exception should be thrown trying to use a closed connection"); + }catch(RedisSystemException e) { + } + connection.close(); + factory2.destroy(); } @Test(expected = UnsupportedOperationException.class) @@ -136,6 +200,17 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra connection.select(1); } + @Test + public void testSelectNotShared() { + LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort(), + new LettucePool(SettingsUtils.getHost(), SettingsUtils.getPort())); + factory2.setShareNativeConnection(false); + factory2.afterPropertiesSet(); + RedisConnection connection = factory2.getConnection(); + connection.select(2); + factory2.destroy(); + } + @Test(expected=UnsupportedOperationException.class) @IfProfileValue(name = "redisVersion", value = "2.6") public void testBitOpNotMultipleSources() {