Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-830 - Polishing.

Browse Source 
Reformat code. Add override-comments. Encapsulate connection in LettuceSubscription. Fix typos.

Original Pull Request: #341
This commit is contained in:
Mark Paluch
2018-05-09 08:53:46 +02:00
committed by Christoph Strobl
parent a2bc45cfa2
commit e766b18d8e
4 changed files with 138 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -424,6 +424,10 @@ public class LettuceConnection extends AbstractRedisConnection {
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.AbstractRedisConnection#close()
*/
@Override
public void close() throws DataAccessException {
@@ -453,26 +457,48 @@ public class LettuceConnection extends AbstractRedisConnection {
this.dbIndex = defaultDbIndex;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisConnection#isClosed()
*/
@Override
public boolean isClosed() {
return isClosed && !isSubscribed();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisConnection#getNativeConnection()
*/
@Override
public RedisClusterAsyncCommands<byte[], byte[]> getNativeConnection() {
return (subscription != null ? subscription.pubsub.async() : getAsyncConnection());
LettuceSubscription subscription = this.subscription;
return (subscription != null ? subscription.getNativeConnection().async() : getAsyncConnection());
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisConnection#isQueueing()
*/
@Override
public boolean isQueueing() {
return isMulti;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisConnection#isPipelined()
*/
@Override
public boolean isPipelined() {
return isPipelined;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisConnection#openPipeline()
*/
@Override
public void openPipeline() {
if (!isPipelined) {
@@ -481,6 +507,10 @@ public class LettuceConnection extends AbstractRedisConnection {
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisConnection#closePipeline()
*/
@Override
public List<Object> closePipeline() {
@@ -547,7 +577,6 @@ public class LettuceConnection extends AbstractRedisConnection {
* @see org.springframework.data.redis.connection.RedisServerCommands#shutdown(org.springframework.data.redis.connection.RedisServerCommands.ShutdownOption)
*/
@Override
public byte[] echo(byte[] message) {
try {
if (isPipelined()) {
@@ -714,43 +743,62 @@ public class LettuceConnection extends AbstractRedisConnection {
// Pub/Sub functionality
//
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisPubSubCommands#publish(byte[], byte[])
*/
@Override
public Long publish(byte[] channel, byte[] message) {
try {
if (isPipelined()) {
pipeline(newLettuceResult(getAsyncConnection().publish(channel, message)));
return null;
}
if (isQueueing()) {
transaction(newLettuceResult(getAsyncConnection().publish(channel, message)));
return null;
}
return getConnection().publish(channel, message);
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisPubSubCommands#getSubscription()
*/
@Override
public Subscription getSubscription() {
return subscription;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisPubSubCommands#isSubscribed()
*/
@Override
public boolean isSubscribed() {
return (subscription != null && subscription.isAlive());
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisPubSubCommands#pSubscribe(org.springframework.data.redis.connection.MessageListener, byte[][])
*/
@Override
public void pSubscribe(MessageListener listener, byte[]... patterns) {
checkSubscription();
if (isQueueing()) {
throw new UnsupportedOperationException();
}
if (isPipelined()) {
throw new UnsupportedOperationException();
if (isQueueing() || isPipelined()) {
throw new UnsupportedOperationException("Transaction/Pipelining is not supported for Pub/Sub subscriptions!");
}
try {
subscription = new LettuceSubscription(listener, switchToPubSub(), connectionProvider);
subscription.pSubscribe(patterns);
@@ -759,17 +807,22 @@ public class LettuceConnection extends AbstractRedisConnection {
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisPubSubCommands#subscribe(org.springframework.data.redis.connection.MessageListener, byte[][])
*/
@Override
public void subscribe(MessageListener listener, byte[]... channels) {
checkSubscription();
if (isPipelined()) {
throw new UnsupportedOperationException();
if (isQueueing() || isPipelined()) {
throw new UnsupportedOperationException("Transaction/Pipelining is not supported for Pub/Sub subscriptions!");
}
try {
subscription = new LettuceSubscription(listener, switchToPubSub(), connectionProvider);
subscription.subscribe(channels);
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}

27
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceMessageListener.java
View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection.lettuce;
import io.lettuce.core.pubsub.RedisPubSubListener;
@@ -32,23 +31,47 @@ class LettuceMessageListener implements RedisPubSubListener<byte[], byte[]> {
private final MessageListener listener;
LettuceMessageListener(MessageListener listener) {
Assert.notNull(listener, "message listener is required");
Assert.notNull(listener, "MessageListener must not be null!");
this.listener = listener;
}
/*
* (non-Javadoc)
* @see io.lettuce.core.pubsub.RedisPubSubListener#message(java.lang.Object, java.lang.Object)
*/
public void message(byte[] channel, byte[] message) {
listener.onMessage(new DefaultMessage(channel, message), null);
}
/*
* (non-Javadoc)
* @see io.lettuce.core.pubsub.RedisPubSubListener#message(java.lang.Object, java.lang.Object, java.lang.Object)
*/
public void message(byte[] pattern, byte[] channel, byte[] message) {
listener.onMessage(new DefaultMessage(channel, message), pattern);
}
/*
* (non-Javadoc)
* @see io.lettuce.core.pubsub.RedisPubSubListener#subscribed(java.lang.Object, long)
*/
public void subscribed(byte[] channel, long count) {}
/*
* (non-Javadoc)
* @see io.lettuce.core.pubsub.RedisPubSubListener#psubscribed(java.lang.Object, long)
*/
public void psubscribed(byte[] pattern, long count) {}
/*
* (non-Javadoc)
* @see io.lettuce.core.pubsub.RedisPubSubListener#unsubscribed(java.lang.Object, long)
*/
public void unsubscribed(byte[] channel, long count) {}
/*
* (non-Javadoc)
* @see io.lettuce.core.pubsub.RedisPubSubListener#punsubscribed(java.lang.Object, long)
*/
public void punsubscribed(byte[] pattern, long count) {}
}

2
src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java
View File

@@ -99,7 +99,7 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red
}
throw new IllegalStateException(
String.format("Undyerlying connection provider %s does not implement RedisClientProvider!",
String.format("Underlying connection provider %s does not implement RedisClientProvider!",
connectionProvider.getClass().getName()));
}

66
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java
View File

@@ -13,10 +13,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection.lettuce;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.util.AbstractSubscription;
@@ -29,46 +29,78 @@ import org.springframework.data.redis.connection.util.AbstractSubscription;
*/
class LettuceSubscription extends AbstractSubscription {
final StatefulRedisPubSubConnection<byte[], byte[]> pubsub;
private LettuceMessageListener listener;
private final StatefulRedisPubSubConnection<byte[], byte[]> connection;
private final LettuceMessageListener listener;
private final LettuceConnectionProvider connectionProvider;
private final RedisPubSubCommands<byte[], byte[]> pubsub;
LettuceSubscription(MessageListener listener, StatefulRedisPubSubConnection<byte[], byte[]> pubsubConnection,
LettuceConnectionProvider connectionProvider) {
LettuceSubscription(MessageListener listener, StatefulRedisPubSubConnection<byte[], byte[]> pubsubConnection, LettuceConnectionProvider connectionProvider) {
super(listener);
this.pubsub = pubsubConnection;
this.connection = pubsubConnection;
this.listener = new LettuceMessageListener(listener);
this.connectionProvider = connectionProvider;
this.pubsub = connection.sync();
pubsub.addListener(this.listener);
this.connection.addListener(this.listener);
}
protected StatefulRedisPubSubConnection<byte[], byte[]> getNativeConnection() {
return connection;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doClose()
*/
protected void doClose() {
if (!getChannels().isEmpty()) {
pubsub.sync().unsubscribe(new byte[0]);
pubsub.unsubscribe(new byte[0]);
}
if (!getPatterns().isEmpty()) {
pubsub.sync().punsubscribe(new byte[0]);
pubsub.punsubscribe(new byte[0]);
}
pubsub.removeListener(this.listener);
connectionProvider.release(pubsub);
connection.removeListener(this.listener);
connectionProvider.release(connection);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doPsubscribe(byte[][])
*/
protected void doPsubscribe(byte[]... patterns) {
pubsub.sync().psubscribe(patterns);
pubsub.psubscribe(patterns);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doPUnsubscribe(boolean, byte[][])
*/
protected void doPUnsubscribe(boolean all, byte[]... patterns) {
// lettuce doesn't automatically subscribe from all channels
pubsub.sync().punsubscribe(patterns);
// lettuce doesn't automatically unsubscribe from all patterns
pubsub.punsubscribe(patterns);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doSubscribe(byte[][])
*/
protected void doSubscribe(byte[]... channels) {
pubsub.sync().subscribe(channels);
pubsub.subscribe(channels);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doUnsubscribe(boolean, byte[][])
*/
protected void doUnsubscribe(boolean all, byte[]... channels) {
// lettuce doesn't automatically subscribe from all patterns
pubsub.sync().unsubscribe(channels);
// lettuce doesn't automatically unsubscribe from all channels
pubsub.unsubscribe(channels);
}
}