Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Merge pull request #64 from tolgakavukcuevam/2.1-SNAPSHOT

Browse Source 
2.1 snapshot
This commit is contained in:
Rob Austin
2016-01-21 09:16:41 +00:00
parent ae4ac6491d 30237f08a1
commit 1cf67b492f
2 changed files with 263 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+1
View File
@@ -655,6 +655,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
replica.acquireModificationIterator(remoteIdentifier);
attached.remoteModificationIterator.setModificationNotifier(attached);
writer.writeRemoteBootstrapTimestamp(timeStampOfLastMessage);
writer.writeServerVersion();
src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java
+262 -9
View File
@@ -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<Integer, CharSequence> 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<java.lang.String, Integer> 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<java.lang.String, Integer> 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<java.lang.String, Integer> 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++;
}
reader.stop();
channelIdMap.close();
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<java.lang.String, Integer> 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 < 15000){
i = 400;
while (i < 500){
channelIdMap.update(Integer.toString(i),i+1);
i++;
}
Thread.sleep(1);
j++;
}
reader.stop();
channelIdMap.close();
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<Integer, CharSequence> 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,142 @@ public class TCPSocketReplicationBootStrapTest {
}
}
private class MapReader implements Runnable {
private volatile ChronicleMap<java.lang.String, Integer> chronicleMap;
private boolean stop = false;
private int firstGatheredSize = 0;
private Object firstGatheredElement;
public MapReader() {
}
public void stop() {
this.stop = false;
while (firstGatheredSize == 0){
this.stop = false;
}
this.stop = true;
}
public void run() {
try {
SingleChronicleHashReplication replicationHubForChannelIdMap2 =
SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5086, new InetSocketAddress("localhost", 5085)))
.createWithId((byte) 2);
ChronicleMap<String, Integer> 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<java.lang.String, Integer> chronicleMap;
private boolean stop = false;
private int firstGatheredSize = 0;
private CacheMapListener mapListener;
public MapReaderWithListener() {
}
public void stop() {
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<String, Integer> 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.chronicleMap.close();
} catch (Exception e) {
e.printStackTrace();
}
}
public int getFirstGatheredSize() {
return this.firstGatheredSize;
}
public Object getFirstGatheredElement() {
return this.mapListener.firstPuttedValue;
}
}
public class CacheMapListener<K,V> extends MapEventListener<K,V> {
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) {
}
}
}