Fix MessageListenerAdapter sending msg instead of body

when no Serializer is specified DATAREDIS-233

Author: jencompgeek

src/main/java/org/springframework/data/redis/listener/adapter/MessageListenerAdapter.java

@@ -372,7 +372,7 @@ protected Object extractMessage(Message message) {
 if (serializer != null) {
 return serializer.deserialize(message.getBody());
 }
-return message;
+return message.getBody();
 }
 
 /**

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

@@ -21,6 +21,7 @@
 import org.springframework.data.redis.ObjectFactory;
 import org.springframework.data.redis.Person;
 import org.springframework.data.redis.PersonObjectFactory;
+import org.springframework.data.redis.RawObjectFactory;
 import org.springframework.data.redis.SettingsUtils;
 import org.springframework.data.redis.StringObjectFactory;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
@@ -39,6 +40,7 @@ public static Collection<Object[]> testParams() {
 // create Jedis Factory
 ObjectFactory<String> stringFactory = new StringObjectFactory();
 ObjectFactory<Person> personFactory = new PersonObjectFactory();
+ObjectFactory<byte[]> rawFactory = new RawObjectFactory();
 
 JedisConnectionFactory jedisConnFactory = new JedisConnectionFactory();
 jedisConnFactory.setUsePool(true);
@@ -52,6 +54,10 @@ public static Collection<Object[]> testParams() {
 RedisTemplate<String, Person> personTemplate = new RedisTemplate<String, Person>();
 personTemplate.setConnectionFactory(jedisConnFactory);
 personTemplate.afterPropertiesSet();
+RedisTemplate<byte[], byte[]> rawTemplate = new RedisTemplate<byte[], byte[]>();
+rawTemplate.setEnableDefaultSerializer(false);
+rawTemplate.setConnectionFactory(jedisConnFactory);
+rawTemplate.afterPropertiesSet();
 
 // add Lettuce
 LettuceConnectionFactory lettuceConnFactory = new LettuceConnectionFactory();
@@ -63,6 +69,10 @@ public static Collection<Object[]> testParams() {
 RedisTemplate<String, Person> personTemplateLtc = new RedisTemplate<String, Person>();
 personTemplateLtc.setConnectionFactory(lettuceConnFactory);
 personTemplateLtc.afterPropertiesSet();
+RedisTemplate<byte[], byte[]> rawTemplateLtc = new RedisTemplate<byte[], byte[]>();
+rawTemplateLtc.setEnableDefaultSerializer(false);
+rawTemplateLtc.setConnectionFactory(lettuceConnFactory);
+rawTemplateLtc.afterPropertiesSet();
 
 // SRP
 SrpConnectionFactory srpConnFactory = new SrpConnectionFactory();
@@ -74,12 +84,17 @@ public static Collection<Object[]> testParams() {
 RedisTemplate<String, Person> personTemplateSrp = new RedisTemplate<String, Person>();
 personTemplateSrp.setConnectionFactory(srpConnFactory);
 personTemplateSrp.afterPropertiesSet();
+RedisTemplate<byte[], byte[]> rawTemplateSrp = new RedisTemplate<byte[], byte[]>();
+rawTemplateSrp.setEnableDefaultSerializer(false);
+rawTemplateSrp.setConnectionFactory(srpConnFactory);
+rawTemplateSrp.afterPropertiesSet();
 
 // JRedis does not support pub/sub
 
 return Arrays.asList(new Object[][] { { stringFactory, stringTemplate }, { personFactory, personTemplate },
-{ stringFactory, stringTemplateLtc }, { personFactory, personTemplateLtc },
-{ stringFactory, stringTemplateSrp }, { personFactory, personTemplateSrp }
+{rawFactory, rawTemplate}, { stringFactory, stringTemplateLtc }, { personFactory, personTemplateLtc },
+{rawFactory, rawTemplateLtc}, { stringFactory, stringTemplateSrp }, { personFactory, personTemplateSrp },
+{rawFactory, rawTemplateSrp}
 });
 }
 }

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

@@ -16,8 +16,9 @@
 package org.springframework.data.redis.listener;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.matchers.JUnitMatchers.hasItems;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -57,11 +58,11 @@ public class PubSubTests<T> {
 @SuppressWarnings("rawtypes")
 protected RedisTemplate template;
 
-private final BlockingDeque<String> bag = new LinkedBlockingDeque<String>(99);
+private final BlockingDeque<Object> bag = new LinkedBlockingDeque<Object>(99);
 
 private final Object handler = new Object() {
 @SuppressWarnings("unused")
-public void handleMessage(String message) {
+public void handleMessage(Object message) {
 bag.add(message);
 }
 };
@@ -117,27 +118,27 @@ protected T getT() {
 return factory.instance();
 }
 
+@SuppressWarnings("unchecked")
 @Test
 public void testContainerSubscribe() throws Exception {
-String payload1 = "do";
-String payload2 = "re mi";
+T payload1 = getT();
+T payload2 = getT();
 
 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));
+Set<T> set = new LinkedHashSet<T>();
+set.add((T) bag.poll(1, TimeUnit.SECONDS));
+set.add((T) bag.poll(1, TimeUnit.SECONDS));
 
-assertTrue(set.contains(payload1));
-assertTrue(set.contains(payload2));
+assertThat(set, hasItems(payload1, payload2));
 }
 
 @Test
 public void testMessageBatch() throws Exception {
 int COUNT = 10;
 for (int i = 0; i < COUNT; i++) {
-template.convertAndSend(CHANNEL, "message=" + i);
+template.convertAndSend(CHANNEL, getT());
 }
 
 Thread.sleep(1000);
@@ -146,8 +147,8 @@ public void testMessageBatch() throws Exception {
 
 @Test
 public void testContainerUnsubscribe() throws Exception {
-String payload1 = "do";
-String payload2 = "re mi";
+T payload1 = getT();
+T payload2 = getT();
 
 container.removeMessageListener(adapter, new ChannelTopic(CHANNEL));
 template.convertAndSend(CHANNEL, payload1);