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

Author: Jennifer Hickey

src/main/java/org/springframework/data/redis/core/RedisOperations.java

@@ -117,7 +117,7 @@ public interface RedisOperations<K, V> {
 	List<Object> exec();
 
 	// pubsub functionality on the template
-	Long convertAndSend(String destination, Object message);
+	void convertAndSend(String destination, Object message);
 
 
 	// operation types

src/main/java/org/springframework/data/redis/core/RedisTemplate.java

@@ -496,16 +496,17 @@ public Boolean doInRedis(RedisConnection connection) {
 	}
 
 
-	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<Long>() {
+		execute(new RedisCallback<Object>() {
 
-			public Long doInRedis(RedisConnection connection) {
-				return connection.publish(rawChannel, rawMessage);
+			public Object doInRedis(RedisConnection connection) {
+				connection.publish(rawChannel, rawMessage);
+				return null;
 			}
 		}, true);
 	}

src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java

@@ -46,26 +46,19 @@
 
 /**
  * @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;
 	protected RedisTemplate template;
 
 	private final BlockingDeque<String> bag = new LinkedBlockingDeque<String>(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 void testContainerPatternResubscribe() throws Exception {
 		final String PATTERN = "p*";
 		final String ANOTHER_CHANNEL = "pubsub::test::extra";
 
-		MessageListenerAdapter anotherListener = new MessageListenerAdapter(handler);
+		BlockingDeque<String> bag2 = new LinkedBlockingDeque<String>(99);
+		MessageListenerAdapter anotherListener = new MessageListenerAdapter(new MessageHandler("handler2", bag2));
 		anotherListener.setSerializer(template.getValueSerializer());
 		anotherListener.afterPropertiesSet();
 
@@ -130,29 +124,36 @@ public void testContainerPatternResubscribe() throws Exception {
 		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<String> msgs = new ArrayList<String>();
-		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 void testContainerChannelResubscribe() throws Exception {
 		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<String> set = new LinkedHashSet<String>();
 		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<String> bag;
+		private final String name;
+
+		public MessageHandler(String name, BlockingDeque<String> bag) {
+			this.bag = bag;
+			this.name = name;
+		}
+		public void handleMessage(String message) {
+			System.out.println(name + ": " + message);
+			bag.add(message);
+		}
+	}
 }

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 @@
 public class PubSubTests<T> {
 
 	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<T> factory;
@@ -121,8 +118,8 @@ public void testContainerSubscribe() throws Exception {
 		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<String> set = new LinkedHashSet<String>();
 		set.add(bag.poll(1, TimeUnit.SECONDS));
@@ -149,14 +146,9 @@ public void testContainerUnsubscribe() throws Exception {
 		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<String> set = new LinkedHashSet<String>();
-		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));
 	}
 }