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

migrating the code towards the https://github.com/OpenHFT/RFC/blob/master/Chronicle/Map/Remote/Chronicle-Map-0.1.md specification - still work in progress - DON'T REFACTOR

Browse Source
This commit is contained in:
Rob Austin
2015-03-31 18:19:10 +01:00
parent 75ecfb6c7e
commit 92741c26fa
4 changed files with 149 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+123 -26
View File
@@ -23,6 +23,7 @@ import net.openhft.chronicle.hash.function.SerializableFunction;
import net.openhft.chronicle.map.MapWireHandlerBuilder.Fields;
import net.openhft.chronicle.wire.Marshallable;
import net.openhft.chronicle.wire.Wire;
import net.openhft.chronicle.wire.WireOut;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
@@ -275,8 +276,15 @@ class ClientWiredStatelessChronicleMap<K, V>
public void createChannel(short channelID) {
proxyReturnVoid(
() -> {
hub.outWire().write(Fields.methodName).text(createChannel.toString());
hub.outWire().write(Fields.arg1).int16(channelID);
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
hub.outWire().write(Fields.methodName).text(createChannel.toString());
hub.outWire().write(Fields.arg1).int16(channelID);
}
});
});
}
@@ -301,7 +309,7 @@ class ClientWiredStatelessChronicleMap<K, V>
assert hub.outBytesLock().isHeldByCurrentThread();
hub.markSize(wire);
hub.startTime(startTime);
wire.write(Fields.type).text("MAP");
wire.write(Fields.csp).text("MAP");
wire.write(Fields.tid).int64(tid);
wire.write(Fields.timeStamp).int64(startTime);
wire.write(Fields.channelId).int16(channelID);
@@ -313,8 +321,10 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outWire().write(Fields.hasNext).bool(iterator.hasNext());
writeField(Fields.arg1, e.getKey());
writeField(Fields.arg2, e.getValue());
// todo
// writeField(wireOut, Fields.arg1, e.getKey());
// writeField(wireOut, Fields.arg2, e.getValue());
final Bytes<?> bytes = wire.bytes();
if (bytes.remaining() < bytes.capacity() * 0.5)
@@ -353,7 +363,15 @@ class ClientWiredStatelessChronicleMap<K, V>
}
public void clear() {
proxyReturnVoid(() -> writeField(Fields.methodName, clear.toString()));
proxyReturnVoid(() -> hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(clear.toString());
}
}
));
}
@@ -525,12 +543,12 @@ class ClientWiredStatelessChronicleMap<K, V>
}
private void writeField(Fields fieldName, Object value) {
writeField(fieldName, value, hub.outWire());
private void writeField(WireOut wireOut, Fields fieldName, Object value) {
writeField(fieldName, value, wireOut);
}
private void writeField(Fields fieldName, Object value, Wire wire) {
private void writeField(Fields fieldName, Object value, WireOut wire) {
assert hub.outBytesLock().isHeldByCurrentThread();
assert !hub.inBytesLock().isHeldByCurrentThread();
@@ -681,9 +699,17 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().write(Fields.methodName).text(methodName.toString());
writeField(Fields.arg1, key);
writeField(Fields.arg2, value);
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
writeField(wireOut, Fields.arg1, key);
writeField(wireOut, Fields.arg2, value);
}
}
);
hub.writeSocket(hub.outWire());
} finally {
@@ -703,10 +729,18 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().write(Fields.methodName).text(methodName.toString());
writeField(Fields.arg1, key);
writeField(Fields.arg2, value1);
writeField(Fields.arg3, value2);
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
writeField(wireOut, Fields.arg1, key);
writeField(wireOut, Fields.arg2, value1);
writeField(wireOut, Fields.arg3, value2);
}
}
);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -738,7 +772,16 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().write(Fields.methodName).text(methodName.toString());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
}
}
);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -756,7 +799,13 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().write(Fields.methodName).text(methodName.toString());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
}
}
);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -794,7 +843,15 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().write(Fields.methodName).text(methodName.toString());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
}
}
);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -817,9 +874,19 @@ class ClientWiredStatelessChronicleMap<K, V>
assert !hub.inBytesLock().isHeldByCurrentThread();
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().write(Fields.methodName).text(methodName.toString());
writeField(Fields.arg1, key);
writeField(Fields.arg2, value);
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
writeField(wireOut, Fields.arg1, key);
writeField(wireOut, Fields.arg2, value);
}
}
);
hub.writeSocket(hub.outWire());
} finally {
@@ -861,8 +928,19 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().write(Fields.methodName).text(methodName.toString());
writeField(Fields.arg1, arg1);
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
writeField(wireOut, Fields.arg1, arg1);
}
}
);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -876,8 +954,17 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().write(Fields.methodName).text(methodName.toString());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
}
}
);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
@@ -890,7 +977,17 @@ class ClientWiredStatelessChronicleMap<K, V>
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().write(Fields.methodName).text(methodName.toString());
hub.outWire().write(Fields.arg1).int64(arg1);
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
writeField(wireOut, Fields.arg1, arg1);
}
}
);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java
+17 -3
View File
@@ -27,8 +27,8 @@ import net.openhft.chronicle.map.MapWireHandler.EventId;
import net.openhft.chronicle.network2.event.EventGroup;
import net.openhft.chronicle.wire.TextWire;
import net.openhft.chronicle.wire.Wire;
import net.openhft.chronicle.wire.WireIn;
import net.openhft.chronicle.wire.WireOut;
import net.openhft.chronicle.wire.Wires;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
@@ -373,9 +373,14 @@ public class ClientWiredStatelessTcpConnectionHub {
System.out.println("\n--------------------------------\n" +
"client read:\n\n" + Bytes.toDebugString(bytes));
int headerlen = bytes.readVolatileInt();
assert !Wires.isData(headerlen);
long tid0 = intWire.read(MapWireHandlerBuilder.Fields.tid).int64();
// check the transaction id is reasonable
/* assert tid0 > 1410000000000L * TcpReplicator.TIMESTAMP_FACTOR :
"TransactionId too small " + tid0 + " messageSize " + messageSize;
@@ -593,7 +598,13 @@ public class ClientWiredStatelessTcpConnectionHub {
outBytesLock().lock();
try {
long tid = writeHeader(startTime, channelID, wire);
wire.write(MapWireHandlerBuilder.Fields.methodName).text(methodName.toString());
wire.writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(MapWireHandlerBuilder.Fields.methodName).text(methodName.toString());
}
});
writeSocket(wire);
return tid;
} finally {
@@ -649,13 +660,16 @@ public class ClientWiredStatelessTcpConnectionHub {
wire.writeDocument(true, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(type).text("MAP");
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;
}
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+8 -1
View File
@@ -204,6 +204,11 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
timestamp = inWire.read(timeStamp).int64();
channelId = inWire.read(Fields.channelId).int16();
final Bytes<?> bytes1 = inWire.bytes();
final int body = bytes1.readVolatileInt();
assert Wires.isData(body);
final StringBuilder methodName = Wires.acquireStringBuilder();
inWire.read(Fields.methodName).text(methodName);
@@ -252,8 +257,10 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
return;
}
outWire.writeDocument(true, wire -> outWire.write(Fields.tid).int64(tid));
// write the transaction id
outWire.write(Fields.tid).int64(tid);
if (createChannel.contentEquals(methodName)) {
src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java
+1 -1
View File
@@ -44,7 +44,7 @@ public class MapWireHandlerBuilder {
timeStamp,
channelId,
methodName,
type,
csp,
tid, // transactionId
result,
resultKey,