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 8632de64f..018750975 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 @@ -88,35 +88,46 @@ public class LettuceConnection implements RedisConnection { /** * Instantiates a new lettuce connection. * - * @param dedicatedConnection - * A dedicated native connection. Can be used for any operation. * @param timeout * The connection timeout (in milliseconds) * @param client - * The {@link RedisClient} to use when making pub/sub connections + * The {@link RedisClient} to use when instantiating a native + * connection */ - public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection dedicatedConnection, long timeout, - RedisClient client) { - this(null, dedicatedConnection, timeout, client, null); + public LettuceConnection(long timeout, RedisClient client) { + this(null, timeout, client, null); } /** * Instantiates a new lettuce connection. * - * @param dedicatedConnection - * A dedicated native connection. Can be used for any operation. + * @param timeout + * The connection timeout (in milliseconds) * @param client The + * {@link RedisClient} to use when instantiating a pub/sub + * connection + * @param pool + * The connection pool to use for all other native connections + */ + public LettuceConnection(long timeout, RedisClient client, LettucePool pool) { + this(null, timeout, client, pool); + } + + /** + * Instantiates a new lettuce connection. + * + * @param sharedConnection + * A native connection that is shared with other + * {@link LettuceConnection}s. Will 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 - * An optional connection pool, from which the dedicated - * connection came. If pool is set, the dedicated connection will - * be returned to the pool on close + * The {@link RedisClient} to use when making pub/sub, blocking, + * and tx connections */ - public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection dedicatedConnection, long timeout, - RedisClient client, LettucePool pool) { - this(null, dedicatedConnection, timeout, client, pool); + public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection sharedConnection, + long timeout, RedisClient client) { + this(sharedConnection, timeout, client, null); } /** @@ -126,47 +137,19 @@ public class LettuceConnection implements RedisConnection { * A native connection that is shared with other * {@link LettuceConnection}s. Should not be used for * transactions or blocking operations - * @param dedicatedConnection - * A dedicated native connection. Can be used for any operation. * @param timeout * The connection timeout (in milliseconds) * @param client - * The {@link RedisClient} to use when making pub/sub connections - */ - public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection sharedConnection, - com.lambdaworks.redis.RedisAsyncConnection dedicatedConnection, long timeout, - RedisClient client) { - this(sharedConnection, dedicatedConnection, timeout, client, null); - } - - /** - * Instantiates a new lettuce connection. - * - * @param sharedConnection - * A native connection that is shared with other - * {@link LettuceConnection}s. Should not be used for - * transactions or blocking operations - * @param dedicatedConnection - * A dedicated native connection. Can be used for any operation. - * @param timeout - * The connection timeout (in milliseconds) - * @param client - * The {@link RedisClient} to use when making pub/sub connections + * The {@link RedisClient} to use when making pub/sub connections * @param pool - * An optional connection pool, from which the dedicated - * connection came. If pool is set, the dedicated connection will - * be returned to the pool on close + * The connection pool to use for blocking and tx operations */ public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection sharedConnection, - com.lambdaworks.redis.RedisAsyncConnection dedicatedConnection, long timeout, - RedisClient client, LettucePool pool) { - Assert.notNull(dedicatedConnection, "a valid dedicated connection is required"); + long timeout, RedisClient client, LettucePool pool) { this.asyncSharedConn = sharedConnection; - this.asyncDedicatedConn = dedicatedConnection; this.timeout = timeout; - this.sharedConn = sharedConnection != null ? - new com.lambdaworks.redis.RedisConnection(asyncSharedConn) : null; - this.dedicatedConn = new com.lambdaworks.redis.RedisConnection(dedicatedConnection); + this.sharedConn = sharedConnection != null ? new com.lambdaworks.redis.RedisConnection( + asyncSharedConn) : null; this.client = client; this.pool = pool; } @@ -223,19 +206,20 @@ public class LettuceConnection implements RedisConnection { public void close() throws DataAccessException { isClosed = true; - if(pool != null) { - if (!broken) { - pool.returnResource(asyncDedicatedConn); - }else { - pool.returnBrokenResource(asyncDedicatedConn); + if(asyncDedicatedConn != null) { + if(pool != null) { + if (!broken) { + pool.returnResource(asyncDedicatedConn); + }else { + pool.returnBrokenResource(asyncDedicatedConn); + } + } else { + try { + asyncDedicatedConn.close(); + } catch (RuntimeException ex) { + throw convertLettuceAccessException(ex); + } } - return; - } - - try { - asyncDedicatedConn.close(); - } catch (RuntimeException ex) { - throw convertLettuceAccessException(ex); } if (subscription != null) { @@ -549,30 +533,28 @@ public class LettuceConnection implements RedisConnection { } } - public void discard() { isMulti = false; try { if (isPipelined()) { - pipeline(getAsyncTxConnection().discard()); + pipeline(getAsyncDedicatedConnection().discard()); return; } - getTxConnection().discard(); + getDedicatedConnection().discard(); } catch (Exception ex) { throw convertLettuceAccessException(ex); } } - public List exec() { isMulti = false; try { if (isPipelined()) { - getAsyncTxConnection().exec(); - return null; + getAsyncDedicatedConnection().exec(); + return null; } - List results = getTxConnection().exec(); - if(results.isEmpty()) { + List results = getDedicatedConnection().exec(); + if (results.isEmpty()) { return null; } return results; @@ -701,16 +683,15 @@ public class LettuceConnection implements RedisConnection { isMulti = true; try { if (isPipelined()) { - pipeline(getAsyncTxConnection().multi()); + pipeline(getAsyncDedicatedConnection().multi()); return; } - getTxConnection().multi(); + getDedicatedConnection().multi(); } catch (Exception ex) { throw convertLettuceAccessException(ex); } } - public Boolean persist(byte[] key) { try { if (isPipelined()) { @@ -816,28 +797,25 @@ public class LettuceConnection implements RedisConnection { } } - public void unwatch() { try { if (isPipelined()) { - pipeline(getAsyncTxConnection().unwatch()); + pipeline(getAsyncDedicatedConnection().unwatch()); return; } - getTxConnection().unwatch(); + getDedicatedConnection().unwatch(); } catch (Exception ex) { throw convertLettuceAccessException(ex); } } - public void watch(byte[]... keys) { try { if (isPipelined()) { - pipeline(getAsyncTxConnection().watch(keys)); + pipeline(getAsyncDedicatedConnection().watch(keys)); return; - } - else { - getTxConnection().watch(keys); + } else { + getDedicatedConnection().watch(keys); } } catch (Exception ex) { throw convertLettuceAccessException(ex); @@ -1159,10 +1137,10 @@ public class LettuceConnection implements RedisConnection { public List bLPop(int timeout, byte[]... keys) { try { if (isPipelined()) { - pipeline(getAsyncBlockingConnection().blpop(timeout, keys)); + pipeline(getAsyncDedicatedConnection().blpop(timeout, keys)); return null; } - return LettuceUtils.toList(getBlockingConnection().blpop(timeout, keys)); + return LettuceUtils.toList(getDedicatedConnection().blpop(timeout, keys)); } catch (Exception ex) { throw convertLettuceAccessException(ex); } @@ -1171,16 +1149,15 @@ public class LettuceConnection implements RedisConnection { public List bRPop(int timeout, byte[]... keys) { try { if (isPipelined()) { - pipeline(getAsyncBlockingConnection().brpop(timeout, keys)); + pipeline(getAsyncDedicatedConnection().brpop(timeout, keys)); return null; } - return LettuceUtils.toList(getBlockingConnection().brpop(timeout, keys)); + return LettuceUtils.toList(getDedicatedConnection().brpop(timeout, keys)); } catch (Exception ex) { throw convertLettuceAccessException(ex); } } - public byte[] lIndex(byte[] key, long index) { try { if (isPipelined()) { @@ -1307,10 +1284,10 @@ public class LettuceConnection implements RedisConnection { public byte[] bRPopLPush(int timeout, byte[] srcKey, byte[] dstKey) { try { if (isPipelined()) { - pipeline(getAsyncBlockingConnection().brpoplpush(timeout, srcKey, dstKey)); + pipeline(getAsyncDedicatedConnection().brpoplpush(timeout, srcKey, dstKey)); return null; } - return getBlockingConnection().brpoplpush(timeout, srcKey, dstKey); + return getDedicatedConnection().brpoplpush(timeout, srcKey, dstKey); } catch (Exception ex) { throw convertLettuceAccessException(ex); } @@ -2220,42 +2197,43 @@ public class LettuceConnection implements RedisConnection { } private RedisAsyncConnection getAsyncConnection() { - if(isQueueing()) { - return getAsyncTxConnection(); + if (isQueueing()) { + return getAsyncDedicatedConnection(); } - if(asyncSharedConn != null) { + if (asyncSharedConn != null) { return asyncSharedConn; } - return asyncDedicatedConn; + return getAsyncDedicatedConnection(); } private com.lambdaworks.redis.RedisConnection getConnection() { - if(isQueueing()) { - return getTxConnection(); + if (isQueueing()) { + return getDedicatedConnection(); } - if(sharedConn != null) { + if (sharedConn != null) { return sharedConn; } - return dedicatedConn; + return getDedicatedConnection(); } - private RedisAsyncConnection getAsyncBlockingConnection() { + private RedisAsyncConnection getAsyncDedicatedConnection() { + if(asyncDedicatedConn == null) { + if(this.pool != null) { + this.asyncDedicatedConn = pool.getResource(); + } else { + this.asyncDedicatedConn = client.connectAsync(LettuceUtils.CODEC); + } + } return asyncDedicatedConn; } - private com.lambdaworks.redis.RedisConnection getBlockingConnection() { + private com.lambdaworks.redis.RedisConnection getDedicatedConnection() { + if(dedicatedConn == null) { + this.dedicatedConn = new com.lambdaworks.redis.RedisConnection(getAsyncDedicatedConnection()); + } return dedicatedConn; } - private RedisAsyncConnection getAsyncTxConnection() { - return asyncDedicatedConn; - } - - private com.lambdaworks.redis.RedisConnection getTxConnection() { - return dedicatedConn; - - } - private Future asyncBitOp(BitOperation op, byte[] destination, byte[]... keys) { switch (op) { case AND: 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 ee6674cd7..b60ffca15 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 @@ -106,7 +106,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea } public RedisConnection getConnection() { - return new LettuceConnection(getSharedConnection(), createLettuceConnector(false), timeout, client, pool); + return new LettuceConnection(getSharedConnection(), timeout, client, pool); } public void initConnection() { @@ -114,7 +114,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea if (this.connection != null) { resetConnection(); } - this.connection = createLettuceConnector(true); + this.connection = createLettuceConnector(); } } @@ -313,16 +313,13 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea } } - protected RedisAsyncConnection createLettuceConnector(boolean shared) { - if(pool != null && !shared) { - return pool.getResource(); - } + protected RedisAsyncConnection createLettuceConnector() { try { - RedisAsyncConnection dedicatedConnection = client.connectAsync(LettuceUtils.CODEC); + RedisAsyncConnection connection = client.connectAsync(LettuceUtils.CODEC); if(dbIndex > 0) { - dedicatedConnection.select(dbIndex); + connection.select(dbIndex); } - return dedicatedConnection; + return connection; } 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 017db1cc6..e7f2601ff 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 @@ -16,13 +16,12 @@ package org.springframework.data.redis.connection.lettuce; import static org.junit.Assert.assertEquals; -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 static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; -import org.apache.commons.pool.impl.GenericObjectPool.Config; import org.junit.After; import org.junit.Before; import org.junit.Ignore; @@ -31,7 +30,6 @@ 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; @@ -182,14 +180,6 @@ public class LettuceConnectionFactoryTests { factory.afterPropertiesSet(); } - @Test(expected=RedisConnectionFailureException.class) - public void testGetConnectionNotSharedException() { - factory.setShareNativeConnection(false); - factory.setHostName("fakeHost"); - factory.afterPropertiesSet(); - factory.getConnection(); - } - @Test(expected=RedisConnectionFailureException.class) public void testGetConnectionSharedException() { factory.setShareNativeConnection(false); @@ -217,29 +207,22 @@ public class LettuceConnectionFactoryTests { } @Test - public void testCreateLettuceConnectorWithPool() { - Config poolConfig = new Config(); - poolConfig.maxActive = 1; - poolConfig.maxWait = 1; - DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort(), poolConfig); + public void testCreateFactoryWithPool() { + DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(), + SettingsUtils.getPort()); pool.afterPropertiesSet(); LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool); 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) { - } + factory2.getConnection(); } @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(new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort())); + DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort()); + pool.afterPropertiesSet(); + final LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool); factory2.afterPropertiesSet(); for(int i=1;i< 1000;i++) { Thread th = new Thread(new Runnable() {