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 10:28:23 +00:00
parent 0c755817c1
commit e28d980c61
3 changed files with 45 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java
+27 -19
View File
@@ -153,8 +153,8 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
public void onRead(SocketChannel socketChannel, SelectionKey key) throws IOException {
this.key = key;
if (!handshingComplete)
outWire.bytes().writeByte(localIdentifier);
// if (!handshingComplete)
// outWire.bytes().writeByte(localIdentifier);
final int len = readSocket(socketChannel);
assert inWire.bytes().remaining() < 999;
@@ -184,16 +184,16 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
return;
}
if (!handshingComplete) {
System.out.println("doHandshaking");
// if (!handshingComplete) {
System.out.println("doHandshaking");
onHandShaking();
System.out.println("end-doHandshaking");
} else {
System.out.println("onEvent");
onEvent();
System.out.println("end-onEvent");
}
// 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;
@@ -279,7 +279,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
}
void onHandShaking() {
remoteIdentifier = inWire.read(() -> "IDENTIFIER").int8();
// remoteIdentifier = inWire.read(() -> "IDENTIFIER").int8();
handshingComplete = true;
}
@@ -287,21 +287,29 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
Work onEvent() {
// 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();
timestamp = inWire.read(() -> "TIME_STAMP").int64();
channelId = inWire.read(() -> "CHANNEL_ID").int16();
inWire.read(() -> "METHOD_NAME").text(methodName);
// for the length
long markStart = outWire.bytes().position();
outWire.bytes().skip(4);
// write the transaction id
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
try {
// write the transaction id
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
if ("PUT".contentEquals(methodName))
if ("PUT".contentEquals(methodName))
return put();
else
throw new IllegalStateException("unsupported event=" + methodName);
else
throw new IllegalStateException("unsupported event=" + methodName);
} finally {
int len = (int) (outWire.bytes().position() - markStart);
outWire.bytes().writeInt(markStart, len);
}
}
@@ -436,8 +444,8 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
// todo remove
e.printStackTrace();
// move back to the start
outWire.bytes().position(2);
// move back to just after the size
outWire.bytes().position(4);
return sendException(outWire, e);
} finally {
src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java
+17 -16
View File
@@ -39,7 +39,6 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE;
import static net.openhft.chronicle.map.AbstractChannelReplicator.openSocketChannel;
import static net.openhft.chronicle.map.WiredStatelessChronicleMap.EventId.APPLICATION_VERSION;
/**
@@ -186,7 +185,7 @@ public class WiredStatelessClientTcpConnectionHub {
}
clientChannel = result;
ByteBuffer byteBuffer = ByteBuffer.allocateDirect(128);
/* ByteBuffer byteBuffer = ByteBuffer.allocateDirect(128);
Bytes<ByteBuffer> bufferBytes = Bytes.wrap(byteBuffer);
TextWire wire = new TextWire(bufferBytes);
wire.bytes().skip(4);
@@ -221,7 +220,7 @@ public class WiredStatelessClientTcpConnectionHub {
//
}
}
}*/
}
@@ -328,7 +327,7 @@ public class WiredStatelessClientTcpConnectionHub {
@SuppressWarnings("WeakerAccess")
@NotNull
String clientVersion() {
String clientVersion() {
return BuildVersion.version();
}
@@ -336,7 +335,7 @@ public class WiredStatelessClientTcpConnectionHub {
/**
* sends data to the server via TCP/IP
*/
void writeSocket() {
void writeSocket() {
assert outBytesLock().isHeldByCurrentThread();
assert !inBytesLock().isHeldByCurrentThread();
@@ -430,11 +429,13 @@ public class WiredStatelessClientTcpConnectionHub {
final int remainingBytes0 = messageSize - SIZE_OF_SIZE;
readSocket(remainingBytes0, timeoutTime);
intWire.bytes().skip(4);
intWire.bytes().skip(4);
intWire.bytes().limit(messageSize);
String text = intWire.read(() -> "TRANSACTION_ID").text();
System.out.println(text);
long transactionId0 =0;
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);
// check the transaction id is reasonable
/* assert transactionId0 > 1410000000000L * TcpReplicator.TIMESTAMP_FACTOR :
@@ -633,8 +634,8 @@ public class WiredStatelessClientTcpConnectionHub {
@SuppressWarnings("SameParameterValue")
@Nullable
<O extends Marshallable> O fetchObject(final Class<O> tClass,
@NotNull final String methodName, short channelID) {
<O extends Marshallable> O fetchObject(final Class<O> tClass,
@NotNull final String methodName, short channelID) {
final long startTime = System.currentTimeMillis();
long transactionId;
@@ -663,7 +664,7 @@ public class WiredStatelessClientTcpConnectionHub {
@SuppressWarnings("SameParameterValue")
@Nullable
String proxyReturnString(@NotNull final String messageId, short channelID) {
String proxyReturnString(@NotNull final String messageId, short channelID) {
final long startTime = System.currentTimeMillis();
long transactionId;
@@ -712,11 +713,11 @@ public class WiredStatelessClientTcpConnectionHub {
/**
* mark the location of the outWire size
*/
void markSize() {
void markSize() {
assert outBytesLock().isHeldByCurrentThread();
assert outBytesLock().isHeldByCurrentThread();
// this is where the size will go
// this is where the size will go
sizeMark = outWire.bytes().position();
@@ -724,7 +725,7 @@ public class WiredStatelessClientTcpConnectionHub {
outWire.bytes().skip(4);
}
void startTime(long startTime) {
void startTime(long startTime) {
this.startTime = startTime;
}
}
src/test/java/net/openhft/chronicle/wire/TestWire.java
+1 -2
View File
@@ -26,7 +26,6 @@ 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.Ignore;
import org.junit.Test;
import java.io.IOException;
@@ -62,7 +61,7 @@ public class TestWire {
private ReplicationHub hubA;
@Ignore
@Test
public void test() throws IOException {