mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
wire stateless client - still work in progress
This commit is contained in:
Rob Austin
@@ -27,7 +27,9 @@ import net.openhft.chronicle.bytes.BytesMarshallable;
|
||||
import net.openhft.chronicle.core.OS;
|
||||
import net.openhft.chronicle.wire.TextWire;
|
||||
import net.openhft.chronicle.wire.Wire;
|
||||
import net.openhft.lang.io.ByteBufferBytes;
|
||||
import net.openhft.lang.io.DirectStore;
|
||||
import net.openhft.lang.io.IByteBufferBytes;
|
||||
import net.openhft.lang.io.NativeBytes;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
import org.jetbrains.annotations.Nullable;
|
||||
@@ -40,6 +42,7 @@ import java.io.StringWriter;
|
||||
import java.lang.reflect.Field;
|
||||
import java.lang.reflect.Method;
|
||||
import java.nio.Buffer;
|
||||
import java.nio.BufferOverflowException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.SelectionKey;
|
||||
import java.nio.channels.SocketChannel;
|
||||
@@ -88,7 +91,50 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
NATIVE_CAPACITY = nCapacity;
|
||||
}
|
||||
|
||||
|
||||
private final NativeBytes inLangBytes = new NativeBytes(0, 0);
|
||||
|
||||
// private net.openhft.lang.io.Bytes outLangBytes = ByteBufferBytes.wrap(ByteBuffer.allocate(1024));
|
||||
|
||||
private IByteBufferBytes outLangBytes = ByteBufferBytes.wrap(ByteBuffer.allocate(1024));
|
||||
private Bytes outChronBytes = Bytes.wrap(outLangBytes.buffer());
|
||||
|
||||
MapIOBuffer outBuffer = new MapIOBuffer() {
|
||||
|
||||
@Override
|
||||
public void ensureBufferSize(long l) {
|
||||
|
||||
if (outLangBytes.remaining() >= l)
|
||||
return;
|
||||
|
||||
long size = outLangBytes.capacity() + l;
|
||||
if (size > Integer.MAX_VALUE)
|
||||
throw new BufferOverflowException();
|
||||
|
||||
// record the current possition and limit
|
||||
long position = outLangBytes.position();
|
||||
|
||||
// create a new buffer and copy the data into it
|
||||
outLangBytes.clear();
|
||||
final IByteBufferBytes newOutLangBytes = ByteBufferBytes.wrap(ByteBuffer.allocate((int) size));
|
||||
newOutLangBytes.write(outLangBytes);
|
||||
|
||||
outChronBytes = Bytes.wrap(outLangBytes.buffer());
|
||||
|
||||
newOutLangBytes.limit(newOutLangBytes.capacity());
|
||||
newOutLangBytes.position(position);
|
||||
|
||||
outChronBytes.limit(outChronBytes.capacity());
|
||||
outChronBytes.position(position);
|
||||
}
|
||||
|
||||
@Override
|
||||
public net.openhft.lang.io.Bytes in() {
|
||||
return outLangBytes;
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
private final TextWire inWire = new TextWire(Bytes.elasticByteBuffer());
|
||||
|
||||
private boolean handshingComplete;
|
||||
@@ -97,7 +143,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
private ArrayList<BytesChronicleMap> bytesChronicleMaps = new ArrayList<>();
|
||||
private final byte localIdentifier;
|
||||
private final StringBuilder methodName = new StringBuilder();
|
||||
private final ByteBuffer resultBuffer = ByteBuffer.allocateDirect(64);
|
||||
//private final ByteBuffer resultBuffer = ByteBuffer.allocateDirect(64);
|
||||
private ByteBuffer inLanByteBuffer = ByteBuffer.allocateDirect(64);
|
||||
private final MapIOBuffer mapIOBuffer = new MapIOBuffer() {
|
||||
|
||||
@@ -139,7 +185,11 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
// maybe null if the server has not been set up to use channel
|
||||
private List<Replica> channelList;
|
||||
private byte remoteIdentifier;
|
||||
private Bytes resultBytes = Bytes.wrap(resultBuffer);
|
||||
// private Bytes resultBytes = Bytes.wrap(resultBuffer);
|
||||
|
||||
// todo improve later, dont make fixed sie
|
||||
private Bytes resultLangBytes = Bytes.wrap(ByteBuffer.allocate(1024));
|
||||
|
||||
private SelectionKey key;
|
||||
private long transactionId;
|
||||
|
||||
@@ -151,6 +201,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
this.localIdentifier = localIdentifier;
|
||||
}
|
||||
|
||||
|
||||
public void onRead(SocketChannel socketChannel, SelectionKey key) throws IOException {
|
||||
this.key = key;
|
||||
|
||||
@@ -182,11 +233,13 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
onEvent();
|
||||
assert inWire.bytes().remaining() < 999;
|
||||
inWire.bytes().limit(limit);
|
||||
assert inWire.bytes().remaining() < 999;
|
||||
long nextPosition = inWire.bytes().limit();
|
||||
try {
|
||||
onEvent();
|
||||
} finally {
|
||||
inWire.bytes().position(nextPosition);
|
||||
inWire.bytes().limit(limit);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -208,13 +261,17 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
|
||||
public void onWrite(SocketChannel socketChannel, SelectionKey key) throws IOException {
|
||||
|
||||
|
||||
((ByteBuffer) outWire.bytes().underlyingObject()).limit((int) outWire.bytes().position());
|
||||
System.out.println(Bytes.toHex(outWire.bytes().flip()));
|
||||
socketChannel.write((ByteBuffer) outWire.bytes().underlyingObject());
|
||||
|
||||
if (((ByteBuffer) outWire.bytes().underlyingObject()).remaining() == 0) {
|
||||
((ByteBuffer) outWire.bytes().underlyingObject()).clear();
|
||||
outWire.bytes().clear();
|
||||
key.interestOps(OP_READ);
|
||||
outChronBytes.clear();
|
||||
outLangBytes.clear();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -250,7 +307,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
inWire.bytes().ensureCapacity(bytes.position() + size);
|
||||
|
||||
if (bytes.remaining() < size) {
|
||||
|
||||
assert size < 100000;
|
||||
return null;
|
||||
}
|
||||
@@ -278,16 +334,25 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
channelId = inWire.read(() -> "CHANNEL_ID").int16();
|
||||
inWire.read(() -> "METHOD_NAME").text(methodName);
|
||||
|
||||
|
||||
if ("PUT_WITHOUT_ACC".contentEquals(methodName))
|
||||
return putWithOutAcc();
|
||||
|
||||
// for the length
|
||||
long markStart = outWire.bytes().position();
|
||||
outWire.bytes().skip(4);
|
||||
|
||||
try {
|
||||
|
||||
// write the transaction id
|
||||
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
|
||||
|
||||
if ("PUT".contentEquals(methodName))
|
||||
return put();
|
||||
|
||||
else if ("GET".contentEquals(methodName))
|
||||
return get();
|
||||
|
||||
else
|
||||
throw new IllegalStateException("unsupported event=" + methodName);
|
||||
} finally {
|
||||
@@ -322,6 +387,8 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
// copy the bytes to the reader
|
||||
for (final String field : args) {
|
||||
wire.read(() -> field).bytes(source -> {
|
||||
|
||||
System.out.println("field=" + field + ", source='" + new String(source) + "'");
|
||||
bytes.writeStopBit(source.length);
|
||||
bytes.write(source);
|
||||
});
|
||||
@@ -337,6 +404,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
e.printStackTrace(pw);
|
||||
outWire.write(() -> "IS_EXCEPTION").bool(true);
|
||||
wire.write(() -> "EXCEPTION").text(sw.toString());
|
||||
nofityDataWritten();
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -367,7 +435,7 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
private BytesChronicleMap bytesMap(short channelId) {
|
||||
|
||||
final BytesChronicleMap bytesChronicleMap = (channelId < bytesChronicleMaps.size())
|
||||
? bytesChronicleMaps.get(channelId - 1)
|
||||
? bytesChronicleMaps.get(channelId)
|
||||
: null;
|
||||
|
||||
if (bytesChronicleMap != null)
|
||||
@@ -385,15 +453,13 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
|
||||
}
|
||||
|
||||
|
||||
@Nullable
|
||||
private Work put() {
|
||||
|
||||
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2");
|
||||
|
||||
// skip 4 bytes where we will write the size
|
||||
long locationOfSize = outWire.bytes().position();
|
||||
outWire.bytes().skip(4);
|
||||
|
||||
final long start = outWire.bytes().position();
|
||||
final BytesChronicleMap bytesMap = bytesMap(channelId);
|
||||
|
||||
try {
|
||||
@@ -401,44 +467,72 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
inLanByteBuffer.clear();
|
||||
inLangBytes.clear();
|
||||
|
||||
// todo
|
||||
net.openhft.lang.io.Bytes result = bytesMap.put(reader, reader, timestamp, remoteIdentifier);
|
||||
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);
|
||||
bytesMap.output = outBuffer;
|
||||
bytesMap.put(reader, reader, timestamp, remoteIdentifier);
|
||||
|
||||
outLangBytes.flip();
|
||||
|
||||
|
||||
outChronBytes.position(outLangBytes.position());
|
||||
outChronBytes.limit(outLangBytes.limit());
|
||||
|
||||
//todo not sure why we have to skip 1, what is this field
|
||||
outChronBytes.skip(1);
|
||||
|
||||
boolean isException = false;
|
||||
outWire.write(() -> "IS_EXCEPTION").bool(isException);
|
||||
|
||||
if (!isException) {
|
||||
boolean isNull = outChronBytes.readBoolean();
|
||||
outWire.write(() -> "RESULT_IS_NULL").bool(isNull);
|
||||
|
||||
// the size of the result in bytes
|
||||
long len = outChronBytes.readStopBit();
|
||||
|
||||
outWire.write(() -> "RESULT").bytes(outChronBytes);
|
||||
} else {
|
||||
outWire.write(() -> "RESULT_IS_NULL").bool(true);
|
||||
outWire.write(() -> "EXCEPTION").bytes(outChronBytes);
|
||||
}
|
||||
|
||||
|
||||
nofityDataWritten();
|
||||
|
||||
} catch (Throwable e) {
|
||||
|
||||
e.printStackTrace();
|
||||
// move back to just after where the size
|
||||
outWire.bytes().position(locationOfSize + 4);
|
||||
outWire.bytes().position(start);
|
||||
return sendException(outWire, e);
|
||||
} finally {
|
||||
bytesMap.output = null;
|
||||
writeLength(outWire, locationOfSize);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
|
||||
@Nullable
|
||||
private Work putWithOutAcc() {
|
||||
|
||||
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2");
|
||||
|
||||
// skip 4 bytes where we will write the size
|
||||
|
||||
final BytesChronicleMap bytesMap = bytesMap(channelId);
|
||||
|
||||
inLanByteBuffer.clear();
|
||||
inLangBytes.clear();
|
||||
|
||||
bytesMap.output = outBuffer;
|
||||
bytesMap.put(reader, reader, timestamp, remoteIdentifier);
|
||||
|
||||
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 long start = outWire.bytes().position();
|
||||
final BytesChronicleMap bytesMap = bytesMap(channelId);
|
||||
|
||||
try {
|
||||
@@ -446,29 +540,40 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
|
||||
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);
|
||||
bytesMap.output = outBuffer;
|
||||
bytesMap.get(reader);
|
||||
|
||||
// the result of the method above is written here
|
||||
outLangBytes.flip();
|
||||
outChronBytes.position(outLangBytes.position());
|
||||
outChronBytes.limit(outLangBytes.limit());
|
||||
|
||||
//todo not sure why we have to skip 1, what is this field
|
||||
outChronBytes.skip(1);
|
||||
|
||||
boolean isException = false;
|
||||
outWire.write(() -> "IS_EXCEPTION").bool(isException);
|
||||
|
||||
if (!isException) {
|
||||
boolean isNull = outChronBytes.readBoolean();
|
||||
outWire.write(() -> "RESULT_IS_NULL").bool(isNull);
|
||||
|
||||
// the size of the result in bytes
|
||||
long len = outChronBytes.readStopBit();
|
||||
|
||||
outWire.write(() -> "RESULT").bytes(outChronBytes);
|
||||
} else {
|
||||
outWire.write(() -> "RESULT_IS_NULL").bool(true);
|
||||
outWire.write(() -> "EXCEPTION").bytes(outChronBytes);
|
||||
}
|
||||
|
||||
|
||||
nofityDataWritten();
|
||||
|
||||
} catch (Throwable e) {
|
||||
|
||||
e.printStackTrace();
|
||||
// move back to just after where the size
|
||||
outWire.bytes().position(locationOfSize + 4);
|
||||
outWire.bytes().position(start);
|
||||
return sendException(outWire, e);
|
||||
} finally {
|
||||
bytesMap.output = null;
|
||||
writeLength(outWire, locationOfSize);
|
||||
}
|
||||
|
||||
return null;
|
||||
|
||||
@@ -18,6 +18,9 @@
|
||||
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import net.openhft.chronicle.bytes.Bytes;
|
||||
import net.openhft.chronicle.bytes.EscapingStopCharTester;
|
||||
import net.openhft.chronicle.bytes.StopCharTesters;
|
||||
import net.openhft.chronicle.hash.function.SerializableFunction;
|
||||
import net.openhft.chronicle.wire.Marshallable;
|
||||
import net.openhft.chronicle.wire.Wire;
|
||||
@@ -339,7 +342,7 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
|
||||
}
|
||||
|
||||
|
||||
private V readValue(long transactionId, long startTime, V usingValue) {
|
||||
private V readValue(long transactionId, long startTime, final V usingValue) {
|
||||
assert !hub.outBytesLock().isHeldByCurrentThread();
|
||||
long timeoutTime = startTime + hub.timeoutMs;
|
||||
|
||||
@@ -348,22 +351,40 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
|
||||
|
||||
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
|
||||
|
||||
if (usingValue == null)
|
||||
usingValue = newValueInstance();
|
||||
|
||||
if (wireIn.read(() -> "IS_EXCEPTION").bool())
|
||||
throw new RuntimeException(wireIn.read(() -> "EXCEPTION").text());
|
||||
|
||||
if (wireIn.read(() -> "RESULT_IS_NULL").bool())
|
||||
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) {
|
||||
} else if (Marshallable.class.isAssignableFrom(vClass)) {
|
||||
|
||||
if (usingValue == null)
|
||||
try {
|
||||
V v = vClass.newInstance();
|
||||
wireIn.read(() -> "RESULT").marshallable((Marshallable) v);
|
||||
return v;
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
|
||||
wireIn.read(() -> "RESULT").marshallable((Marshallable) usingValue);
|
||||
|
||||
return usingValue;
|
||||
} else if (usingValue instanceof StringBuilder) {
|
||||
|
||||
// todo optomize
|
||||
|
||||
StringBuilder sb = new StringBuilder();
|
||||
Bytes b = Bytes.elasticByteBuffer();
|
||||
wireIn.read(() -> "RESULT").bytes(b);
|
||||
b.parseUTF(sb, EscapingStopCharTester.escaping(StopCharTesters.COMMA_STOP));
|
||||
return (V) sb.toString();
|
||||
|
||||
} else if (CharSequence.class.isAssignableFrom(vClass)) {
|
||||
return (V) wireIn.read(() -> "RESULT").text();
|
||||
return (V) new String(wireIn.read(() -> "RESULT").bytes());
|
||||
} else {
|
||||
throw new IllegalStateException("unsupported type");
|
||||
}
|
||||
@@ -586,6 +607,7 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
|
||||
|
||||
if (resultType == vClass)
|
||||
return (R) readValue(transactionId, startTime, null);
|
||||
|
||||
else
|
||||
throw new UnsupportedOperationException("class of type class=" + resultType + " is not " +
|
||||
"supported");
|
||||
|
||||
@@ -19,18 +19,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 org.jetbrains.annotations.NotNull;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
import static java.nio.ByteBuffer.allocate;
|
||||
import static java.util.concurrent.TimeUnit.SECONDS;
|
||||
import static net.openhft.chronicle.map.ChronicleMapBuilder.of;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
@@ -62,7 +61,7 @@ public class TestWire {
|
||||
|
||||
|
||||
@Test
|
||||
public void testPutStringWithChannels() throws IOException {
|
||||
public void testSimplePutStringWithChannels() throws IOException, InterruptedException {
|
||||
|
||||
{
|
||||
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
|
||||
@@ -76,48 +75,28 @@ public class TestWire {
|
||||
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;
|
||||
try (ChronicleMap<String, String> statelessMap = localClient(8086, identifier, String.class, String.class, channelID)) {
|
||||
|
||||
for (int i = 0; i < 10; i++) {
|
||||
try {
|
||||
statelessMap.put("Hello" + i, "World");
|
||||
} catch (Throwable t) {
|
||||
t.printStackTrace();
|
||||
}
|
||||
}
|
||||
final WiredChronicleMapStatelessClientBuilder<String, String> builder = new WiredChronicleMapStatelessClientBuilder<String, String>(new InetSocketAddress("localhost", 8086), String.class, String.class, channelID);
|
||||
builder.identifier(identifier);
|
||||
builder.putReturnsNull(true);
|
||||
|
||||
assertEquals(10, map1a.size());
|
||||
|
||||
// check that the bytes maps is correct
|
||||
assertEquals("World", new String(map1a.get("Hello1".getBytes())));
|
||||
try (ChronicleMap<String, String> statelessMap = builder.create()) {
|
||||
statelessMap.put("Hello", "World");
|
||||
assertEquals("World", statelessMap.get("Hello"));
|
||||
}
|
||||
|
||||
|
||||
net.openhft.chronicle.bytes.Bytes bytes = BytesStore.wrap(allocate(128)).bytes();
|
||||
TextWire textWire = new TextWire(bytes);
|
||||
textWire.write(() -> "TYPE").text("TextWire");
|
||||
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
@Ignore
|
||||
@Test
|
||||
public void testPutMarshanleWithChannels() throws IOException {
|
||||
public void testPutMarshanleWithChannels() throws IOException, InterruptedException {
|
||||
|
||||
{
|
||||
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
|
||||
@@ -139,17 +118,20 @@ public class TestWire {
|
||||
.entries(1000)
|
||||
.instance().replicatedViaChannel(hubA.createChannel((short) 3)).create();
|
||||
|
||||
|
||||
byte[] key = new byte[4];
|
||||
|
||||
short channelID = (short) 1;
|
||||
byte identifier = (byte) 2;
|
||||
|
||||
final ChronicleMap<String, Details> statelessMap = localClient(8086, identifier, String.class, Details.class, (byte) 1);
|
||||
|
||||
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 details = new Details();
|
||||
details.keyClass = String.class;
|
||||
details.valueClass = String.class;
|
||||
|
||||
statelessMap.put("FirstMap", details);
|
||||
|
||||
final Details firstMap = statelessMap.get("FirstMap");
|
||||
|
||||
Thread.sleep(100);
|
||||
Assert.assertNotNull(map1a.get("FirstMap".getBytes()));
|
||||
Assert.assertEquals(1, firstMap.channelID);
|
||||
|
||||
}
|
||||
@@ -161,7 +143,7 @@ public class TestWire {
|
||||
|
||||
TextWire wire = new TextWire(net.openhft.chronicle.bytes.Bytes.elasticByteBuffer());
|
||||
|
||||
WireOut world = wire.write(() -> "hello").text("world");
|
||||
wire.write(() -> "hello").text("world");
|
||||
System.out.println(wire.bytes());
|
||||
|
||||
|
||||
@@ -173,11 +155,6 @@ public class TestWire {
|
||||
Class valueClass;
|
||||
short channelID;
|
||||
|
||||
public Details(Class keyClass, Class valueClass, short channelID) {
|
||||
this.keyClass = keyClass;
|
||||
this.valueClass = valueClass;
|
||||
this.channelID = channelID;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeMarshallable(WireOut wire) {
|
||||
|
||||
Reference in New Issue
Block a user