added a test to check that we can support a thousand channels

This commit is contained in:
Rob Austin
2015-04-28 11:45:01 +01:00
parent dadc3db8be
commit e9f475a3ed
@@ -3,13 +3,16 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.ReplicationChannel;
import net.openhft.chronicle.hash.replication.ReplicationHub;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;
import static net.openhft.chronicle.map.ChronicleMapBuilder.of;
import static org.junit.Assert.assertEquals;
/**
@@ -121,4 +124,122 @@ public class TestEntriesSpanningSeveralChunksReplicationHub {
favoriteColourServer1.close();
}
public static final int NUMBER_OF_CHANNELS = 1000;
@Test(timeout = 50000)
public void test1000Channels()
throws IOException, InterruptedException {
final HashMap<Short, ChronicleMap> map1 = new HashMap<>();
final HashMap<Short, ChronicleMap> map2 = new HashMap<>();
// server 1 with identifier = 1
{
ChronicleMapBuilder<CharSequence, Short> builder =
of(CharSequence.class, Short.class)
.averageKeySize(10)
.averageValueSize(100)
.entries(1000);
byte identifier = (byte) 1;
TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(8021);
ReplicationHub hubOnServer1 = ReplicationHub.builder()
.tcpTransportAndNetwork(tcpConfig)
.maxNumberOfChannels(NUMBER_OF_CHANNELS)
.createWithId(identifier);
for (short channelId = 1; channelId < NUMBER_OF_CHANNELS; channelId++) {
ReplicationChannel channel = hubOnServer1.createChannel(channelId);
ChronicleMap map = builder.instance().replicatedViaChannel(channel).create();
map.put("key", channelId);
map1.put(channelId, map);
}
}
{
ChronicleMapBuilder<CharSequence, Short> builder =
of(CharSequence.class, Short.class)
.averageKeySize(10)
.averageValueSize(100)
.entries(1000);
byte identifier = (byte) 2;
TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(8022, new InetSocketAddress("localhost", 8021));
ReplicationHub hubOnServer1 = ReplicationHub.builder()
.tcpTransportAndNetwork(tcpConfig)
.maxNumberOfChannels(NUMBER_OF_CHANNELS)
.createWithId(identifier);
for (short channelId = 1; channelId < NUMBER_OF_CHANNELS; channelId++) {
ReplicationChannel channel = hubOnServer1.createChannel(channelId);
ChronicleMap map = builder.instance().replicatedViaChannel(channel).create();
map.put("key", channelId);
map2.put(channelId, map);
}
}
// check that the maps are the same, if the are not the same the test will timeout
// we have to give a few seconds for the replication to propagate
for (short channelId = 1; channelId < NUMBER_OF_CHANNELS; channelId++) {
// allow time for the recompilation to resolve
final ChronicleMap v1 = map1.get(channelId);
final ChronicleMap v2 = map2.get(channelId);
for (; ; ) {
if (v1 == null || v2 == null) {
Thread.sleep(10);
continue;
}
if (v1.isEmpty() || v2.isEmpty()) {
Thread.sleep(10);
continue;
}
if (!v1.equals(v2)) {
Thread.sleep(10);
continue;
}
Assert.assertEquals(channelId, v1.get("key"));
Assert.assertEquals(channelId, v2.get("key"));
break;
}
}
// close
for (short channelId = 1; channelId < NUMBER_OF_CHANNELS; channelId++) {
final ChronicleMap v1 = map1.get(channelId);
if (v1 != null)
v1.close();
final ChronicleMap v2 = map2.get(channelId);
if (v2 != null)
v2.close();
}
}
}