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

added number based tests - still work in progress

Browse Source
This commit is contained in:
Rob Austin
2015-03-04 15:07:07 +00:00
parent e28d980c61
commit 2741d22504
5 changed files with 128 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+2 -2
View File
@@ -545,13 +545,13 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
private boolean testTimeStampInSensibleRange() {
if (rm().timeProvider == TimeProvider.SYSTEM) {
/* if (rm().timeProvider == TimeProvider.SYSTEM) {
long currentTime = TimeProvider.SYSTEM.currentTime();
assert Math.abs(currentTime - timestamp) <= 100000000 :
"unrealistic timestamp: " + timestamp;
assert Math.abs(currentTime - newTimestamp) <= 100000000 :
"unrealistic newTimestamp: " + newTimestamp;
}
}*/
return true;
}
src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java
+63 -40
View File
@@ -141,6 +141,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
private byte remoteIdentifier;
private Bytes resultBytes = Bytes.wrap(resultBuffer);
private SelectionKey key;
private long transactionId;
public StatelessWiredConnector(@Nullable List<Replica> channelList, byte localIdentifier) {
this.channelList = channelList;
@@ -153,7 +154,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
public void onRead(SocketChannel socketChannel, SelectionKey key) throws IOException {
this.key = key;
// if (!handshingComplete)
// if (!handshingComplete)
// outWire.bytes().writeByte(localIdentifier);
final int len = readSocket(socketChannel);
@@ -164,14 +165,11 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
}
if (len == 0) {
System.out.println("nothing read");
return;
}
assert inWire.bytes().remaining() < 999;
System.out.println("remaining=>" + inWire.bytes().remaining());
while (inWire.bytes().remaining() > 4) {
final long limit = inWire.bytes().limit();
@@ -184,18 +182,8 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
return;
}
// if (!handshingComplete) {
System.out.println("doHandshaking");
// onHandShaking();
// System.out.println("end-doHandshaking");
// } else {
System.out.println("onEvent");
onEvent();
System.out.println("end-onEvent");
// }
System.out.println("limit=>" + outWire.bytes().limit());
assert inWire.bytes().remaining() < 999;
inWire.bytes().limit(limit);
assert inWire.bytes().remaining() < 999;
@@ -203,7 +191,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
}
private void clearInBuffer() {
System.out.println("clear");
inWire.bytes().clear();
inWireBuffer().clear();
}
@@ -233,7 +220,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
}
private void compactInBuffer() {
System.out.println("compacting");
inWireBuffer().position((int) inWire.bytes().position());
inWireBuffer().limit(inWireBuffer().position());
inWireBuffer().compact();
@@ -249,7 +235,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
private int readSocket(SocketChannel socketChannel) throws IOException {
ByteBuffer dst = inWireBuffer();
int len = socketChannel.read(dst);
System.out.println("len=" + len);
int readUpTo = dst.position();
inWire.bytes().limit(readUpTo);
return len;
@@ -288,8 +273,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
// it is assumed by this point that the buffer has all the bytes in it for this message
System.out.println("StatelessWiredConnector - trans=" + Bytes.toHex(inWire.bytes()));
long transactionId = inWire.read(() -> "TRANSACTION_ID").int64();
transactionId = inWire.read(() -> "TRANSACTION_ID").int64();
timestamp = inWire.read(() -> "TIME_STAMP").int64();
channelId = inWire.read(() -> "CHANNEL_ID").int16();
inWire.read(() -> "METHOD_NAME").text(methodName);
@@ -314,13 +298,13 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
}
private void writeLength(@NotNull final Wire textWire) {
private void writeLength(@NotNull final Wire textWire, long start) {
if (textWire.bytes().position() > Integer.MAX_VALUE)
throw new IllegalStateException("position too large");
// write the size
textWire.bytes().writeInt(0, (int) textWire.bytes().position());
textWire.bytes().writeInt(start, (int) (textWire.bytes().position() - start));
}
/**
@@ -351,6 +335,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
final StringWriter sw = new StringWriter();
final PrintWriter pw = new PrintWriter(sw);
e.printStackTrace(pw);
outWire.write(() -> "IS_EXCEPTION").bool(true);
wire.write(() -> "EXCEPTION").text(sw.toString());
return null;
}
@@ -405,8 +390,8 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2");
// skip 4 bytes for the size
// skip 4 bytes where we will write the size
long locationOfSize = outWire.bytes().position();
outWire.bytes().skip(4);
final BytesChronicleMap bytesMap = bytesMap(channelId);
@@ -418,43 +403,81 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
// todo
net.openhft.lang.io.Bytes result = bytesMap.put(reader, reader, timestamp, remoteIdentifier);
// clear the buffer we just used to write the lang bytes
inLanByteBuffer.clear();
inLangBytes.clear();
outWire.write(() -> "IS_EXCEPTION").bool(false);
if (result != null && result.remaining() > 0) {
BB_ADDRESS.set(resultBuffer, result.address());
BB_CAPACITY.set(resultBuffer, result.capacity());
resultBytes.limit(result.capacity());
outWire.write(() -> "RESULT_IS_NULL").bool(false);
outWire.write(() -> "RESULT").bytes(resultBytes);
} else {
outWire.write(() -> "RESULT_IS_NULL").bool(true);
}
key.interestOps(OP_WRITE | OP_READ);
nofityDataWritten();
} catch (Throwable e) {
// todo remove
e.printStackTrace();
// move back to just after the size
outWire.bytes().position(4);
// move back to just after where the size
outWire.bytes().position(locationOfSize + 4);
return sendException(outWire, e);
} finally {
bytesMap.output = null;
writeLength(outWire);
writeLength(outWire, locationOfSize);
}
return null;
}
@Nullable
private Work get() {
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1");
// skip 4 bytes where we will write the size
long locationOfSize = outWire.bytes().position();
outWire.bytes().skip(4);
final BytesChronicleMap bytesMap = bytesMap(channelId);
try {
inLanByteBuffer.clear();
inLangBytes.clear();
// todo
net.openhft.lang.io.Bytes result = bytesMap.get(reader);
outWire.write(() -> "IS_EXCEPTION").bool(false);
if (result != null && result.remaining() > 0) {
BB_ADDRESS.set(resultBuffer, result.address());
BB_CAPACITY.set(resultBuffer, result.capacity());
resultBytes.limit(result.capacity());
outWire.write(() -> "RESULT_IS_NULL").bool(false);
outWire.write(() -> "RESULT").bytes(resultBytes);
} else {
outWire.write(() -> "RESULT_IS_NULL").bool(true);
}
nofityDataWritten();
} catch (Throwable e) {
// move back to just after where the size
outWire.bytes().position(locationOfSize + 4);
return sendException(outWire, e);
} finally {
bytesMap.output = null;
writeLength(outWire, locationOfSize);
}
return null;
}
private void nofityDataWritten() {
key.interestOps(OP_WRITE | OP_READ);
}
}
src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java
+7 -6
View File
@@ -330,9 +330,7 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
assert !hub.inBytesLock().isHeldByCurrentThread();
if (value instanceof CharSequence) {
hub.outWire().write(() -> fieldName).text((CharSequence) value);
System.out.println(hub.outWire().bytes().position() + "fieldName=" + fieldName + ",value=" + value);
} else if (value instanceof Marshallable) {
hub.outWire().write(() -> fieldName).marshallable((Marshallable) value);
} else {
@@ -353,15 +351,19 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
if (usingValue == null)
usingValue = newValueInstance();
if (wireIn.read(() -> "RESULT_IS_NULL").bool())
return null;
if (wireIn.read(() -> "IS_EXCEPTION").bool())
throw new RuntimeException(wireIn.read(() -> "EXCEPTION").text());
if (usingValue instanceof StringBuilder) {
if (wireIn.read(() -> "RESULT_IS_NULL").bool())
return null;
if (StringBuilder.class.isAssignableFrom(vClass)) {
wireIn.read(() -> "RESULT").text((StringBuilder) usingValue);
return usingValue;
} else if (usingValue instanceof Marshallable) {
wireIn.read(() -> "RESULT").marshallable((Marshallable) usingValue);
return usingValue;
} else if (CharSequence.class.isAssignableFrom(vClass)) {
return (V) wireIn.read(() -> "RESULT").text();
} else {
throw new IllegalStateException("unsupported type");
}
@@ -427,7 +429,6 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
}
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBoolean(@NotNull final String methodName, K key, V value1, V value2) {
final long startTime = System.currentTimeMillis();
src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java
+3 -6
View File
@@ -377,7 +377,6 @@ public class WiredStatelessClientTcpConnectionHub {
if (position > Integer.MAX_VALUE || position < Integer.MIN_VALUE)
throw new IllegalStateException("message too large");
System.out.println("writeLength len=" + position);
outWire.bytes().writeInt(sizeMark, (int) position);
}
@@ -431,12 +430,10 @@ public class WiredStatelessClientTcpConnectionHub {
intWire.bytes().skip(4);
intWire.bytes().limit(messageSize);
System.out.println("trans="+Bytes.toHex(intWire.bytes()));
long trans = intWire.read(() -> "TRANSACTION_ID").int64();
System.out.println(trans);
long transactionId0 = 0;
System.out.println("transactionId0=" + transactionId0);
long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64();
// check the transaction id is reasonable
/* assert transactionId0 > 1410000000000L * TcpReplicator.TIMESTAMP_FACTOR :
"TransactionId too small " + transactionId0 + " messageSize " + messageSize;
src/test/java/net/openhft/chronicle/wire/TestWire.java
+53 -12
View File
@@ -18,19 +18,17 @@
package net.openhft.chronicle.wire;
import junit.framework.Assert;
import net.openhft.chronicle.bytes.BytesStore;
import net.openhft.chronicle.hash.replication.ReplicationHub;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.WiredChronicleMapStatelessClientBuilder;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import org.jetbrains.annotations.NotNull;
import org.junit.Test;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import static java.nio.ByteBuffer.allocate;
import static java.util.concurrent.TimeUnit.SECONDS;
@@ -55,6 +53,7 @@ public class TestWire {
builder.identifier(identifier);
return builder.create();
}
private ChronicleMap<byte[], byte[]> map2a;
private ChronicleMap<byte[], byte[]> map3a;
@@ -63,7 +62,7 @@ public class TestWire {
@Test
public void test() throws IOException {
public void testPutStringWithChannels() throws IOException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
@@ -75,34 +74,35 @@ public class TestWire {
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 2)).create();
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
map2a = of(byte[].class, byte[].class)
.entries(1000)
.instance().replicatedViaChannel(hubA.createChannel((short) 3)).create();
.instance().replicatedViaChannel(hubA.createChannel((short) 2)).create();
map3a = of(byte[].class, byte[].class)
.entries(1000)
.instance().replicatedViaChannel(hubA.createChannel((short) 4)).create();
.instance().replicatedViaChannel(hubA.createChannel((short) 3)).create();
byte[] key = new byte[4];
Bytes keyBuffer = new ByteBufferBytes(ByteBuffer.wrap(key));
short channelID = (short) 2;
short channelID = (short) 1;
byte identifier = (byte) 2;
try (ChronicleMap<String, String> statelessMap = localClient(8086, identifier, String.class, String.class, channelID)) {
for (int i = 0; i < 10; i++) {
try {
statelessMap.put("Hello", "World");
statelessMap.put("Hello" + i, "World");
} catch (Throwable t) {
t.printStackTrace();
}
}
assertEquals(10, statelessMap.size());
assertEquals(10, map1a.size());
// check that the bytes maps is correct
assertEquals("World", new String(map1a.get("Hello1".getBytes())));
}
@@ -115,6 +115,47 @@ public class TestWire {
}
@Test
public void testPutMarshanleWithChannels() throws IOException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(8086)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
map2a = of(byte[].class, byte[].class)
.entries(1000)
.instance().replicatedViaChannel(hubA.createChannel((short) 2)).create();
map3a = of(byte[].class, byte[].class)
.entries(1000)
.instance().replicatedViaChannel(hubA.createChannel((short) 3)).create();
byte[] key = new byte[4];
short channelID = (short) 1;
byte identifier = (byte) 2;
ChronicleMap<String, Details> statelessMap = localClient(8086, identifier, String.class, Details.class, (byte) 1);
statelessMap.put("FirstMap", new Details(String.class, String.class, (byte) 1));
final Details firstMap = statelessMap.get("FirstMap");
Assert.assertEquals(1, firstMap.channelID);
}
}
@Test
public void testName() throws Exception {