diff --git a/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java b/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java index 7bc23c9d8..90a65da22 100644 --- a/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java +++ b/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java @@ -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 map1 = new HashMap<>(); + final HashMap map2 = new HashMap<>(); + + + // server 1 with identifier = 1 + { + ChronicleMapBuilder 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 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(); + } + } + } +