HCOLL-164 added a void marshaller and Map replication test using VOID value

This commit is contained in:
Rob Austin
2014-10-10 13:10:29 +01:00
parent 11fb016e97
commit 3630cfe555
12 changed files with 272 additions and 49 deletions
@@ -21,8 +21,12 @@ import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;
import java.util.concurrent.ArrayBlockingQueue;
import static java.util.concurrent.TimeUnit.SECONDS;
import static net.openhft.chronicle.map.Replicators.tcp;
/**
* @author Rob Austin.
*/
@@ -103,6 +107,26 @@ public class Builder {
};
}
public static <T extends ChronicleMap<Integer, Void>> T newTcpSocketShmIntString(
final byte identifier,
final int serverPort,
final InetSocketAddress... endpoints) throws IOException {
return (T) newTcpSocketShmBuilder(Integer.class, Void.class,
identifier, serverPort, endpoints).create(getPersistenceFile());
}
public static <K, V> ChronicleMapBuilder<K, V> newTcpSocketShmBuilder(
Class<K> kClass, Class<V> vClass,
final byte identifier,
final int serverPort,
final InetSocketAddress... endpoints) throws IOException {
TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(serverPort, endpoints)
.heartBeatInterval(1L, SECONDS);
return ChronicleMapBuilder.of(kClass, vClass)
.entries(20000L)
.addReplicator(tcp(identifier, tcpConfig));
}
interface MapProvider<T> {
T getMap();
@@ -26,9 +26,8 @@ import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import static java.util.concurrent.TimeUnit.SECONDS;
import static net.openhft.chronicle.map.Builder.getPersistenceFile;
import static net.openhft.chronicle.map.Replicators.tcp;
import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -44,18 +43,6 @@ public class TCPSocketReplication4WayMapTest {
private ChronicleMap<Integer, CharSequence> map3;
private ChronicleMap<Integer, CharSequence> map4;
public static <K, V> ChronicleMapBuilder<K, V> newTcpSocketShmBuilder(
Class<K> kClass, Class<V> vClass,
final byte identifier,
final int serverPort,
final InetSocketAddress... endpoints) throws IOException {
TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(serverPort, endpoints)
.heartBeatInterval(1L, SECONDS);
return ChronicleMapBuilder.of(kClass, vClass)
.entries(20000L)
.addReplicator(tcp(identifier, tcpConfig));
}
public static <T extends ChronicleMap<Integer, CharSequence>> T newTcpSocketShmIntString(
final byte identifier,
final int serverPort,
@@ -64,6 +51,8 @@ public class TCPSocketReplication4WayMapTest {
identifier, serverPort, endpoints).create(getPersistenceFile());
}
static ChronicleMap<IntValue, CharSequence> newTcpSocketShmIntValueString(
final byte identifier,
final int serverPort,
@@ -25,7 +25,7 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import static net.openhft.chronicle.map.Builder.getPersistenceFile;
import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmBuilder;
import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder;
import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmIntString;
import static org.junit.Assert.assertEquals;
@@ -30,7 +30,7 @@ import java.nio.ByteBuffer;
import java.util.Random;
import static net.openhft.chronicle.map.Builder.getPersistenceFile;
import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmBuilder;
import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder;
import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmIntValueString;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -26,7 +26,7 @@ import java.net.InetSocketAddress;
import java.util.Random;
import static net.openhft.chronicle.map.Builder.getPersistenceFile;
import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmBuilder;
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.assertTrue;
@@ -0,0 +1,138 @@
/*
* Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.openhft.chronicle.map;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import static net.openhft.chronicle.map.Builder.newTcpSocketShmIntString;
import static org.junit.Assert.*;
/**
* Test ReplicatedChronicleMap where the Replicated is over a TCP Socket
*
* @author Rob Austin.
*/
public class TCPSocketReplicationTest3VoidValueTest {
private ChronicleMap<Integer, Void> map1;
private ChronicleMap<Integer, Void> map2;
private ChronicleMap<Integer, Void> map3;
@Before
public void setup() throws IOException {
map1 = newTcpSocketShmIntString((byte) 1, 8076, new InetSocketAddress("localhost", 8077),
new InetSocketAddress("localhost", 8079));
map2 = newTcpSocketShmIntString((byte) 2, 8077, new InetSocketAddress("localhost", 8079));
map3 = newTcpSocketShmIntString((byte) 3, 8079);
}
@After
public void tearDown() throws InterruptedException {
Thread.sleep(5);
for (final Closeable closeable : new Closeable[]{map1, map2, map3}) {
try {
closeable.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
@Test
public void test3() throws IOException, InterruptedException {
assertEquals(null, map3.put(5, null));
// allow time for the recompilation to resolve
waitTillEqual(15000);
assertEquals(map1, map2);
assertEquals(map3, map2);
assertTrue(!map1.isEmpty());
}
@Test
public void test() throws IOException, InterruptedException {
assertEquals(null, map1.put(1, null));
assertEquals(null, map1.put(2, null));
assertEquals(null, map2.put(5, null));
assertEquals(null, map2.put(6,null));
map1.remove(2);
map2.remove(3);
map1.remove(3);
map2.put(5,null);
// allow time for the recompilation to resolve
waitTillEqual(5000);
assertEquals(map1, map2);
assertEquals(map3, map3);
assertTrue(!map1.isEmpty());
}
@Test
public void testClear() throws IOException, InterruptedException {
assertEquals(null, map1.put(1, null));
assertEquals(null, map1.put(2,null));
assertEquals(null, map2.put(5, null));
assertEquals(null, map2.put(6, null));
map1.clear();
map2.put(5, null);
// allow time for the recompilation to resolve
waitTillEqual(5000);
assertEquals(map1, map2);
assertEquals(map3, map3);
assertTrue(!map1.isEmpty());
}
private void waitTillEqual(final int timeOutMs) throws InterruptedException {
int t = 0;
for (; t < timeOutMs; t++) {
if (map1.equals(map2) &&
map1.equals(map3))
break;
Thread.sleep(1);
}
}
}
@@ -23,7 +23,7 @@ import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmBuilder;
import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;