diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index 4bf7dca8d..c8b15a46c 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -34,6 +34,7 @@ import java.util.Properties; import java.util.Set; import java.util.UUID; import java.util.concurrent.BlockingDeque; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -77,6 +78,7 @@ import org.springframework.test.annotation.ProfileValueSourceConfiguration; * @author Jennifer Hickey * @author Christoph Strobl * @author Thomas Darimont + * @author Mark Paluch */ @ProfileValueSourceConfiguration(RedisTestProfileValueSource.class) public abstract class AbstractConnectionIntegrationTests { @@ -340,10 +342,12 @@ public abstract class AbstractConnectionIntegrationTests { assumeTrue(RedisVersionUtils.atLeast("2.6", byteConnection)); initConnection(); final AtomicBoolean scriptDead = new AtomicBoolean(false); + final CountDownLatch sync = new CountDownLatch(1); Thread th = new Thread(new Runnable() { public void run() { DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(connectionFactory.getConnection()); try { + sync.countDown(); conn2.eval("local time=1 while time < 10000000000 do time=time+1 end", ReturnType.BOOLEAN, 0); } catch (DataAccessException e) { scriptDead.set(true); @@ -352,7 +356,8 @@ public abstract class AbstractConnectionIntegrationTests { } }); th.start(); - Thread.sleep(1000); + sync.await(2, TimeUnit.SECONDS); + Thread.sleep(200); connection.scriptKill(); getResults(); assertTrue(waitFor(new TestCondition() { @@ -377,11 +382,10 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "runLongTests", value = "true") public void testPersist() throws Exception { connection.set("exp3", "true"); - actual.add(connection.expire("exp3", 1)); + actual.add(connection.expire("exp3", 30)); actual.add(connection.persist("exp3")); - Thread.sleep(1500); - actual.add(connection.exists("exp3")); - verifyResults(Arrays.asList(new Object[] { true, true, true })); + actual.add(connection.ttl("exp3")); + verifyResults(Arrays.asList(new Object[] { true, true, -1L })); } @Test @@ -411,7 +415,6 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "runLongTests", value = "true") public void testBRPopTimeout() throws Exception { actual.add(connection.bRPop(1, "alist")); - Thread.sleep(1500l); verifyResults(Arrays.asList(new Object[] { null })); } @@ -419,7 +422,6 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "runLongTests", value = "true") public void testBLPopTimeout() throws Exception { actual.add(connection.bLPop(1, "alist")); - Thread.sleep(1500l); verifyResults(Arrays.asList(new Object[] { null })); } @@ -427,7 +429,6 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "runLongTests", value = "true") public void testBRPopLPushTimeout() throws Exception { actual.add(connection.bRPopLPush(1, "alist", "foo")); - Thread.sleep(1500l); verifyResults(Arrays.asList(new Object[] { null })); } @@ -630,12 +631,16 @@ public abstract class AbstractConnectionIntegrationTests { Thread th = new Thread(new Runnable() { public void run() { - // sleep 1/2 second to let the registration happen + // sync to let the registration happen + waitFor(new TestCondition() { + @Override + public boolean passes() { + return connection.isSubscribed(); + } + }, 2000); try { Thread.sleep(500); - } catch (InterruptedException ex) { - throw new RuntimeException(ex); - } + } catch (InterruptedException o_O) {} // open a new connection RedisConnection connection2 = connectionFactory.getConnection(); @@ -678,12 +683,17 @@ public abstract class AbstractConnectionIntegrationTests { Thread th = new Thread(new Runnable() { public void run() { - // sleep 1/2 second to let the registration happen + // sync to let the registration happen + waitFor(new TestCondition() { + @Override + public boolean passes() { + return connection.isSubscribed(); + } + }, 2000); + try { Thread.sleep(500); - } catch (InterruptedException ex) { - throw new RuntimeException(ex); - } + } catch (InterruptedException o_O) {} // open a new connection RedisConnection connection2 = connectionFactory.getConnection(); @@ -1004,7 +1014,7 @@ public abstract class AbstractConnectionIntegrationTests { actual.add(connection.get("testing")); connection.restore("testing".getBytes(), 100l, (byte[]) results.get(0)); verifyResults(Arrays.asList(new Object[] { 1l, null })); - assertTrue(waitFor(new KeyExpired("testing"), 300l)); + assertTrue(waitFor(new KeyExpired("testing"), 400l)); } @Test diff --git a/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java b/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java index 61b0f0927..3529680b8 100644 --- a/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java +++ b/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java @@ -186,8 +186,8 @@ public class RedisTemplateTests { }); List list = Collections.singletonList(listValue); Set set = new HashSet(Collections.singletonList(setValue)); - Set> tupleSet = new LinkedHashSet>(Collections.singletonList(new DefaultTypedTuple( - zsetValue, 1d))); + Set> tupleSet = new LinkedHashSet>( + Collections.singletonList(new DefaultTypedTuple(zsetValue, 1d))); assertThat(results, isEqual(Arrays.asList(new Object[] { value1, 1l, list, 1l, set, true, tupleSet }))); } @@ -433,7 +433,8 @@ public class RedisTemplateTests { final K key1 = keyFactory.instance(); V value1 = valueFactory.instance(); redisTemplate.boundValueOps(key1).set(value1); - redisTemplate.expire(key1, 10, TimeUnit.MILLISECONDS); + redisTemplate.expire(key1, 500, TimeUnit.MILLISECONDS); + assertTrue(redisTemplate.getExpire(key1, TimeUnit.MILLISECONDS) > 0l); // Timeout is longer because expire will be 1 sec if pExpire not supported waitFor(new TestCondition() { @@ -695,10 +696,8 @@ public class RedisTemplateTests { final DefaultRedisScript script = new DefaultRedisScript(); script.setScriptText("return 'Hey'"); script.setResultType(String.class); - assertEquals( - "Hey", - redisTemplate.execute(script, redisTemplate.getValueSerializer(), new StringRedisSerializer(), - Collections.singletonList(key1))); + assertEquals("Hey", redisTemplate.execute(script, redisTemplate.getValueSerializer(), new StringRedisSerializer(), + Collections.singletonList(key1))); } @Test 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 077b2398f..79d50ccdd 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2014 the original author or authors. + * Copyright 2011-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,6 +19,7 @@ package org.springframework.data.redis.listener; import static org.hamcrest.core.Is.*; import static org.junit.Assert.*; import static org.junit.Assume.*; +import static org.springframework.data.redis.SpinBarrier.*; import java.util.ArrayList; import java.util.Arrays; @@ -44,6 +45,7 @@ import org.springframework.core.task.SyncTaskExecutor; import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.RedisTestProfileValueSource; import org.springframework.data.redis.SettingsUtils; +import org.springframework.data.redis.TestCondition; import org.springframework.data.redis.connection.ConnectionUtils; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; @@ -58,6 +60,7 @@ import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; * @author Costin Leau * @author Jennifer Hickey * @author Christoph Strobl + * @author Mark Paluch */ @RunWith(Parameterized.class) public class PubSubResubscribeTests { @@ -149,7 +152,12 @@ public class PubSubResubscribeTests { container.afterPropertiesSet(); container.start(); - Thread.sleep(1000); + waitFor(new TestCondition() { + @Override + public boolean passes() { + return container.getConnectionFactory().getConnection().isSubscribed(); + } + }, 1000); } @After @@ -175,14 +183,17 @@ public class PubSubResubscribeTests { container.addMessageListener(anotherListener, new PatternTopic(PATTERN)); container.removeMessageListener(adapter); + // Wait for async subscription tasks to setup + Thread.sleep(400); + // test no messages are sent just to patterns template.convertAndSend(CHANNEL, payload1); template.convertAndSend(ANOTHER_CHANNEL, payload2); // anotherListener receives both messages List msgs = new ArrayList(); - msgs.add(bag2.poll(1, TimeUnit.SECONDS)); - msgs.add(bag2.poll(1, TimeUnit.SECONDS)); + msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS)); + msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS)); assertEquals(2, msgs.size()); assertTrue(msgs.contains(payload1)); @@ -190,21 +201,29 @@ public class PubSubResubscribeTests { msgs.clear(); // unsubscribed adapter did not receive message - assertNull(bag.poll(1, TimeUnit.SECONDS)); + assertNull(bag.poll(500, TimeUnit.MILLISECONDS)); // bind original listener on another channel container.addMessageListener(adapter, new ChannelTopic(ANOTHER_CHANNEL)); + // Wait for async subscription tasks to setup + Thread.sleep(400); + template.convertAndSend(CHANNEL, payload1); template.convertAndSend(ANOTHER_CHANNEL, payload2); // original listener received only one message on another channel - assertEquals(payload2, bag.poll(1, TimeUnit.SECONDS)); - assertNull(bag.poll(1, TimeUnit.SECONDS)); + msgs.clear(); + msgs.add(bag.poll(500, TimeUnit.MILLISECONDS)); + msgs.add(bag.poll(500, TimeUnit.MILLISECONDS)); + + assertTrue(msgs.contains(payload2)); + assertTrue(msgs.contains(null)); // another listener receives messages on both channels - msgs.add(bag2.poll(1, TimeUnit.SECONDS)); - msgs.add(bag2.poll(1, TimeUnit.SECONDS)); + msgs.clear(); + msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS)); + msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS)); assertEquals(2, msgs.size()); assertTrue(msgs.contains(payload1)); assertTrue(msgs.contains(payload2)); @@ -225,6 +244,10 @@ public class PubSubResubscribeTests { container.addMessageListener(adapter, new ChannelTopic(ANOTHER_CHANNEL)); container.removeMessageListener(null, new ChannelTopic(CHANNEL)); + // timing: There's currently no other way to synchronize + // than to hope the subscribe/unsubscribe are executed within the time. + Thread.sleep(400); + // Listener removed from channel template.convertAndSend(CHANNEL, payload1); template.convertAndSend(CHANNEL, payload2); @@ -234,8 +257,8 @@ public class PubSubResubscribeTests { template.convertAndSend(ANOTHER_CHANNEL, anotherPayload2); Set set = new LinkedHashSet(); - set.add(bag.poll(1, TimeUnit.SECONDS)); - set.add(bag.poll(1, TimeUnit.SECONDS)); + set.add(bag.poll(500, TimeUnit.MILLISECONDS)); + set.add(bag.poll(500, TimeUnit.MILLISECONDS)); assertFalse(set.contains(payload1)); assertFalse(set.contains(payload2)); @@ -259,15 +282,16 @@ public class PubSubResubscribeTests { Arrays.asList(new Topic[] { new ChannelTopic(CHANNEL), new PatternTopic("s*") })); container.start(); - // Wait for async subscription tasks to setup + // timing: There's currently no other way to synchronize + // than to hope the subscribe/unsubscribe are executed within the time. Thread.sleep(1000); template.convertAndSend("somechannel", "HELLO"); template.convertAndSend(CHANNEL, "WORLD"); Set set = new LinkedHashSet(); - set.add(bag.poll(1, TimeUnit.SECONDS)); - set.add(bag.poll(1, TimeUnit.SECONDS)); + set.add(bag.poll(500, TimeUnit.MILLISECONDS)); + set.add(bag.poll(500, TimeUnit.MILLISECONDS)); assertEquals(new HashSet(Arrays.asList(new String[] { "HELLO", "WORLD" })), set); } diff --git a/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java b/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java index 2435dd549..80325c481 100644 --- a/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2014 the original author or authors. + * Copyright 2011-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -47,6 +47,7 @@ import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; * @author Jennifer Hickey * @author Thomas Darimont * @author Christoph Strobl + * @author Mark Paluch */ @RunWith(Parameterized.class) public class SubscriptionConnectionTests { @@ -135,9 +136,11 @@ public class SubscriptionConnectionTests { container.afterPropertiesSet(); container.start(); - // Need to sleep shortly as jedis cannot deal propery with multiple repsonses within one connection - // @see https://github.com/xetorthio/jedis/issues/186 - Thread.sleep(1000); + if (connectionFactory instanceof JedisConnectionFactory) { + // Need to sleep shortly as jedis cannot deal propery with multiple repsonses within one connection + // @see https://github.com/xetorthio/jedis/issues/186 + Thread.sleep(100); + } container.stop(); containers.add(container);