DATAKV-22

+ more bug fixes

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisSubscription.java

@@ -44,21 +44,27 @@ class JedisSubscription implements Subscription {
 		this.jedisPubSub = jedisPubSub;
 
 		if (!ObjectUtils.isEmpty(channels)) {
-			for (byte[] bs : channels) {
-				this.channels.add(bs);
+			synchronized (this.channels) {
+				for (byte[] bs : channels) {
+					this.channels.add(bs);
+				}
 			}
 		}
 
 		if (!ObjectUtils.isEmpty(patterns)) {
-			for (byte[] bs : patterns) {
-				this.patterns.add(bs);
+			synchronized (this.patterns) {
+				for (byte[] bs : patterns) {
+					this.patterns.add(bs);
+				}
 			}
 		}
 	}
 
 	@Override
 	public Collection<byte[]> getChannels() {
-		return channels;
+		synchronized (channels) {
+			return new ArrayList<byte[]>(channels);
+		}
 	}
 
 	@Override
@@ -68,15 +74,19 @@ class JedisSubscription implements Subscription {
 
 	@Override
 	public Collection<byte[]> getPatterns() {
-		return patterns;
+		synchronized (patterns) {
+			return new ArrayList<byte[]>(patterns);
+		}
 	}
 
 	@Override
 	public void pSubscribe(byte[]... patterns) {
 		Assert.notEmpty(patterns, "at least one pattern required");
 
-		for (byte[] bs : patterns) {
-			this.patterns.add(bs);
+		synchronized (this.patterns) {
+			for (byte[] bs : patterns) {
+				this.patterns.add(bs);
+			}
 		}
 
 		jedisPubSub.psubscribe(JedisUtils.convert(patterns));
@@ -84,6 +94,9 @@ class JedisSubscription implements Subscription {
 
 	@Override
 	public void pUnsubscribe() {
+		synchronized (patterns) {
+			patterns.clear();
+		}
 		jedisPubSub.punsubscribe();
 	}
 
@@ -94,8 +107,10 @@ class JedisSubscription implements Subscription {
 		}
 
 		else {
-			for (byte[] bs : patterns) {
-				this.patterns.remove(bs);
+			synchronized (this.patterns) {
+				for (byte[] bs : patterns) {
+					this.patterns.remove(bs);
+				}
 			}
 
 			jedisPubSub.punsubscribe(JedisUtils.convert(patterns));
@@ -106,8 +121,10 @@ class JedisSubscription implements Subscription {
 	public void subscribe(byte[]... channels) {
 		Assert.notEmpty(patterns, "at least one pattern required");
 
-		for (byte[] bs : patterns) {
-			this.patterns.add(bs);
+		synchronized (this.channels) {
+			for (byte[] bs : channels) {
+				this.channels.add(bs);
+			}
 		}
 
 		jedisPubSub.subscribe(JedisUtils.convert(channels));
@@ -115,6 +132,9 @@ class JedisSubscription implements Subscription {
 
 	@Override
 	public void unsubscribe() {
+		synchronized (channels) {
+			channels.clear();
+		}
 		jedisPubSub.unsubscribe();
 	}
 
@@ -124,8 +144,10 @@ class JedisSubscription implements Subscription {
 			unsubscribe();
 		}
 		else {
-			for (byte[] bs : patterns) {
-				this.patterns.remove(bs);
+			synchronized (this.channels) {
+				for (byte[] bs : channels) {
+					this.channels.remove(bs);
+				}
 			}
 
 			jedisPubSub.unsubscribe(JedisUtils.convert(channels));

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnectionIntegrationTests.java

@@ -44,7 +44,7 @@ public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrati
 	}
 
 	@Test
-	public void testPubSub() {
+	public void testPubSubWithNamedChannels() {
 		final byte[] expectedChannel = "channel1".getBytes();
 		final byte[] expectedMessage = "msg".getBytes();
 
@@ -81,6 +81,45 @@ public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrati
 		connection.subscribe(listener, expectedChannel);
 	}
 
+	@Test
+	public void testPubSubWithPatterns() {
+		final byte[] expectedPattern = "channel*".getBytes();
+		final byte[] expectedMessage = "msg".getBytes();
+
+		MessageListener listener = new MessageListener() {
+
+			@Override
+			public void onMessage(byte[] message, byte[] channel, byte[] pattern) {
+				assertArrayEquals(expectedPattern, pattern);
+				assertArrayEquals(expectedMessage, message);
+				System.out.println("Received message '" + new String(message) + "'");
+			}
+		};
+
+		Thread th = new Thread(new Runnable() {
+			@Override
+			public void run() {
+				// sleep 1 second to let the registration happen
+				try {
+					Thread.currentThread().sleep(1000);
+				} catch (InterruptedException ex) {
+					throw new RuntimeException(ex);
+				}
+
+				// open a new connection
+				JedisConnection connection2 = factory.getConnection();
+				connection2.publish(expectedMessage, "channel1".getBytes());
+				connection2.publish(expectedMessage, "channel2".getBytes());
+				connection2.close();
+				// unsubscribe connection
+				connection.getSubscription().pUnsubscribe(expectedPattern);
+			}
+		});
+
+		th.start();
+		connection.pSubscribe(listener, expectedPattern);
+	}
+
 //	@Test
 //	public void setAdd() {
 //		connection.sadd("s1", "1");