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

CE-20 code clean up

Browse Source
This commit is contained in:
Rob Austin
2015-04-06 18:51:55 +01:00
parent a0c22f8795
commit 272f20f15e
3 changed files with 60 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+31 -65
View File
@@ -275,12 +275,9 @@ class ClientWiredStatelessChronicleMap<K, V>
proxyReturnVoid(
() -> {
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
hub.outWire().writeEventName(createChannel);
hub.outWire().writeValue().int16(channelID);
}
hub.outWire().writeDocument(false, wireOut -> {
hub.outWire().writeEventName(createChannel);
hub.outWire().writeValue().int16(channelID);
});
});
@@ -364,12 +361,7 @@ class ClientWiredStatelessChronicleMap<K, V>
}
public void clear() {
proxyReturnVoid(() -> hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.writeEventName(clear);
}
}
proxyReturnVoid(() -> hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(clear)
));
}
@@ -761,35 +753,32 @@ class ClientWiredStatelessChronicleMap<K, V>
assert args.length > 0;
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
hub.outWire().writeDocument(false, wireOut -> {
final ValueOut out = wireOut.writeEventName(methodName);
final ValueOut out = wireOut.writeEventName(methodName);
final MapWireHandler.Params[] paramNames = methodName.params();
final MapWireHandler.Params[] paramNames = methodName.params();
if (paramNames.length == 1) {
writeField(out, args[0]);
return;
}
if (paramNames.length == 1) {
writeField(out, args[0]);
return;
}
assert args.length == paramNames.length :
"methodName=" + methodName +
", args.length=" + args.length +
", paramNames.length=" + paramNames.length;
assert args.length == paramNames.length :
"methodName=" + methodName +
", args.length=" + args.length +
", paramNames.length=" + paramNames.length;
out.marshallable(m -> {
out.marshallable(m -> {
for (int i = 0; i < paramNames.length; i++) {
final ValueOut vo = m.write(paramNames[i]);
writeField(vo, args[i]);
}
});
}
for (int i = 0; i < paramNames.length; i++) {
final ValueOut vo = m.write(paramNames[i]);
writeField(vo, args[i]);
}
});
}
);
}
@@ -816,12 +805,7 @@ class ClientWiredStatelessChronicleMap<K, V>
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.writeEventName(() -> methodName);
}
}
hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(() -> methodName)
);
@@ -842,12 +826,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.writeEventName(() -> methodName);
}
}
hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(() -> methodName)
);
hub.writeSocket(hub.outWire());
} finally {
@@ -887,12 +866,7 @@ class ClientWiredStatelessChronicleMap<K, V>
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.writeEventName(() -> methodName);
}
}
hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(() -> methodName)
);
hub.writeSocket(hub.outWire());
@@ -977,12 +951,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.writeEventName(methodName);
}
}
hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(methodName)
);
hub.writeSocket(hub.outWire());
@@ -1001,13 +970,10 @@ class ClientWiredStatelessChronicleMap<K, V>
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().write(Fields.eventName).text(methodName.toString());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
final ValueOut out = wireOut.writeEventName(methodName);
writeField(out, arg1);
}
}
hub.outWire().writeDocument(false, wireOut -> {
final ValueOut out = wireOut.writeEventName(methodName);
writeField(out, arg1);
}
);
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java
+9 -27
View File
@@ -359,11 +359,9 @@ public class ClientWiredStatelessTcpConnectionHub {
// reads just the size
readSocket(SIZE_OF_SIZE, timeoutTime);
// System.out.println("in=" + Bytes.toHex(intWire.bytes()));
final int messageSize = bytes.readUnsignedShort(bytes.position());
System.out.println("messageSize=" + messageSize);
try {
assert messageSize > 0 : "Invalid message size " + messageSize;
assert messageSize < 1024 : "Invalid message size " + messageSize;
@@ -378,18 +376,13 @@ public class ClientWiredStatelessTcpConnectionHub {
bytes.limit(bytes.position() + messageSize);
try {
System.out.println("\n--------------------------------\nclient reads\n" +
Wires.fromSizePrefixedBlobs(bytes));
} catch (Exception e) {
System.out.println("\n--------------------------------\nclient reads\n" +
"client read:\n\n" + Bytes.toDebugString(bytes));
}
System.out.println("client read:\n\n" + Bytes.toDebugString(bytes));
int headerlen = bytes.readVolatileInt();
@@ -633,12 +626,7 @@ public class ClientWiredStatelessTcpConnectionHub {
outBytesLock().lock();
try {
long tid = writeHeader(startTime, channelID, wire);
wire.writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.writeEventName(methodName);
}
});
wire.writeDocument(false, wireOut -> wireOut.writeEventName(methodName));
writeSocket(wire);
return tid;
@@ -699,20 +687,14 @@ public class ClientWiredStatelessTcpConnectionHub {
startTime(startTime);
long tid = nextUniqueTransaction(startTime);
wire.writeDocument(true, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(csp).text("MAP");
wireOut.write(MapWireHandlerBuilder.Fields.tid).int64(tid);
wireOut.write(timeStamp).int64(startTime);
wireOut.write(channelId).int16(channelID);
}
wire.writeDocument(true, wireOut -> {
wireOut.write(csp).text("MAP");
wireOut.write(MapWireHandlerBuilder.Fields.tid).int64(tid);
wireOut.write(timeStamp).int64(startTime);
wireOut.write(channelId).int16(channelID);
});
final Bytes<?> bytes = wire.bytes();
System.out.println(Bytes.toDebugString(wire.bytes(), 0, bytes.position()));
return tid;
return tid;
}
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+20 -59
View File
@@ -66,6 +66,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
private final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory;
private Wire inWire = null;
private Wire outWire = null;
private final Consumer writeElement = new Consumer<Iterator<byte[]>>() {
@Override
@@ -75,6 +76,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
bytes.write(iterator.next());
}
};
private final Consumer writeEntry = new Consumer<Iterator<Map.Entry<byte[], byte[]>>>() {
@Override
@@ -93,7 +95,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
};
private WireHandlers publishLater;
private byte localIdentifier;
private long timestamp;
private short channelId;
private List<Replica> channels;
private ReplicationHub hub;
@@ -201,7 +203,9 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
// it is assumed by this point that the buffer has all the bytes in it for this message
final long tid = inWire.read(Fields.tid).int64();
timestamp = inWire.read(timeStamp).int64();
// timestamp
inWire.read(timeStamp).int64();
channelId = inWire.read(Fields.channelId).int16();
final Bytes<?> bytes1 = inWire.bytes();
@@ -330,8 +334,6 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
MapWireHandler.this.writeValue(b -> {
final byte[] result = b.put(key1, value1);
System.out.println("key='" + new String(b.keySet().iterator().next()) + "'");
return result;
});
@@ -442,17 +444,11 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
"server writes:\n\n<EMPTY>");
} else {
try {
System.out.println("--------------------------------------------\n" +
"server writes:\n\n" +
Wires.fromSizePrefixedBlobs(bytes, SIZE_OF_SIZE, len));
//Bytes.toDebugString(bytes, SIZE_OF_SIZE, len));
} catch (Exception e) {
System.out.println("--------------------------------------------\n" +
System.out.println("--------------------------------------------\n" +
"server writes:\n\n" +
Bytes.toDebugString(bytes, SIZE_OF_SIZE, len));
}
Wires.fromSizePrefixedBlobs(bytes, SIZE_OF_SIZE, len));
}
}
}
@@ -460,18 +456,6 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
}
// todo remove
private byte[] toByteArray(net.openhft.lang.io.Bytes bytes) {
if (bytes == null || bytes.remaining() == 0)
return new byte[]{};
if (bytes.remaining() > Integer.MAX_VALUE)
throw new BufferOverflowException();
byte[] result = new byte[(int) bytes.remaining()];
bytes.write(result);
return result;
}
// todo remove
private byte[] toBytes(WireKey fieldName) {
@@ -616,39 +600,25 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
final net.openhft.lang.io.Bytes bytes = DirectStore.allocate(inSize + 1024).bytes();
valueIn.marshallable(new ReadMarshallable() {
@Override
public void readMarshallable(WireIn wire) throws IllegalStateException {
for (Params p : params) {
valueIn.marshallable(wire -> {
final ValueIn read = wire.read(p);
final Bytes<?> bytes2 = read.wireIn().bytes();
final byte[] data = read.bytes();
System.out.println("out='" + new String(data) + "'");
bytes.writeStopBit(data.length);
bytes.write(data);
}
for (Params p : params) {
final ValueIn read = wire.read(p);
final byte[] data = read.bytes();
bytes.writeStopBit(data.length);
bytes.write(data);
}
});
return bytes.flip();
}
/**
* creates a lang buffer that holds just the payload of the args
*
* @param wire the inbound wire
* @return a new lang buffer containing the bytes of the args
*/
// todo remove this method - just added to get it to work for now
private byte[] toByteArray(@NotNull Wire wire, @NotNull Params field) {
final ValueIn read = wire.read(field);
return toByteArray(read);
}
private byte[] toByteArray(ValueIn valueIn) {
@@ -839,15 +809,6 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
}
/**
* only used for debugging
*/
@SuppressWarnings("UnusedDeclaration")
private void showOutWire() {
Bytes<?> bytes = outWire.bytes();
System.out.println("pos=" + bytes.position() + ",bytes=" +
Bytes.toDebugString(bytes, 0, bytes.position()));
}
/**
* converts the exception into a String, so that it can be sent to c# clients
@@ -863,7 +824,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
this.localIdentifier = localIdentifier;
}
static enum Params implements WireKey {
enum Params implements WireKey {
key,
value,
@@ -871,7 +832,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
newValue
}
static enum EventId implements WireKey {
enum EventId implements WireKey {
longSize,
size,