diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index ce442529a..5a55c4265 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -650,11 +650,12 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo LOG.debug("handshaking for localIdentifier=" + localIdentifier + "," + "remoteIdentifier=" + remoteIdentifier); - long timeStampOfLastMessage = replica.lastModificationTime(remoteIdentifier); + attached.remoteModificationIterator = replica.acquireModificationIterator(remoteIdentifier); attached.remoteModificationIterator.setModificationNotifier(attached); + long timeStampOfLastMessage = replica.lastModificationTime(remoteIdentifier); writer.writeRemoteBootstrapTimestamp(timeStampOfLastMessage); writer.writeServerVersion(); diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java index b61139887..921c92ef5 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java @@ -16,7 +16,9 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import org.jetbrains.annotations.Nullable; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -33,6 +35,7 @@ import static net.openhft.chronicle.map.Builder.getPersistenceFile; import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder; import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmIntString; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; /** * Test ReplicatedChronicleMap where the Replicated is over a TCP Socket @@ -60,7 +63,7 @@ public class TCPSocketReplicationBootStrapTest { long lastModificationTime; - // lets make sure that the message has got to map 1 + // lets make sure that the message has got to chronicleMap 1 do { lastModificationTime = map1.lastModificationTime((byte) 2); Thread.yield(); @@ -70,7 +73,7 @@ public class TCPSocketReplicationBootStrapTest { map2a.close(); { - // restart map 2 but don't doConnect it to map one + // restart chronicleMap 2 but don't doConnect it to chronicleMap one final ChronicleMap map2b = map2aBuilder.createPersistedTo(map2File); // add data into it @@ -89,6 +92,117 @@ public class TCPSocketReplicationBootStrapTest { persistenceFile.delete(); } + @Test + public void testReplication() throws IOException, InterruptedException { + + SingleChronicleHashReplication replicationHubForChannelIdMap = + SingleChronicleHashReplication.builder().bootstrapOnlyLocalEntries(true) + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5085,new InetSocketAddress[0])) + .createWithId((byte)1); + replicationHubForChannelIdMap.bootstrapOnlyLocalEntries(); + + ChronicleMap channelIdMap = ChronicleMapBuilder.of(java.lang.String.class, Integer.class) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap) + .create(); + + + channelIdMap.put("1",1); + SingleChronicleHashReplication replicationHubForChannelIdMap2 = + SingleChronicleHashReplication.builder().bootstrapOnlyLocalEntries(true) + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5086,new InetSocketAddress("localhost",5085))) + .createWithId((byte)2); + replicationHubForChannelIdMap.bootstrapOnlyLocalEntries(); + + ChronicleMap channelIdMap2 = ChronicleMapBuilder.of(java.lang.String.class, Integer.class) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap2) + .create(); + + + Thread.sleep(300); + + int map2Size = channelIdMap2.size(); + channelIdMap.close(); + channelIdMap2.close(); + assertEquals(1,map2Size); + } + + @Test + public void testReplicationWhileModifying() throws IOException, InterruptedException { + + SingleChronicleHashReplication replicationHubForChannelIdMap = + SingleChronicleHashReplication.builder().bootstrapOnlyLocalEntries(true) + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5085,new InetSocketAddress[0])) + .createWithId((byte)1); + + ChronicleMap channelIdMap = ChronicleMapBuilder.of(java.lang.String.class, Integer.class) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap) + .create(); + + int i; + for(i = 0;i < 500;i++){ + channelIdMap.put(Integer.toString(i),i); + } + MapReader reader = new MapReader(); + Thread t = new Thread(reader); + t.start(); + int j = 0; + while (j < 100){ + i = 400; + while (i < 500){ + channelIdMap.update(Integer.toString(i),i+1); + i++; + } + j++; + } + channelIdMap.close(); + reader.stop(); + int firstGatheredSize = reader.getFirstGatheredSize(); + assertEquals(0,reader.getFirstGatheredElement()); + assertEquals(500,firstGatheredSize); + } + + @Test + public void testReplicationWhileModifyingWithMapListener() throws IOException, InterruptedException { + + SingleChronicleHashReplication replicationHubForChannelIdMap = + SingleChronicleHashReplication.builder() + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5085,new InetSocketAddress[0])) + .createWithId((byte)1); + + ChronicleMap channelIdMap = ChronicleMapBuilder.of(java.lang.String.class, Integer.class) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap) + .create(); + + int i; + for(i = 0;i < 500;i++){ + channelIdMap.put(Integer.toString(i),i); + } + MapReaderWithListener reader = new MapReaderWithListener(); + Thread t = new Thread(reader); + t.start(); + int j = 0; + while (j < 100){ + i = 400; + while (i < 500){ + channelIdMap.update(Integer.toString(i),i+1); + i++; + } + j++; + } + Thread.sleep(10000); + channelIdMap.close(); + reader.stop(); + assertEquals(0,reader.getFirstGatheredElement()); + } + @Test public void testBootstrapAndHeartbeat() throws IOException, InterruptedException { @@ -113,7 +227,7 @@ public class TCPSocketReplicationBootStrapTest { long lastModificationTime; - // lets make sure that the message has got to map 1 + // lets make sure that the message has got to chronicleMap 1 do { lastModificationTime = map1.lastModificationTime((byte) 2); Thread.yield(); @@ -122,7 +236,7 @@ public class TCPSocketReplicationBootStrapTest { map2a.close(); { - // restart map 2 but does not connect it to map1 + // restart chronicleMap 2 but does not connect it to map1 final ChronicleMap map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class).replication((byte) 2).createPersistedTo(persistenceFile); // add data into it @@ -143,11 +257,13 @@ public class TCPSocketReplicationBootStrapTest { @After public void tearDown() { - for (final Closeable closeable : new Closeable[]{map1, map2}) { - try { - closeable.close(); - } catch (Exception e) { - e.printStackTrace(); + if(map1 != null && map2 != null) { + for (final Closeable closeable : new Closeable[]{map1, map2}) { + try { + closeable.close(); + } catch (Exception e) { + e.printStackTrace(); + } } } System.gc(); @@ -180,5 +296,140 @@ public class TCPSocketReplicationBootStrapTest { } } + private class MapReader implements Runnable { + + private volatile ChronicleMap chronicleMap; + private boolean stop = false; + private int firstGatheredSize = 0; + private Object firstGatheredElement; + + public MapReader() { + } + + public void stop() { + this.stop = true; + } + + public void run() { + + try { + SingleChronicleHashReplication replicationHubForChannelIdMap2 = + SingleChronicleHashReplication.builder() + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5086, new InetSocketAddress("localhost", 5085))) + .createWithId((byte) 2); + ChronicleMap channelIdMap2 = ChronicleMapBuilder.of(String.class, Integer.class) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap2) + .create(); + this.chronicleMap = channelIdMap2; + Object firstEl = 0; + while (!stop) { + int size = channelIdMap2.size(); + //System.out.println("Size=[" + size + "] on timestamp [" + Long.toString(System.currentTimeMillis()) + "]"); + if (size != 0 && this.firstGatheredSize == 0) { + this.firstGatheredSize = size; + firstEl = channelIdMap2.get("0"); + } + } + this.chronicleMap.close(); + firstGatheredElement = firstEl; + } catch (Exception e) { + e.printStackTrace(); + } + + } + + public int getFirstGatheredSize() { + return this.firstGatheredSize; + } + + public Object getFirstGatheredElement() { + return this.firstGatheredElement; + } + + } + + private class MapReaderWithListener implements Runnable { + + private volatile ChronicleMap chronicleMap; + private boolean stop = false; + private int firstGatheredSize = 0; + private Object firstGatheredElement; + private CacheMapListener mapListener; + + public MapReaderWithListener() { + } + + public void stop() { + while (mapListener.firstPuttedValue == null){ + this.stop = false; + } + this.stop = true; + } + + public void run() { + + try { + this.mapListener = new CacheMapListener<>(); + + SingleChronicleHashReplication replicationHubForChannelIdMap2 = + SingleChronicleHashReplication.builder() + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5086, new InetSocketAddress("localhost", 5085))) + .createWithId((byte) 2); + ChronicleMap channelIdMap2 = ChronicleMapBuilder.of(String.class, Integer.class).eventListener(this.mapListener) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap2) + .create(); + this.chronicleMap = channelIdMap2; + while (!stop) { + channelIdMap2.size(); + } + this.firstGatheredElement = mapListener.firstPuttedValue; + this.chronicleMap.close(); + } catch (Exception e) { + e.printStackTrace(); + } + + } + + public int getFirstGatheredSize() { + return this.firstGatheredSize; + } + + public Object getFirstGatheredElement() { + return this.firstGatheredElement; + } + + } + + public class CacheMapListener extends MapEventListener { + + + private Object firstPuttedValue = null; + + public CacheMapListener() { + + } + + @Override + public void onPut(K key, V newValue, @Nullable V replacedValue, boolean replicationEvent, + boolean added, boolean hasValueChanged, byte identifier, + byte replacedIdentifier, long timeStamp, long replacedTimeStamp) { + if(this.firstPuttedValue == null){ + this.firstPuttedValue = newValue; + } + } + + @Override + public void onRemove(K key, V value, boolean replicationEvent, byte identifier, + byte replacedIdentifier, long timestamp, long replacedTimeStamp) { + + } + } + + + }