DATAREDIS-463 - Improve test synchronization.
Several tests rely heavily on time to perform synchronization. Therfore we remove Thread.sleep from tests which do not require a sleep at all, use a condition where possible and increase timeouts on tests known to fail very likely due to a short Thread.sleep. Original Pull Request: #171
This commit is contained in:
committed by
Christoph Strobl
parent
6ae7e432eb
commit
d44103f399
@@ -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
|
||||
|
||||
@@ -186,8 +186,8 @@ public class RedisTemplateTests<K, V> {
|
||||
});
|
||||
List<V> list = Collections.singletonList(listValue);
|
||||
Set<V> set = new HashSet<V>(Collections.singletonList(setValue));
|
||||
Set<TypedTuple<V>> tupleSet = new LinkedHashSet<TypedTuple<V>>(Collections.singletonList(new DefaultTypedTuple<V>(
|
||||
zsetValue, 1d)));
|
||||
Set<TypedTuple<V>> tupleSet = new LinkedHashSet<TypedTuple<V>>(
|
||||
Collections.singletonList(new DefaultTypedTuple<V>(zsetValue, 1d)));
|
||||
assertThat(results, isEqual(Arrays.asList(new Object[] { value1, 1l, list, 1l, set, true, tupleSet })));
|
||||
}
|
||||
|
||||
@@ -433,7 +433,8 @@ public class RedisTemplateTests<K, V> {
|
||||
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<K, V> {
|
||||
final DefaultRedisScript<String> script = new DefaultRedisScript<String>();
|
||||
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
|
||||
|
||||
@@ -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<String> msgs = new ArrayList<String>();
|
||||
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<String> set = new LinkedHashSet<String>();
|
||||
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<String> set = new LinkedHashSet<String>();
|
||||
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<String>(Arrays.asList(new String[] { "HELLO", "WORLD" })), set);
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user