From d46bee7e5769ba6abf5969a5b00ceaae4a78115d Mon Sep 17 00:00:00 2001 From: Jennifer Hickey Date: Wed, 6 Mar 2013 09:37:57 -0800 Subject: [PATCH] Revert "convertAndSend returns the nr of targeted channels" - DATAREDIS-118 - Change method return type back to void to fix backwards compatibility issues - Modify tests to assert specific clients receive or don't receive messages vs comparing expected client counts --- .../data/redis/core/RedisOperations.java | 2 +- .../data/redis/core/RedisTemplate.java | 9 ++- .../listener/PubSubResubscribeTests.java | 73 +++++++++++-------- .../data/redis/listener/PubSubTests.java | 20 ++--- 4 files changed, 56 insertions(+), 48 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/core/RedisOperations.java b/src/main/java/org/springframework/data/redis/core/RedisOperations.java index 72056c07e..a5fe3dbe3 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisOperations.java +++ b/src/main/java/org/springframework/data/redis/core/RedisOperations.java @@ -117,7 +117,7 @@ public interface RedisOperations { List exec(); // pubsub functionality on the template - Long convertAndSend(String destination, Object message); + void convertAndSend(String destination, Object message); // operation types diff --git a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java index d3407988f..6354d2277 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java @@ -496,16 +496,17 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation } - public Long convertAndSend(String channel, Object message) { + public void convertAndSend(String channel, Object message) { Assert.hasText(channel, "a non-empty channel is required"); final byte[] rawChannel = rawString(channel); final byte[] rawMessage = rawValue(message); - return execute(new RedisCallback() { + execute(new RedisCallback() { - public Long doInRedis(RedisConnection connection) { - return connection.publish(rawChannel, rawMessage); + public Object doInRedis(RedisConnection connection) { + connection.publish(rawChannel, rawMessage); + return null; } }, true); } diff --git a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java index ad682b6cc..06418f65e 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java @@ -46,13 +46,11 @@ import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; /** * @author Costin Leau + * @author Jennifer Hickey */ public class PubSubResubscribeTests { private static final String CHANNEL = "pubsub::test"; - private final Long ZERO = Long.valueOf(0); - private final Long ONE = Long.valueOf(1); - private final Long TWO = Long.valueOf(2); protected RedisMessageListenerContainer container; protected RedisConnectionFactory factory; @@ -60,12 +58,7 @@ public class PubSubResubscribeTests { private final BlockingDeque bag = new LinkedBlockingDeque(99); - private final Object handler = new Object() { - public void handleMessage(String message) { - System.out.println(message); - bag.add(message); - } - }; + private final Object handler = new MessageHandler("handler1", bag); private final MessageListenerAdapter adapter = new MessageListenerAdapter(handler); @@ -121,7 +114,8 @@ public class PubSubResubscribeTests { final String PATTERN = "p*"; final String ANOTHER_CHANNEL = "pubsub::test::extra"; - MessageListenerAdapter anotherListener = new MessageListenerAdapter(handler); + BlockingDeque bag2 = new LinkedBlockingDeque(99); + MessageListenerAdapter anotherListener = new MessageListenerAdapter(new MessageHandler("handler2", bag2)); anotherListener.setSerializer(template.getValueSerializer()); anotherListener.afterPropertiesSet(); @@ -130,29 +124,36 @@ public class PubSubResubscribeTests { container.removeMessageListener(adapter); // test no messages are sent just to patterns - assertEquals(ONE, template.convertAndSend(CHANNEL, payload1)); - assertEquals(ONE, template.convertAndSend(ANOTHER_CHANNEL, payload2)); + template.convertAndSend(CHANNEL, payload1); + template.convertAndSend(ANOTHER_CHANNEL, payload2); + // anotherListener receives both messages List msgs = new ArrayList(); - msgs.add(bag.poll(1, TimeUnit.SECONDS)); - msgs.add(bag.poll(1, TimeUnit.SECONDS)); + msgs.add(bag2.poll(1, TimeUnit.SECONDS)); + msgs.add(bag2.poll(1, TimeUnit.SECONDS)); assertEquals(2, msgs.size()); + assertTrue(msgs.contains(payload1)); + assertTrue(msgs.contains(payload2)); + msgs.clear(); + + // unsubscribed adapter did not receive message + assertNull(bag.poll(1, TimeUnit.SECONDS)); + // bind original listener on another channel container.addMessageListener(adapter, new ChannelTopic(ANOTHER_CHANNEL)); - assertEquals(ONE, template.convertAndSend(CHANNEL, payload1)); - assertEquals(TWO, template.convertAndSend(ANOTHER_CHANNEL, payload2)); + template.convertAndSend(CHANNEL, payload1); + template.convertAndSend(ANOTHER_CHANNEL, payload2); - msgs.add(bag.poll(1, TimeUnit.SECONDS)); - msgs.add(bag.poll(1, TimeUnit.SECONDS)); - msgs.add(bag.poll(1, TimeUnit.SECONDS)); - // this message will not arrive on time + // original listener received only one message on another channel + assertEquals(payload2,bag.poll(1, TimeUnit.SECONDS)); assertNull(bag.poll(1, TimeUnit.SECONDS)); - // same message received first per channel subscription, second based on the pattern - assertEquals(5, msgs.size()); - + //another listener receives messages on both channels + msgs.add(bag2.poll(1, TimeUnit.SECONDS)); + msgs.add(bag2.poll(1, TimeUnit.SECONDS)); + assertEquals(2, msgs.size()); assertTrue(msgs.contains(payload1)); assertTrue(msgs.contains(payload2)); } @@ -171,22 +172,36 @@ public class PubSubResubscribeTests { container.addMessageListener(adapter, new ChannelTopic(ANOTHER_CHANNEL)); container.removeMessageListener(null, new ChannelTopic(CHANNEL)); - assertEquals(ZERO, template.convertAndSend(CHANNEL, payload1)); - assertEquals(ZERO, template.convertAndSend(CHANNEL, payload2)); + // Listener removed from channel + template.convertAndSend(CHANNEL, payload1); + template.convertAndSend(CHANNEL, payload2); - assertEquals(ONE, template.convertAndSend(ANOTHER_CHANNEL, anotherPayload1)); - assertEquals(ONE, template.convertAndSend(ANOTHER_CHANNEL, anotherPayload2)); + // Listener receives messages on another channel + template.convertAndSend(ANOTHER_CHANNEL, anotherPayload1); + template.convertAndSend(ANOTHER_CHANNEL, anotherPayload2); Set set = new LinkedHashSet(); set.add(bag.poll(1, TimeUnit.SECONDS)); set.add(bag.poll(1, TimeUnit.SECONDS)); - System.out.println(set); - assertFalse(set.contains(payload1)); assertFalse(set.contains(payload2)); assertTrue(set.contains(anotherPayload1)); assertTrue(set.contains(anotherPayload2)); } + + private class MessageHandler { + private final BlockingDeque bag; + private final String name; + + public MessageHandler(String name, BlockingDeque bag) { + this.bag = bag; + this.name = name; + } + public void handleMessage(String message) { + System.out.println(name + ": " + message); + bag.add(message); + } + } } \ No newline at end of file diff --git a/src/test/java/org/springframework/data/redis/listener/PubSubTests.java b/src/test/java/org/springframework/data/redis/listener/PubSubTests.java index 65d658b75..23e79e285 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubTests.java @@ -16,8 +16,8 @@ package org.springframework.data.redis.listener; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.assertNull; import java.util.Arrays; import java.util.Collection; @@ -50,9 +50,6 @@ import org.springframework.data.redis.support.collections.ObjectFactory; public class PubSubTests { private static final String CHANNEL = "pubsub::test"; - private final Long ZERO = Long.valueOf(0); - private final Long ONE = Long.valueOf(1); - private final Long TWO = Long.valueOf(2); protected RedisMessageListenerContainer container; protected ObjectFactory factory; @@ -121,8 +118,8 @@ public class PubSubTests { String payload1 = "do"; String payload2 = "re mi"; - assertEquals(ONE, template.convertAndSend(CHANNEL, payload1)); - assertEquals(ONE, template.convertAndSend(CHANNEL, payload2)); + template.convertAndSend(CHANNEL, payload1); + template.convertAndSend(CHANNEL, payload2); Set set = new LinkedHashSet(); set.add(bag.poll(1, TimeUnit.SECONDS)); @@ -149,14 +146,9 @@ public class PubSubTests { String payload2 = "re mi"; container.removeMessageListener(adapter, new ChannelTopic(CHANNEL)); - assertEquals(ZERO, template.convertAndSend(CHANNEL, payload1)); - assertEquals(ZERO, template.convertAndSend(CHANNEL, payload2)); + template.convertAndSend(CHANNEL, payload1); + template.convertAndSend(CHANNEL, payload2); - Set set = new LinkedHashSet(); - set.add(bag.poll(1, TimeUnit.SECONDS)); - set.add(bag.poll(1, TimeUnit.SECONDS)); - - assertFalse(set.contains(payload1)); - assertFalse(set.contains(payload2)); + assertNull(bag.poll(1, TimeUnit.SECONDS)); } } \ No newline at end of file