From 52fe4cd827183faec02974b515866d6706dd4088 Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Thu, 17 Mar 2011 13:57:40 +0200 Subject: [PATCH] DATAKV-49 + add more integration tests --- .../AbstractConnectionIntegrationTests.java | 133 ++++++++++++++++++ .../JedisConnectionIntegrationTests.java | 80 ----------- .../JRedisConnectionIntegrationTests.java | 13 ++ .../redis/listener/PubSubTestParams.java | 4 +- .../keyvalue/redis/listener/PubSubTests.java | 2 +- 5 files changed, 149 insertions(+), 83 deletions(-) diff --git a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/AbstractConnectionIntegrationTests.java b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/AbstractConnectionIntegrationTests.java index d3056e4bc..cb8c408ce 100644 --- a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/AbstractConnectionIntegrationTests.java +++ b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/AbstractConnectionIntegrationTests.java @@ -22,6 +22,9 @@ import java.util.Arrays; import java.util.List; import java.util.Properties; import java.util.UUID; +import java.util.concurrent.BlockingDeque; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.atomic.AtomicBoolean; import org.junit.After; import org.junit.AfterClass; @@ -184,4 +187,134 @@ public abstract class AbstractConnectionIntegrationTests { assertNull(connection.hKeys("~")); connection.closePipeline(); } + + // pub sub test + + @Test + public void testPubSub() throws Exception { + + final BlockingDeque queue = new LinkedBlockingDeque(); + + final MessageListener ml = new MessageListener() { + @Override + public void onMessage(Message message, byte[] pattern) { + queue.add(message); + System.out.println("received message"); + } + }; + + final byte[] channel = "foo.tv".getBytes(); + final RedisConnection subConn = getConnectionFactory().getConnection(); + + assertNotSame(connection, subConn); + + + final AtomicBoolean flag = new AtomicBoolean(true); + + Runnable listener = new Runnable() { + @Override + public void run() { + subConn.subscribe(ml, channel); + System.out.println("Subscribed"); + while (flag.get()) { + try { + Thread.currentThread().wait(2000); + } catch (Exception ex) { + return; + } + } + } + }; + + Thread th = new Thread(listener, "listener"); + th.start(); + + try { + Thread.sleep(1500); + connection.publish(channel, "one".getBytes()); + connection.publish(channel, "two".getBytes()); + connection.publish(channel, "I see you".getBytes()); + System.out.println("Done publishing..."); + Thread.sleep(3000); + } finally { + flag.set(false); + } + assertEquals(3, queue.size()); + } + + @Test + public void testPubSubWithNamedChannels() { + final byte[] expectedChannel = "channel1".getBytes(); + final byte[] expectedMessage = "msg".getBytes(); + + MessageListener listener = new MessageListener() { + + @Override + public void onMessage(Message message, byte[] pattern) { + assertArrayEquals(expectedChannel, message.getChannel()); + assertArrayEquals(expectedMessage, message.getBody()); + } + }; + + 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 + RedisConnection connection2 = getConnectionFactory().getConnection(); + connection2.publish(expectedMessage, expectedChannel); + connection2.close(); + // unsubscribe connection + connection.getSubscription().unsubscribe(); + } + }); + + th.start(); + 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(Message message, byte[] pattern) { + assertArrayEquals(expectedPattern, pattern); + assertArrayEquals(expectedMessage, message.getBody()); + System.out.println("Received message '" + new String(message.getBody()) + "'"); + } + }; + + 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 + RedisConnection connection2 = getConnectionFactory().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); + } } \ No newline at end of file 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 75a9e7e87..302a94e49 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 @@ -16,13 +16,9 @@ package org.springframework.data.keyvalue.redis.connection.jedis; -import static org.junit.Assert.*; - import org.junit.Test; import org.springframework.data.keyvalue.redis.SettingsUtils; import org.springframework.data.keyvalue.redis.connection.AbstractConnectionIntegrationTests; -import org.springframework.data.keyvalue.redis.connection.Message; -import org.springframework.data.keyvalue.redis.connection.MessageListener; import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory; import redis.clients.jedis.BinaryJedis; @@ -47,82 +43,6 @@ public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrati return factory; } - @Test - public void testPubSubWithNamedChannels() { - final byte[] expectedChannel = "channel1".getBytes(); - final byte[] expectedMessage = "msg".getBytes(); - - MessageListener listener = new MessageListener() { - - @Override - public void onMessage(Message message, byte[] pattern) { - assertArrayEquals(expectedChannel, message.getChannel()); - assertArrayEquals(expectedMessage, message.getBody()); - } - }; - - 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, expectedChannel); - connection2.close(); - // unsubscribe connection - connection.getSubscription().unsubscribe(); - } - }); - - th.start(); - 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(Message message, byte[] pattern) { - assertArrayEquals(expectedPattern, pattern); - assertArrayEquals(expectedMessage, message.getBody()); - System.out.println("Received message '" + new String(message.getBody()) + "'"); - } - }; - - 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 testMulti() throws Exception { byte[] key = "key".getBytes(); diff --git a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/jredis/JRedisConnectionIntegrationTests.java b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/jredis/JRedisConnectionIntegrationTests.java index 07cdfcf29..ed92f9f63 100644 --- a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/jredis/JRedisConnectionIntegrationTests.java +++ b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/connection/jredis/JRedisConnectionIntegrationTests.java @@ -74,4 +74,17 @@ public class JRedisConnectionIntegrationTests extends AbstractConnectionIntegrat @Ignore public void testNullSerialization() throws Exception { } + + @Ignore + public void testPubSub() throws Exception { + } + + @Ignore + public void testPubSubWithPatterns() { + } + + @Ignore + public void testPubSubWithNamedChannels() { + + } } \ No newline at end of file diff --git a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTestParams.java b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTestParams.java index 28fd764c3..cba742c7f 100644 --- a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTestParams.java +++ b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTestParams.java @@ -61,8 +61,8 @@ public class PubSubTestParams { RedisTemplate personTemplateRJC = new RedisTemplate(rjcConnFactory); - return Arrays.asList(new Object[][] { { stringFactory, stringTemplate }, { personFactory, personTemplate } - //,{ stringFactory, stringTemplateRJC }, { personFactory, personTemplateRJC } + return Arrays.asList(new Object[][] { { stringFactory, stringTemplate }, { personFactory, personTemplate }, + { stringFactory, stringTemplateRJC }, { personFactory, personTemplateRJC } }); } } diff --git a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTests.java b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTests.java index 55bb59fb8..a61f923f0 100644 --- a/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTests.java +++ b/spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTests.java @@ -71,7 +71,7 @@ public class PubSubTests { container.addMessageListener(adapter, Arrays.asList(new ChannelTopic(CHANNEL))); container.afterPropertiesSet(); - Thread.sleep(500); + Thread.sleep(1000); } @After