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

Merge remote-tracking branch 'origin/wire-stateless-client' into wire-stateless-client

Browse Source
This commit is contained in:
Rob Austin
2015-03-11 09:47:42 +00:00
parent 594fb0ca88 d4be5c7a83
commit 0dce39d3c6
4 changed files with 222 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java
+73 -66
View File
@@ -30,6 +30,7 @@ import net.openhft.chronicle.hash.replication.ReplicationHub;
import net.openhft.chronicle.wire.TextWire;
import net.openhft.chronicle.wire.ValueIn;
import net.openhft.chronicle.wire.Wire;
import net.openhft.chronicle.wire.WireKey;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.DirectStore;
import net.openhft.lang.io.IByteBufferBytes;
@@ -60,16 +61,35 @@ import static java.nio.channels.SelectionKey.OP_WRITE;
class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarshallable> {
private static final Logger LOG = LoggerFactory.getLogger(StatelessWiredConnector.class);
static enum Fields implements WireKey {
HAS_NEXT,
TIME_STAMP,
CHANNEL_ID,
METHOD_NAME,
TRANSACTION_ID,
RESULT,
RESULT_KEY,
RESULT_VALUE,
ARG_1,
ARG_2,
ARG_3,
IS_EXCEPTION,
EXCEPTION,
RESULT_IS_NULL
}
private final NativeBytes inLangBytes = new NativeBytes(0, 0);
@NotNull
private final OutMessageAdapter outMessageAdapter = new OutMessageAdapter();
@NotNull
private final ArrayList<BytesChronicleMap> bytesChronicleMaps = new ArrayList<>();
private final TextWire inWire = new TextWire(Bytes.elasticByteBuffer());
private final TextWire outWire = new TextWire(Bytes.elasticByteBuffer());
//private final ByteBuffer resultBuffer = ByteBuffer.allocateDirect(64);
@NotNull
private final ByteBuffer inLanByteBuffer = ByteBuffer.allocateDirect(64);
private final StringBuilder methodName = new StringBuilder();
@@ -80,7 +100,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
private long timestamp;
private short channelId;
@NotNull
private List<Replica> channelList;
private ReplicationHub hub;
@@ -88,7 +107,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
@Override
public void accept(Iterator<byte[]> iterator) {
outWire.write(() -> "RESULT");
outWire.write(Fields.RESULT);
outWire.bytes().write(iterator.next());
}
};
@@ -100,16 +119,14 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
final Map.Entry<byte[], byte[]> entry = iterator.next();
outWire.write(() -> "RESULT_KEY");
outWire.write(Fields.RESULT_KEY);
outWire.bytes().write(entry.getKey());
outWire.write(() -> "RESULT_VALUE");
outWire.write(Fields.RESULT_VALUE);
outWire.bytes().write(entry.getValue());
}
};
private Runnable in = null;
private byte remoteIdentifier;
private byte localIdentifer;
private Runnable out = null;
@@ -123,10 +140,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
private SelectionKey key;
public void setChannelList(@NotNull List<Replica> channelList) {
if (channelList == null)
throw new IllegalStateException("The StatelessWiredConnector currently only support maps " +
"that are set up via the replication channel. localIdentifier=" + localIdentifier);
this.channelList = channelList;
}
@@ -283,12 +296,12 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
out = () -> {
// each chunk has its own transaction-id
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
outWire.write(Fields.TRANSACTION_ID).int64(transactionId);
write(map -> {
boolean hasNext = iterator.hasNext();
outWire.write(() -> "HAS_NEXT").bool(hasNext);
outWire.write(Fields.HAS_NEXT).bool(hasNext);
if (hasNext)
c.accept(iterator);
@@ -307,14 +320,14 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
void onEvent() {
// it is assumed by this point that the buffer has all the bytes in it for this message
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);
long transactionId = inWire.read(Fields.TRANSACTION_ID).int64();
timestamp = inWire.read(Fields.TIME_STAMP).int64();
channelId = inWire.read(Fields.CHANNEL_ID).int16();
inWire.read(Fields.METHOD_NAME).text(methodName);
if ("PUT_WITHOUT_ACC".contentEquals(methodName)) {
writeVoid(bytesMap -> {
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2");
final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2);
bytesMap.put(reader, reader, timestamp, identifier());
});
return;
@@ -346,14 +359,13 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
return;
}
// write the transaction id
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
if ("CREATE_CHANNEL".contentEquals(methodName)) {
writeVoid(() -> {
short channelId1 = inWire.read(() -> "ARG_1").int16();
short channelId1 = inWire.read(Fields.ARG_1).int16();
chronicleHashInstanceBuilder.replicatedViaChannel(hub.createChannel(channelId1)).create();
return null;
});
@@ -362,47 +374,47 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
if ("REMOTE_IDENTIFIER".contentEquals(methodName)) {
this.remoteIdentifier = inWire.read(() -> "RESULT").int8();
this.remoteIdentifier = inWire.read(Fields.RESULT).int8();
return;
}
if ("LONG_SIZE".contentEquals(methodName)) {
write(b -> outWire.write(() -> "RESULT").int64(b.longSize()));
write(b -> outWire.write(Fields.RESULT).int64(b.longSize()));
return;
}
if ("IS_EMPTY".contentEquals(methodName)) {
write(b -> outWire.write(() -> "RESULT").bool(b.isEmpty()));
write(b -> outWire.write(Fields.RESULT).bool(b.isEmpty()));
return;
}
if ("CONTAINS_KEY".contentEquals(methodName)) {
write(b -> outWire.write(() -> "RESULT").
bool(b.containsKey(toReader(inWire, "ARG_1"))));
write(b -> outWire.write(Fields.RESULT).
bool(b.containsKey(toReader(inWire, Fields.ARG_1))));
return;
}
if ("CONTAINS_VALUE".contentEquals(methodName)) {
write(b -> outWire.write(() -> "RESULT").
bool(b.containsKey(toReader(inWire, "ARG_1"))));
write(b -> outWire.write(Fields.RESULT).
bool(b.containsKey(toReader(inWire, Fields.ARG_1))));
return;
}
if ("GET".contentEquals(methodName)) {
writeValue(b -> b.get(toReader(inWire, "ARG_1")));
writeValue(b -> b.get(toReader(inWire, Fields.ARG_1)));
return;
}
if ("PUT".contentEquals(methodName)) {
writeValue(bytesMap -> {
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2");
final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2);
bytesMap.put(reader, reader, timestamp, identifier());
});
return;
}
if ("REMOVE".contentEquals(methodName)) {
writeValue(b -> b.remove(toReader(inWire, "ARG_1")));
writeValue(b -> b.remove(toReader(inWire, Fields.ARG_1)));
return;
}
@@ -413,7 +425,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
if ("REPLACE".contentEquals(methodName)) {
writeValue(bytesMap -> {
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2");
final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2);
bytesMap.replace(reader, reader);
});
return;
@@ -421,7 +433,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
if ("REPLACE_WITH_OLD_AND_NEW_VALUE".contentEquals(methodName)) {
writeValue(bytesMap -> {
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2", "ARG_3");
final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2, Fields.ARG_3);
bytesMap.replace(reader, reader, reader);
});
return;
@@ -429,7 +441,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
if ("PUT_IF_ABSENT".contentEquals(methodName)) {
writeValue(bytesMap -> {
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2");
final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2);
bytesMap.putIfAbsent(reader, reader, timestamp, identifier());
});
return;
@@ -437,42 +449,35 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
if ("REMOVE_WITH_VALUE".contentEquals(methodName)) {
writeValue(bytesMap -> {
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2");
final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2);
bytesMap.remove(reader, reader, timestamp, identifier());
});
return;
}
if ("TO_STRING".contentEquals(methodName)) {
write(b -> outWire.write(() -> "RESULT").text(b.toString()));
write(b -> outWire.write(Fields.RESULT).text(b.toString()));
return;
}
if ("APPLICATION_VERSION".contentEquals(methodName)) {
write(b -> outWire.write(() -> "RESULT").text(applicationVersion()));
write(b -> outWire.write(Fields.RESULT).text(applicationVersion()));
return;
}
if ("PERSISTED_DATA_VERSION".contentEquals(methodName)) {
write(b -> outWire.write(() -> "RESULT").text(persistedDataVersion()));
write(b -> outWire.write(Fields.RESULT).text(persistedDataVersion()));
return;
}
if ("HASH_CODE".contentEquals(methodName)) {
write(b -> outWire.write(() -> "RESULT").int32(b.hashCode()));
write(b -> outWire.write(Fields.RESULT).int32(b.hashCode()));
return;
}
throw new IllegalStateException("unsupported event=" + methodName);
// else if ("MAP_FOR_KEY".contentEquals(methodName)
// else if ("PUT_MAPPED".contentEquals(methodName))
// else if ("VALUE_BUILDER".contentEquals(methodName))
} finally {
@@ -491,10 +496,12 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
}
private byte[] toBytes(String fieldName) {
private byte[] toBytes(WireKey fieldName) {
Wire wire = inWire;
ValueIn read = wire.read(() -> fieldName);
System.out.println(Bytes.toDebugString(inWire.bytes()));
ValueIn read = wire.read(fieldName);
long l = read.readLength();
if (l > Integer.MAX_VALUE)
@@ -540,18 +547,18 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
@Override
public void run() {
if (inWire.read(() -> "HAS_NEXT").bool()) {
collectData.put(toBytes("ARG_1"), toBytes("ARG_2"));
if (inWire.read(Fields.HAS_NEXT).bool()) {
collectData.put(toBytes(Fields.ARG_1), toBytes(Fields.ARG_2));
} else {
// the old code assumed that all the data would fit into a single buffer
// this assumption is invalid
if (!collectData.isEmpty()) {
bytesMap.delegate.putAll((Map) collectData);
incompleteWork.remove(transactionId);
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
outWire.write(Fields.TRANSACTION_ID).int64(transactionId);
// todo handle the case where there is an exception
outWire.write(() -> "IS_EXCEPTION").bool(false);
outWire.write(Fields.IS_EXCEPTION).bool(false);
nofityDataWritten();
}
@@ -590,15 +597,15 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
* @param args the key names of the {@code wire} args
* @return a new lang buffer containing the bytes of the args
*/
private net.openhft.lang.io.Bytes toReader(@NotNull Wire wire, @NotNull String... args) {
private net.openhft.lang.io.Bytes toReader(@NotNull Wire wire, @NotNull WireKey... args) {
long inSize = wire.bytes().limit();
final net.openhft.lang.io.Bytes bytes = DirectStore.allocate(inSize).bytes();
// copy the bytes to the reader
for (final String field : args) {
for (final WireKey field : args) {
ValueIn read = wire.read(() -> field);
ValueIn read = wire.read(field);
long fieldLength = read.readLength();
long endPos = wire.bytes().position() + fieldLength;
@@ -639,7 +646,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
@NotNull
private ReplicatedChronicleMap map(short channelId) {
// todo this cast is a bit of a hack, improve later
final ReplicatedChronicleMap replicas =
(ReplicatedChronicleMap) channelList.get(channelId);
@@ -701,8 +707,8 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
// the idea of wire is that is platform independent,
// so we wil have to send the exception as a String
outWire.write(() -> "IS_EXCEPTION").bool(true);
outWire.write(() -> "EXCEPTION").text(toString(e));
outWire.write(Fields.IS_EXCEPTION).bool(true);
outWire.write(Fields.EXCEPTION).text(toString(e));
LOG.error("", e);
return;
}
@@ -725,7 +731,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
bytesMap.output = null;
outWire.bytes().mark();
outWire.write(() -> "IS_EXCEPTION").bool(false);
outWire.write(Fields.IS_EXCEPTION).bool(false);
try {
c.accept(bytesMap);
@@ -733,8 +739,8 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
outWire.bytes().reset();
// the idea of wire is that is platform independent,
// so we wil have to send the exception as a String
outWire.write(() -> "IS_EXCEPTION").bool(true);
outWire.write(() -> "EXCEPTION").text(toString(e));
outWire.write(Fields.IS_EXCEPTION).bool(true);
outWire.write(Fields.EXCEPTION).text(toString(e));
LOG.error("", e);
return;
}
@@ -755,7 +761,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
bytesMap.output = null;
outWire.bytes().mark();
outWire.write(() -> "IS_EXCEPTION").bool(false);
outWire.write(Fields.IS_EXCEPTION).bool(false);
try {
r.call();
@@ -763,8 +769,8 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
outWire.bytes().reset();
// the idea of wire is that is platform independent,
// so we wil have to send the exception as a String
outWire.write(() -> "IS_EXCEPTION").bool(true);
outWire.write(() -> "EXCEPTION").text(toString(e));
outWire.write(Fields.IS_EXCEPTION).bool(true);
outWire.write(Fields.EXCEPTION).text(toString(e));
LOG.error("", e);
return;
}
@@ -825,6 +831,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
*/
private static class OutMessageAdapter {
@NotNull
private final IByteBufferBytes outLangBytes = ByteBufferBytes.wrap(ByteBuffer.allocate(1024));
private Bytes outChronBytes = Bytes.wrap(outLangBytes.buffer());
@@ -881,7 +888,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
*/
void accept(@NotNull Wire wire) {
wire.write(() -> "IS_EXCEPTION").bool(false);
wire.write(Fields.IS_EXCEPTION).bool(false);
// flips calls flip on this message so that we can read it
flipMessage();
@@ -896,10 +903,10 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
// read the size - not used
outChronBytes.readStopBit();
wire.write(() -> "RESULT_IS_NULL").bool(isNull);
wire.write(Fields.RESULT_IS_NULL).bool(isNull);
if (!isNull) {
// write the result
wire.write(() -> "RESULT");
wire.write(Fields.RESULT);
wire.bytes().write(outChronBytes);
}
}
src/main/java/net/openhft/chronicle/map/WiredChronicleMapStatelessClientBuilder.java
+8 -4
View File
@@ -47,6 +47,13 @@ public final class WiredChronicleMapStatelessClientBuilder<K, V> implements
this.channelID = channelID;
}
public WiredChronicleMapStatelessClientBuilder(WiredStatelessClientTcpConnectionHub hub, Class keyClass, Class valueClass, short channelID) {
this.keyClass = keyClass;
this.valueClass = valueClass;
this.channelID = channelID;
}
// This method is meaningful, despite you could call of().create(), because Java 7 doesn't
// infer type parameters if case of chained calls, like this; and then you couldn't omit class
// name. Then it is
@@ -56,8 +63,7 @@ public final class WiredChronicleMapStatelessClientBuilder<K, V> implements
// -- the second is really shorter.
private final InetSocketAddress remoteAddress;
private InetSocketAddress remoteAddress;
private boolean putReturnsNull = false;
private boolean removeReturnsNull = false;
private long timeoutMs = TimeUnit.SECONDS.toMillis(10);
@@ -67,7 +73,6 @@ public final class WiredChronicleMapStatelessClientBuilder<K, V> implements
private final AtomicBoolean used = new AtomicBoolean(false);
InetSocketAddress remoteAddress() {
return remoteAddress;
}
@@ -149,7 +154,6 @@ public final class WiredChronicleMapStatelessClientBuilder<K, V> implements
}
public WiredStatelessClientTcpConnectionHub hub() {
return hub;
}
src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java
+67 -73
View File
@@ -44,16 +44,6 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
private static final Logger LOG = LoggerFactory.getLogger(WiredStatelessChronicleMap.class);
// todo replace with an ENUM
public static final String METHOD_NAME = "METHOD_NAME";
public static final String RESULT_IS_NULL = "RESULT_IS_NULL";
public static final String HAS_NEXT = "HAS_NEXT";
public static final String RESULT = "RESULT";
public static final String ARG_1 = "ARG_1";
public static final String ARG_2 = "ARG_2";
private final WiredStatelessClientTcpConnectionHub hub;
protected Class<K> kClass;
@@ -270,15 +260,14 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key);
}
@NotNull
public void createChannel(short channelID) {
proxyReturnVoid(() -> {
writeField(METHOD_NAME, "CREATE_CHANNEL");
hub.outWire().write(() -> "ARG_1").int16(channelID);
});
proxyReturnVoid(
() -> {
writeField(StatelessWiredConnector.Fields.METHOD_NAME, "CREATE_CHANNEL");
hub.outWire().write(StatelessWiredConnector.Fields.ARG_1).int16(channelID);
});
}
public void putAll(@NotNull Map<? extends K, ? extends V> map) {
final long startTime = System.currentTimeMillis();
@@ -305,18 +294,18 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.markSize(wire);
hub.startTime(startTime);
wire.write(() -> "TRANSACTION_ID").int64(transactionId);
wire.write(() -> "TIME_STAMP").int64(startTime);
wire.write(() -> "CHANNEL_ID").int16(channelID);
wire.write(StatelessWiredConnector.Fields.TRANSACTION_ID).int64(transactionId);
wire.write(StatelessWiredConnector.Fields.TIME_STAMP).int64(startTime);
wire.write(StatelessWiredConnector.Fields.CHANNEL_ID).int16(channelID);
writeField(METHOD_NAME, "PUT_ALL");
writeField(StatelessWiredConnector.Fields.METHOD_NAME, "PUT_ALL");
hub.outWire().write(() -> "HAS_NEXT").bool(hasNext);
hub.outWire().write(StatelessWiredConnector.Fields.HAS_NEXT).bool(hasNext);
if (hasNext) {
Map.Entry<? extends K, ? extends V> e = iterator.next();
writeField(ARG_1, e.getKey());
writeField(ARG_2, e.getValue());
writeField(StatelessWiredConnector.Fields.ARG_1, e.getKey());
writeField(StatelessWiredConnector.Fields.ARG_2, e.getValue());
}
hub.writeSocket(hub.outWire());
@@ -351,7 +340,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
}
public void clear() {
proxyReturnVoid(() -> writeField(METHOD_NAME, CLEAR.toString()));
proxyReturnVoid(() -> writeField(StatelessWiredConnector.Fields.METHOD_NAME, CLEAR.toString()));
}
@@ -371,15 +360,12 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(() -> HAS_NEXT).bool()) {
K k = readK(RESULT, wireIn, null);
if (wireIn.read(StatelessWiredConnector.Fields.HAS_NEXT).bool()) {
K k = readK(StatelessWiredConnector.Fields.RESULT, wireIn, null);
result.add(k);
} else
break;
// return result;
} finally {
hub.inBytesLock().unlock();
}
@@ -404,8 +390,8 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(() -> HAS_NEXT).bool()) {
V v = readV(RESULT, wireIn, null);
if (wireIn.read(StatelessWiredConnector.Fields.HAS_NEXT).bool()) {
V v = readV(StatelessWiredConnector.Fields.RESULT, wireIn, null);
result.add(v);
} else
break;
@@ -434,10 +420,10 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(() -> HAS_NEXT).bool()) {
if (wireIn.read(StatelessWiredConnector.Fields.HAS_NEXT).bool()) {
result.put(
readK("RESULT_KEY", wireIn, null),
readV("RESULT_VALUE", wireIn, null));
readK(StatelessWiredConnector.Fields.RESULT_KEY, wireIn, null),
readV(StatelessWiredConnector.Fields.RESULT_VALUE, wireIn, null));
} else
break;
@@ -463,33 +449,43 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
// receive
hub.inBytesLock().lock();
try {
return hub.proxyReply(timeoutTime, transactionId).read(() -> RESULT).int64();
return hub.proxyReply(timeoutTime, transactionId).read(StatelessWiredConnector.Fields.RESULT).int64();
} finally {
hub.inBytesLock().unlock();
}
}
private void writeField(String fieldName, Object value) {
private void writeField(StatelessWiredConnector.Fields fieldName, Object value) {
writeField(fieldName, value, hub.outWire());
}
private void writeField(String fieldName, Object value, Wire wire) {
private void writeField(StatelessWiredConnector.Fields fieldName, Object value, Wire wire) {
assert hub.outBytesLock().isHeldByCurrentThread();
assert !hub.inBytesLock().isHeldByCurrentThread();
if (value instanceof CharSequence) {
wire.write(() -> fieldName).text((CharSequence) value);
if (value instanceof Byte)
wire.write(fieldName).int8((Byte) value);
else if (value instanceof Character)
wire.write(fieldName).text(((Character) value).toString());
else if (value instanceof Short)
wire.write(fieldName).int16((Short) value);
else if (value instanceof Integer)
wire.write(fieldName).int32((Integer) value);
else if (value instanceof Long)
wire.write(fieldName).int64((Long) value);
else if (value instanceof CharSequence) {
wire.write(fieldName).text((CharSequence) value);
} else if (value instanceof Marshallable) {
wire.write(() -> fieldName).marshallable((Marshallable) value);
wire.write(fieldName).marshallable((Marshallable) value);
} else {
throw new IllegalStateException("type=" + value.getClass() + " is unsupported, it must either be of type Marshallable or CharSequence");
}
}
private V readValue(String argName, long transactionId, long startTime, final V usingValue) {
private V readValue(StatelessWiredConnector.Fields argName, long transactionId, long startTime, final V usingValue) {
assert !hub.outBytesLock().isHeldByCurrentThread();
long timeoutTime = startTime + hub.timeoutMs;
@@ -498,7 +494,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(() -> RESULT_IS_NULL).bool())
if (wireIn.read(StatelessWiredConnector.Fields.RESULT_IS_NULL).bool())
return null;
return readV(argName, wireIn, usingValue);
@@ -508,54 +504,54 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
}
}
private V readV(String argName, Wire wireIn, V usingValue) {
private V readV(StatelessWiredConnector.Fields argName, Wire wireIn, V usingValue) {
if (StringBuilder.class.isAssignableFrom(vClass)) {
wireIn.read(() -> argName).text((StringBuilder) usingValue);
wireIn.read(argName).text((StringBuilder) usingValue);
return usingValue;
} else if (Marshallable.class.isAssignableFrom(vClass)) {
if (usingValue == null)
try {
V v = vClass.newInstance();
wireIn.read(() -> argName).marshallable((Marshallable) v);
wireIn.read(argName).marshallable((Marshallable) v);
return v;
} catch (Exception e) {
throw new RuntimeException(e);
}
wireIn.read(() -> argName).marshallable((Marshallable) usingValue);
wireIn.read(argName).marshallable((Marshallable) usingValue);
return usingValue;
} else if (String.class.isAssignableFrom(vClass)) {
//noinspection unchecked
return (V) wireIn.read(() -> argName).text();
return (V) wireIn.read(argName).text();
} else {
throw new IllegalStateException("unsupported type");
}
}
private K readK(String argName, Wire wireIn, K usingValue) {
private K readK(StatelessWiredConnector.Fields argName, Wire wireIn, K usingValue) {
if (StringBuilder.class.isAssignableFrom(kClass)) {
wireIn.read(() -> argName).text((StringBuilder) usingValue);
wireIn.read(argName).text((StringBuilder) usingValue);
return usingValue;
} else if (Marshallable.class.isAssignableFrom(kClass)) {
if (usingValue == null)
try {
K v = kClass.newInstance();
wireIn.read(() -> argName).marshallable((Marshallable) v);
wireIn.read(argName).marshallable((Marshallable) v);
return v;
} catch (Exception e) {
throw new RuntimeException(e);
}
wireIn.read(() -> argName).marshallable((Marshallable) usingValue);
wireIn.read(argName).marshallable((Marshallable) usingValue);
return usingValue;
} else if (String.class.isAssignableFrom(vClass)) {
//noinspection unchecked
return (K) wireIn.read(() -> argName).text();
return (K) wireIn.read(argName).text();
} else {
throw new IllegalStateException("unsupported type");
@@ -571,7 +567,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
return wireIn.read(() -> RESULT).bool();
return wireIn.read(StatelessWiredConnector.Fields.RESULT).bool();
} finally {
hub.inBytesLock().unlock();
}
@@ -586,7 +582,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
return wireIn.read(() -> RESULT).int32();
return wireIn.read(StatelessWiredConnector.Fields.RESULT).int32();
} finally {
hub.inBytesLock().unlock();
}
@@ -601,9 +597,9 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(METHOD_NAME, methodName);
writeField(ARG_1, key);
writeField(ARG_2, value);
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.ARG_1, key);
writeField(StatelessWiredConnector.Fields.ARG_2, value);
hub.writeSocket(hub.outWire());
} finally {
@@ -622,9 +618,9 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(METHOD_NAME, methodName);
writeField(ARG_1, value1);
writeField(ARG_2, value2);
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.ARG_1, value1);
writeField(StatelessWiredConnector.Fields.ARG_2, value2);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -656,7 +652,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -674,7 +670,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -712,7 +708,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -722,8 +718,6 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
}
@Nullable
private <R> R proxyReturnObject(@NotNull final String methodName, K key, V value, Class<V> resultType) {
@@ -736,9 +730,9 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
assert !hub.inBytesLock().isHeldByCurrentThread();
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(METHOD_NAME, methodName);
writeField(ARG_1, key);
writeField(ARG_2, value);
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.ARG_1, key);
writeField(StatelessWiredConnector.Fields.ARG_2, value);
hub.writeSocket(hub.outWire());
} finally {
@@ -749,7 +743,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
return null;
if (resultType == vClass)
return (R) readValue(RESULT, transactionId, startTime, null);
return (R) readValue(StatelessWiredConnector.Fields.RESULT, transactionId, startTime, null);
else
throw new UnsupportedOperationException("class of type class=" + resultType + " is not " +
@@ -768,7 +762,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
return null;
if (rClass == vClass)
return (R) readValue(RESULT, transactionId, startTime, null);
return (R) readValue(StatelessWiredConnector.Fields.RESULT, transactionId, startTime, null);
else
throw new UnsupportedOperationException("class of type class=" + rClass + " is not " +
"supported");
@@ -779,8 +773,8 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(METHOD_NAME, methodName);
writeField(ARG_1, arg1);
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.ARG_1, arg1);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -794,7 +788,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
src/test/java/net/openhft/chronicle/wire/TestWire.java → src/test/java/net/openhft/chronicle/map/TestWire.java
+74 -9
View File
@@ -16,14 +16,15 @@
* limitations under the License.
*/
package net.openhft.chronicle.wire;
package net.openhft.chronicle.map;
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.chronicle.map.WiredStatelessChronicleMap;
import net.openhft.chronicle.wire.Marshallable;
import net.openhft.chronicle.wire.WireIn;
import net.openhft.chronicle.wire.WireOut;
import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
@@ -39,7 +40,6 @@ import static org.junit.Assert.*;
*/
public class TestWire {
public static int SERVER_PORT = 8086;
private ChronicleMap<byte[], byte[]> map1a;
private ChronicleMap<byte[], byte[]> map2a;
@@ -47,6 +47,62 @@ public class TestWire {
private ReplicationHub hubA;
@Test
public void testUsingDetails() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.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();
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, Details> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
Details.class,
(short) 1);
builder.identifier(identifier);
final ChronicleMap<String, Details> detailsMap = builder.create();
Details details = detailsMap.get("Colours");
// todo add some sort of cross node locking
if (details == null) {
short channelID = (short) 2;
((WiredStatelessChronicleMap) detailsMap).createChannel(channelID);
details = new Details(String.class, String.class, channelID);
detailsMap.put("Colours", details);
}
ChronicleMap<String, String> colours = new WiredChronicleMapStatelessClientBuilder<String, String>(
builder.hub(),
details.keyClass,
details.valueClass,
details.channelID).hub(builder.hub()).create();
colours.put("Rob", "Blue");
assertEquals(1, colours.size());
Details result = detailsMap.get("Colours");
Assert.assertEquals(2, result.channelID);
map1a.close();
}
}
@Test
public void testCreateChannel() throws IOException, InterruptedException {
@@ -61,8 +117,6 @@ 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) 1)).create();
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, String> builder =
@@ -351,8 +405,8 @@ public class TestWire {
Set<Map.Entry<String, String>> entries = statelessMap.entrySet();
assertEquals(2, values.size());
assertTrue(entries.contains(Collections.singletonMap("Key1","Value1").entrySet().iterator().next()));
assertTrue(entries.contains(Collections.singletonMap("Key2","Value2").entrySet().iterator().next()));
assertTrue(entries.contains(Collections.singletonMap("Key1", "Value1").entrySet().iterator().next()));
assertTrue(entries.contains(Collections.singletonMap("Key2", "Value2").entrySet().iterator().next()));
map1a.close();
@@ -420,6 +474,17 @@ public class TestWire {
short channelID;
public Details() {
}
public Details(Class<String> keyClass, Class<String> valueClass, short channelID) {
this.keyClass = keyClass;
this.valueClass = valueClass;
this.channelID = channelID;
}
@Override
public void writeMarshallable(WireOut wire) {
wire.write(() -> "keyClass").text(keyClass.getName());