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 6f851f411..bfdda40dd 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 @@ -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 getChannels() { - return channels; + synchronized (channels) { + return new ArrayList(channels); + } } @Override @@ -68,15 +74,19 @@ class JedisSubscription implements Subscription { @Override public Collection getPatterns() { - return patterns; + synchronized (patterns) { + return new ArrayList(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)); diff --git a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnectionIntegrationTests.java b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnectionIntegrationTests.java index e8691a8b5..90955d3c8 100644 --- a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnectionIntegrationTests.java +++ b/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");