diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java index 41f1c82e5..7d184e2cd 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java @@ -29,8 +29,8 @@ import org.springframework.data.keyvalue.UncategorizedKeyvalueStoreException; import org.springframework.data.keyvalue.redis.connection.DataType; import org.springframework.data.keyvalue.redis.connection.MessageListener; import org.springframework.data.keyvalue.redis.connection.RedisConnection; -import org.springframework.data.keyvalue.redis.connection.SortParameters; import org.springframework.data.keyvalue.redis.connection.RedisSubscribedConnectionException; +import org.springframework.data.keyvalue.redis.connection.SortParameters; import org.springframework.data.keyvalue.redis.connection.Subscription; import org.springframework.util.ReflectionUtils; @@ -2221,6 +2221,7 @@ public class JedisConnection implements RedisConnection { subscription = new JedisSubscription(listener, jedisPubSub, null, patterns); jedis.psubscribe(jedisPubSub, patterns); + } catch (Exception ex) { throw convertJedisAccessException(ex); } @@ -2245,6 +2246,7 @@ public class JedisConnection implements RedisConnection { subscription = new JedisSubscription(listener, jedisPubSub, channels, null); jedis.subscribe(jedisPubSub, channels); + } catch (Exception ex) { throw convertJedisAccessException(ex); } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisSubscription.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisSubscription.java index 93dfbeeef..a2be1371a 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisSubscription.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisSubscription.java @@ -15,13 +15,8 @@ */ package org.springframework.data.keyvalue.redis.connection.jedis; -import java.util.ArrayList; -import java.util.Collection; - import org.springframework.data.keyvalue.redis.connection.MessageListener; -import org.springframework.data.keyvalue.redis.connection.Subscription; -import org.springframework.util.Assert; -import org.springframework.util.ObjectUtils; +import org.springframework.data.keyvalue.redis.connection.util.AbstractSubscription; import redis.clients.jedis.BinaryJedisPubSub; @@ -30,132 +25,48 @@ import redis.clients.jedis.BinaryJedisPubSub; * * @author Costin Leau */ -class JedisSubscription implements Subscription { +class JedisSubscription extends AbstractSubscription { - private final MessageListener listener; private final BinaryJedisPubSub jedisPubSub; - private final Collection channels = new ArrayList(2); - private final Collection patterns = new ArrayList(2); - JedisSubscription(MessageListener listener, BinaryJedisPubSub jedisPubSub, byte[][] channels, byte[][] patterns) { - Assert.notNull(listener); - this.listener = listener; + super(listener, channels, patterns); this.jedisPubSub = jedisPubSub; - - if (!ObjectUtils.isEmpty(channels)) { - synchronized (this.channels) { - for (byte[] bs : channels) { - this.channels.add(bs); - } - } - } - - if (!ObjectUtils.isEmpty(patterns)) { - synchronized (this.patterns) { - for (byte[] bs : patterns) { - this.patterns.add(bs); - } - } - } } @Override - public Collection getChannels() { - synchronized (channels) { - return new ArrayList(channels); - } - } - - @Override - public MessageListener getListener() { - return listener; - } - - @Override - public Collection getPatterns() { - synchronized (patterns) { - return new ArrayList(patterns); - } - } - - @Override - public void pSubscribe(byte[]... patterns) { - Assert.notEmpty(patterns, "at least one pattern required"); - - synchronized (this.patterns) { - for (byte[] bs : patterns) { - this.patterns.add(bs); - } - } - - jedisPubSub.psubscribe(patterns); - } - - @Override - public void pUnsubscribe() { - synchronized (patterns) { - patterns.clear(); - } + protected void doClose() { + jedisPubSub.unsubscribe(); jedisPubSub.punsubscribe(); } @Override - public void pUnsubscribe(byte[]... patterns) { - if (ObjectUtils.isEmpty(patterns)) { - unsubscribe(); + protected void doPsubscribe(byte[]... patterns) { + jedisPubSub.psubscribe(patterns); + } + + @Override + protected void doPUnsubscribe(boolean all, byte[]... patterns) { + if (all) { + jedisPubSub.punsubscribe(); } - else { - synchronized (this.patterns) { - for (byte[] bs : patterns) { - this.patterns.remove(bs); - } - } - jedisPubSub.punsubscribe(patterns); } } @Override - public void subscribe(byte[]... channels) { - Assert.notEmpty(channels, "at least one channel required"); - - synchronized (this.channels) { - for (byte[] bs : channels) { - this.channels.add(bs); - } - } - + protected void doSubscribe(byte[]... channels) { jedisPubSub.subscribe(channels); } @Override - public void unsubscribe() { - synchronized (channels) { - channels.clear(); - } - jedisPubSub.unsubscribe(); - } - - @Override - public void unsubscribe(byte[]... channels) { - if (ObjectUtils.isEmpty(channels)) { - unsubscribe(); + protected void doUnsubscribe(boolean all, byte[]... channels) { + if (all) { + jedisPubSub.unsubscribe(); } else { - synchronized (this.channels) { - for (byte[] bs : channels) { - this.channels.remove(bs); - } - } - jedisPubSub.unsubscribe(channels); } } - - @Override - public boolean isAlive() { - return jedisPubSub.isSubscribed(); - } } \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcSubscription.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcSubscription.java index a1075a120..fa0aa5a1d 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcSubscription.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcSubscription.java @@ -15,135 +15,58 @@ */ package org.springframework.data.keyvalue.redis.connection.rjc; -import java.util.ArrayList; -import java.util.Collection; - -import org.idevlab.rjc.message.RedisSubscriber; +import org.idevlab.rjc.message.RedisNodeSubscriber; import org.springframework.data.keyvalue.redis.connection.MessageListener; -import org.springframework.data.keyvalue.redis.connection.Subscription; -import org.springframework.util.Assert; -import org.springframework.util.ObjectUtils; +import org.springframework.data.keyvalue.redis.connection.util.AbstractSubscription; /** * Message subscription on top of RJC. * * @author Costin Leau */ -class RjcSubscription implements Subscription { +class RjcSubscription extends AbstractSubscription { - private final MessageListener listener; - private final RedisSubscriber subscriber; + private final RedisNodeSubscriber subscriber; private final RjcMessageListener listenerAdapter; + private final Object pubSubMonitor; - private final Collection channels = new ArrayList(2); - private final Collection patterns = new ArrayList(2); - - RjcSubscription(MessageListener listener, RedisSubscriber subscriber) { - Assert.notNull(listener); - this.listener = listener; + RjcSubscription(MessageListener listener, RedisNodeSubscriber subscriber, Object pubSubMonitor) { + super(listener); this.subscriber = subscriber; this.listenerAdapter = new RjcMessageListener(listener); + this.pubSubMonitor = pubSubMonitor; } @Override - public Collection getChannels() { - synchronized (channels) { - return new ArrayList(channels); + protected void doClose() { + subscriber.close(); + } + + @Override + protected void doPsubscribe(byte[]... patterns) { + for (String str : RjcUtils.decodeMultiple(patterns)) { + subscriber.psubscribe(str, listenerAdapter); } } @Override - public MessageListener getListener() { - return listener; - } - - @Override - public Collection getPatterns() { - synchronized (patterns) { - return new ArrayList(patterns); + protected void doPUnsubscribe(boolean all, byte[]... patterns) { + for (String str : RjcUtils.decodeMultiple(patterns)) { + subscriber.punsubscribe(str); } } @Override - public void pSubscribe(byte[]... patterns) { - Assert.notEmpty(patterns, "at least one pattern required"); - - synchronized (this.patterns) { - for (byte[] bs : patterns) { - this.patterns.add(bs); - } - } - - for (String pattern : RjcUtils.decodeMultiple(patterns)) { - subscriber.psubscribe(pattern, listenerAdapter); + protected void doSubscribe(byte[]... channels) { + for (String str : RjcUtils.decodeMultiple(channels)) { + subscriber.subscribe(str, listenerAdapter); } } @Override - public void pUnsubscribe() { - pUnsubscribe(null); - - synchronized (patterns) { - patterns.clear(); + protected void doUnsubscribe(boolean all, byte[]... channels) { + for (String str : RjcUtils.decodeMultiple(channels)) { + subscriber.unsubscribe(str); } } - - @Override - public void pUnsubscribe(byte[]... patterns) { - if (ObjectUtils.isEmpty(patterns)) { - patterns = this.patterns.toArray(new byte[this.patterns.size()][]); - } - - synchronized (this.patterns) { - for (byte[] bs : patterns) { - this.patterns.remove(bs); - } - } - - subscriber.punsubscribe(RjcUtils.decodeMultiple(patterns)); - } - - @Override - public void subscribe(byte[]... channels) { - Assert.notEmpty(channels, "at least one channel required"); - - synchronized (this.channels) { - for (byte[] bs : channels) { - this.channels.add(bs); - } - } - - for (String channel : RjcUtils.decodeMultiple(channels)) { - subscriber.subscribe(channel, listenerAdapter); - } - } - - @Override - public void unsubscribe() { - unsubscribe(null); - - synchronized (patterns) { - patterns.clear(); - } - } - - @Override - public void unsubscribe(byte[]... channels) { - if (ObjectUtils.isEmpty(channels)) { - channels = this.channels.toArray(new byte[this.channels.size()][]); - } - - synchronized (this.channels) { - for (byte[] bs : channels) { - this.channels.remove(bs); - } - } - - subscriber.punsubscribe(RjcUtils.decodeMultiple(channels)); - } - - @Override - public boolean isAlive() { - return (!channels.isEmpty() || !patterns.isEmpty()); - } } \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/AbstractSubscription.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/AbstractSubscription.java new file mode 100644 index 000000000..76dfbfec1 --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/AbstractSubscription.java @@ -0,0 +1,253 @@ +/* + * Copyright 2011 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.keyvalue.redis.connection.util; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.springframework.data.keyvalue.redis.connection.MessageListener; +import org.springframework.data.keyvalue.redis.connection.RedisInvalidSubscriptionException; +import org.springframework.data.keyvalue.redis.connection.Subscription; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; + +/** + * Base implementation for a subscription handling the channel/pattern registration so subclasses only have to deal + * with the actual registration/unregistration. + * + * @author Costin Leau + */ +public abstract class AbstractSubscription implements Subscription { + + private final Collection channels = new ArrayList(2); + private final Collection patterns = new ArrayList(2); + private final AtomicBoolean alive = new AtomicBoolean(true); + private final MessageListener listener; + + protected AbstractSubscription(MessageListener listener) { + this(listener, null, null); + } + + /** + * Constructs a new AbstractSubscription instance. Allows channels and patterns to be added + * to the subscription w/o triggering a subscription action (as some clients (Jedis) require an initial call + * before entering into listening mode). + * + * @param listener + * @param channels + * @param patterns + */ + protected AbstractSubscription(MessageListener listener, byte[][] channels, byte[][] patterns) { + Assert.notNull(listener); + this.listener = listener; + + synchronized (this.channels) { + remove(this.channels, channels); + } + synchronized (this.patterns) { + remove(this.patterns, patterns); + } + } + + /** + * Subscribe to the given channels. + * + * @param channels channels to subscribe to + */ + protected abstract void doSubscribe(byte[]... channels); + + /** + * Channel unsubscribe. + * + * @param all true if all the channels are unsubscribed (used as a hint for the underlying implementation). + * @param channels channels to be unsubscribed + */ + protected abstract void doUnsubscribe(boolean all, byte[]... channels); + + /** + * Subscribe to the given patterns + * + * @param patterns patterns to subscribe to + */ + protected abstract void doPsubscribe(byte[]... patterns); + + /** + * Pattern unsubscribe. + * + * @param all true if all the patterns are unsubscribed (used as a hint for the underlying implementation). + * @param patterns patterns to be unsubscribed + */ + protected abstract void doPUnsubscribe(boolean all, byte[]... patterns); + + /** + * Shutdown the subscription and free any resources held. + */ + protected abstract void doClose(); + + @Override + public MessageListener getListener() { + return listener; + } + + @Override + public Collection getChannels() { + synchronized (channels) { + return clone(channels); + } + } + + @Override + public Collection getPatterns() { + synchronized (patterns) { + return clone(patterns); + } + } + + @Override + public void pSubscribe(byte[]... patterns) { + checkPulse(); + + Assert.notEmpty(patterns, "at least one pattern required"); + + synchronized (this.patterns) { + add(this.patterns, patterns); + } + + doPsubscribe(patterns); + } + + @Override + public void pUnsubscribe() { + pUnsubscribe((byte[][]) null); + } + + + @Override + public void subscribe(byte[]... channels) { + checkPulse(); + + Assert.notEmpty(channels, "at least one channel required"); + + synchronized (this.channels) { + add(this.channels, channels); + } + + doSubscribe(channels); + } + + @Override + public void unsubscribe() { + unsubscribe((byte[][]) null); + } + + @Override + public void pUnsubscribe(byte[]... patts) { + if (!isAlive()) { + return; + } + + // shortcut for unsubscribing all patterns + if (ObjectUtils.isEmpty(patts)) { + if (!this.patterns.isEmpty()) { + patts = getPatterns().toArray(new byte[this.patterns.size()][]); + synchronized (this.patterns) { + this.patterns.clear(); + } + } + } + else { + synchronized (this.patterns) { + remove(this.patterns, patts); + } + } + + if (isWorking()) { + doPUnsubscribe(this.patterns.isEmpty(), patts); + } + } + + @Override + public void unsubscribe(byte[]... chans) { + if (!isAlive()) { + return; + } + + // shortcut for unsubscribing all channels + if (ObjectUtils.isEmpty(chans)) { + if (!this.channels.isEmpty()) { + chans = getPatterns().toArray(new byte[this.channels.size()][]); + synchronized (this.channels) { + this.channels.clear(); + } + } + } + else { + synchronized (this.channels) { + remove(this.channels, chans); + } + } + + if (isWorking()) { + doUnsubscribe(this.channels.isEmpty(), chans); + } + } + + @Override + public boolean isAlive() { + return alive.get(); + } + + private void checkPulse() { + if (!isAlive()) { + throw new RedisInvalidSubscriptionException("Subscription has been unsubscribed and cannot be used anymore"); + } + } + + private boolean isWorking() { + if (channels.isEmpty() && patterns.isEmpty()) { + alive.set(false); + doClose(); + } + return isAlive(); + } + + + private static Collection clone(Collection col) { + Collection list = new ArrayList(col.size()); + for (ByteArrayWrapper wrapper : col) { + list.add(wrapper.getArray().clone()); + } + return list; + } + + + private static void add(Collection col, byte[]... bytes) { + if (!ObjectUtils.isEmpty(bytes)) { + for (byte[] bs : bytes) { + col.add(new ByteArrayWrapper(bs)); + } + } + } + + private static void remove(Collection col, byte[]... bytes) { + if (!ObjectUtils.isEmpty(bytes)) { + for (byte[] bs : bytes) { + col.remove(new ByteArrayWrapper(bs)); + } + } + } +} \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/ByteArrayWrapper.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/ByteArrayWrapper.java new file mode 100644 index 000000000..7c708192b --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/ByteArrayWrapper.java @@ -0,0 +1,57 @@ +/* + * Copyright 2011 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.keyvalue.redis.connection.util; + +import java.util.Arrays; + +/** + * Simple wrapper class used for wrapping arrays so they can be used as keys inside maps. + * + * @author Costin Leau + */ +public class ByteArrayWrapper { + + private final byte[] array; + private final int hashCode; + + public ByteArrayWrapper(byte[] array) { + this.array = array; + this.hashCode = Arrays.hashCode(array); + } + + @Override + public boolean equals(Object obj) { + if (obj instanceof ByteArrayWrapper) { + return Arrays.equals(array, ((ByteArrayWrapper) obj).array); + } + + return false; + } + + @Override + public int hashCode() { + return hashCode; + } + + /** + * Returns the array. + * + * @return Returns the array + */ + public byte[] getArray() { + return array; + } +} diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/package-info.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/package-info.java new file mode 100644 index 000000000..c8f07d8ce --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/package-info.java @@ -0,0 +1,5 @@ +/** + * Internal utility package for encoding/decoding Strings to byte[] (using Base64) library. + */ +package org.springframework.data.keyvalue.redis.connection.util; + diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/RedisMessageListenerContainer.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/RedisMessageListenerContainer.java index 1c9676e4c..0691b8363 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/RedisMessageListenerContainer.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/RedisMessageListenerContainer.java @@ -16,7 +16,6 @@ package org.springframework.data.keyvalue.redis.listener; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.List; @@ -39,6 +38,7 @@ import org.springframework.data.keyvalue.redis.connection.MessageListener; import org.springframework.data.keyvalue.redis.connection.RedisConnection; import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory; import org.springframework.data.keyvalue.redis.connection.Subscription; +import org.springframework.data.keyvalue.redis.connection.util.ByteArrayWrapper; import org.springframework.data.keyvalue.redis.serializer.RedisSerializer; import org.springframework.data.keyvalue.redis.serializer.StringRedisSerializer; import org.springframework.scheduling.SchedulingAwareRunnable; @@ -101,9 +101,9 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab // to avoid creation of hashes for each message, the maps use raw byte arrays (wrapped to respect the equals/hashcode contract) // lookup map between patterns and listeners - private final Map> patternMapping = new ConcurrentHashMap>(); + private final Map> patternMapping = new ConcurrentHashMap>(); // lookup map between channels and listeners - private final Map> channelMapping = new ConcurrentHashMap>(); + private final Map> channelMapping = new ConcurrentHashMap>(); private final SubscriptionTask subscriptionTask = new SubscriptionTask(); @@ -448,7 +448,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab for (Topic topic : topics) { - ArrayHolder holder = new ArrayHolder(serializer.serialize(topic.getTopic())); + ByteArrayWrapper holder = new ByteArrayWrapper(serializer.serialize(topic.getTopic())); if (topic instanceof ChannelTopic) { Collection collection = channelMapping.get(holder); @@ -457,7 +457,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab channelMapping.put(holder, collection); } collection.add(listener); - channels.add(holder.array); + channels.add(holder.getArray()); if (trace) logger.trace("Adding listener '" + listener + "' on channel '" + topic.getTopic() + "'"); @@ -470,7 +470,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab patternMapping.put(holder, collection); } collection.add(listener); - patterns.add(holder.array); + patterns.add(holder.getArray()); if (trace) logger.trace("Adding listener '" + listener + "' for pattern '" + topic.getTopic() + "'"); @@ -598,7 +598,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } } - private byte[][] unwrap(Collection holders) { + private byte[][] unwrap(Collection holders) { if (CollectionUtils.isEmpty(holders)) { return new byte[0][]; } @@ -606,8 +606,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab byte[][] unwrapped = new byte[holders.size()][]; int index = 0; - for (ArrayHolder arrayHolder : holders) { - unwrapped[index++] = arrayHolder.array; + for (ByteArrayWrapper arrayHolder : holders) { + unwrapped[index++] = arrayHolder.getArray(); } return unwrapped; @@ -700,12 +700,12 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab // do channel matching first byte[] channel = message.getChannel(); - Collection ch = channelMapping.get(new ArrayHolder(channel)); + Collection ch = channelMapping.get(new ByteArrayWrapper(channel)); Collection pt = null; // followed by pattern matching if (pattern != null && pattern.length > 0) { - pt = patternMapping.get(new ArrayHolder(pattern)); + pt = patternMapping.get(new ByteArrayWrapper(pattern)); } if (!CollectionUtils.isEmpty(ch)) { @@ -739,34 +739,4 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } } } - - /** - * Simple wrapper class used for wrapping arrays so they can be used as keys inside maps. - * - * @author Costin Leau - */ - private class ArrayHolder { - - private final byte[] array; - private final int hashCode; - - ArrayHolder(byte[] array) { - this.array = array; - this.hashCode = Arrays.hashCode(array); - } - - @Override - public boolean equals(Object obj) { - if (obj instanceof ArrayHolder) { - return Arrays.equals(array, ((ArrayHolder) obj).array); - } - - return false; - } - - @Override - public int hashCode() { - return hashCode; - } - } } \ No newline at end of file