DATAKV-49

DATAKV-46
+ improved pubsub connection support
+ RJC integration still needs some work

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);
 		}

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<byte[]> channels = new ArrayList<byte[]>(2);
-	private final Collection<byte[]> patterns = new ArrayList<byte[]>(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<byte[]> getChannels() {
-		synchronized (channels) {
-			return new ArrayList<byte[]>(channels);
-		}
-	}
-
-	@Override
-	public MessageListener getListener() {
-		return listener;
-	}
-
-	@Override
-	public Collection<byte[]> getPatterns() {
-		synchronized (patterns) {
-			return new ArrayList<byte[]>(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();
-	}
 }

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<byte[]> channels = new ArrayList<byte[]>(2);
-	private final Collection<byte[]> patterns = new ArrayList<byte[]>(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<byte[]> getChannels() {
-		synchronized (channels) {
-			return new ArrayList<byte[]>(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<byte[]> getPatterns() {
-		synchronized (patterns) {
-			return new ArrayList<byte[]>(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());
-	}
 }

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<ByteArrayWrapper> channels = new ArrayList<ByteArrayWrapper>(2);
+	private final Collection<ByteArrayWrapper> patterns = new ArrayList<ByteArrayWrapper>(2);
+	private final AtomicBoolean alive = new AtomicBoolean(true);
+	private final MessageListener listener;
+
+	protected AbstractSubscription(MessageListener listener) {
+		this(listener, null, null);
+	}
+
+	/**
+	 * Constructs a new <code>AbstractSubscription</code> 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<byte[]> getChannels() {
+		synchronized (channels) {
+			return clone(channels);
+		}
+	}
+
+	@Override
+	public Collection<byte[]> 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<byte[]> clone(Collection<ByteArrayWrapper> col) {
+		Collection<byte[]> list = new ArrayList<byte[]>(col.size());
+		for (ByteArrayWrapper wrapper : col) {
+			list.add(wrapper.getArray().clone());
+		}
+		return list;
+	}
+
+
+	private static void add(Collection<ByteArrayWrapper> col, byte[]... bytes) {
+		if (!ObjectUtils.isEmpty(bytes)) {
+			for (byte[] bs : bytes) {
+				col.add(new ByteArrayWrapper(bs));
+			}
+		}
+	}
+
+	private static void remove(Collection<ByteArrayWrapper> col, byte[]... bytes) {
+		if (!ObjectUtils.isEmpty(bytes)) {
+			for (byte[] bs : bytes) {
+				col.remove(new ByteArrayWrapper(bs));
+			}
+		}
+	}
+}

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;
+	}
+}

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;
+

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<ArrayHolder, Collection<MessageListener>> patternMapping = new ConcurrentHashMap<ArrayHolder, Collection<MessageListener>>();
+	private final Map<ByteArrayWrapper, Collection<MessageListener>> patternMapping = new ConcurrentHashMap<ByteArrayWrapper, Collection<MessageListener>>();
 	// lookup map between channels and listeners
-	private final Map<ArrayHolder, Collection<MessageListener>> channelMapping = new ConcurrentHashMap<ArrayHolder, Collection<MessageListener>>();
+	private final Map<ByteArrayWrapper, Collection<MessageListener>> channelMapping = new ConcurrentHashMap<ByteArrayWrapper, Collection<MessageListener>>();
 
 	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<MessageListener> 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<ArrayHolder> holders) {
+		private byte[][] unwrap(Collection<ByteArrayWrapper> 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<MessageListener> ch = channelMapping.get(new ArrayHolder(channel));
+			Collection<MessageListener> ch = channelMapping.get(new ByteArrayWrapper(channel));
 			Collection<MessageListener> 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;
-		}
-	}
 }