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

HCOLL-312 work in progress updating map to use wire for the stateless client.

Browse Source
This commit is contained in:
Rob Austin
2015-03-05 10:04:51 +00:00
parent ce0013b9fa
commit 56220303d0
3 changed files with 84 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java
+43 -7
View File
@@ -26,6 +26,7 @@ import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.bytes.BytesMarshallable;
import net.openhft.chronicle.core.OS;
import net.openhft.chronicle.wire.TextWire;
import net.openhft.chronicle.wire.ValueIn;
import net.openhft.chronicle.wire.Wire;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.DirectStore;
@@ -386,17 +387,49 @@ 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);
});
ValueIn read = wire.read(() -> field);
long fieldLength = read.readLength();
long endPos = wire.bytes().position() + fieldLength;
long limit = wire.bytes().limit();
try {
final Bytes source = wire.bytes();
source.limit(endPos);
// write the size
bytes.writeStopBit(source.remaining());
System.out.println(Bytes.toDebugString(source));
while (source.remaining() > 0) {
if (source.remaining() >= 8)
bytes.writeLong(source.readLong());
else
bytes.writeByte(source.readByte());
}
} finally {
wire.bytes().position(endPos);
wire.bytes().limit(limit);
}
}
return bytes.flip();
}
private char peekCode(Bytes<?> bytes) {
if (bytes.remaining() < 1)
return (char) -1;
long pos = bytes.position();
return (char) bytes.readUnsignedByte(pos);
}
@Nullable
private Work sendException(@NotNull Wire wire, @NotNull Throwable e) {
final StringWriter sw = new StringWriter();
@@ -489,7 +522,8 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
// the size of the result in bytes
long len = outChronBytes.readStopBit();
outWire.write(() -> "RESULT").bytes(outChronBytes);
outWire.write(() -> "RESULT");
outWire.bytes().write(outChronBytes);
} else {
outWire.write(() -> "EXCEPTION").bytes(outChronBytes);
}
@@ -560,8 +594,10 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
// the size of the result in bytes
long len = outChronBytes.readStopBit();
outWire.write(() -> "RESULT");
outWire.write(() -> "RESULT").bytes(outChronBytes);
System.out.println("get() ->" + Bytes.toDebugString(outChronBytes));
outWire.bytes().write(outChronBytes);
} else {
outWire.write(() -> "EXCEPTION").bytes(outChronBytes);
}
src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java
+8 -14
View File
@@ -18,9 +18,6 @@
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;
@@ -336,6 +333,9 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
hub.outWire().write(() -> fieldName).text((CharSequence) value);
} else if (value instanceof Marshallable) {
hub.outWire().write(() -> fieldName).marshallable((Marshallable) value);
// System.out.println("HERE="+Bytes.toDebugString( hub.outWire().bytes().flip()));
// System.out.println("NOTHING");
} else {
throw new IllegalStateException("type=" + value.getClass() + " is unsupported, it must either be of type Marshallable or CharSequence");
}
@@ -356,6 +356,8 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
if (wireIn.read(() -> "RESULT_IS_NULL").bool())
return null;
if (StringBuilder.class.isAssignableFrom(vClass)) {
wireIn.read(() -> "RESULT").text((StringBuilder) usingValue);
return usingValue;
@@ -371,20 +373,12 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
}
wireIn.read(() -> "RESULT").marshallable((Marshallable) usingValue);
return usingValue;
} else if (usingValue instanceof StringBuilder) {
// todo optomize
} else if (String.class.isAssignableFrom(vClass)) {
//noinspection unchecked
return (V) wireIn.read(() -> "RESULT").text();
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) new String(wireIn.read(() -> "RESULT").bytes());
} else {
throw new IllegalStateException("unsupported type");
}
src/test/java/net/openhft/chronicle/wire/TestWire.java
+33 -14
View File
@@ -24,7 +24,6 @@ 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;
@@ -94,9 +93,8 @@ public class TestWire {
}
@Ignore
@Test
public void testPutMarshanleWithChannels() throws IOException, InterruptedException {
public void testPutMarshallableWithChannels() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
@@ -120,19 +118,20 @@ public class TestWire {
byte identifier = (byte) 2;
final ChronicleMap<String, Details> statelessMap = localClient(8086, identifier, String.class, Details.class, (byte) 1);
final WiredChronicleMapStatelessClientBuilder<String, Details> builder = new WiredChronicleMapStatelessClientBuilder<String, Details>(new InetSocketAddress("localhost", 8086), String.class, Details.class, (short) (byte) 1);
builder.identifier(identifier);
builder.putReturnsNull(true);
final ChronicleMap<String, Details> statelessMap = builder.create();
final Details details = new Details();
details.keyClass = String.class;
details.valueClass = String.class;
final Details expected = new Details();
expected.keyClass = String.class;
expected.valueClass = String.class;
statelessMap.put("FirstMap", details);
statelessMap.put("FirstMap", expected);
final Details firstMap = statelessMap.get("FirstMap");
final Details actual = statelessMap.get("FirstMap");
Thread.sleep(100);
Assert.assertNotNull(map1a.get("FirstMap".getBytes()));
Assert.assertEquals(1, firstMap.channelID);
Assert.assertEquals(expected, actual);
}
@@ -145,8 +144,6 @@ public class TestWire {
wire.write(() -> "hello").text("world");
System.out.println(wire.bytes());
}
public static class Details implements Marshallable {
@@ -173,5 +170,27 @@ public class TestWire {
throw new RuntimeException(e);
}
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Details details = (Details) o;
if (channelID != details.channelID) return false;
if (!keyClass.equals(details.keyClass)) return false;
if (!valueClass.equals(details.valueClass)) return false;
return true;
}
@Override
public int hashCode() {
int result = keyClass.hashCode();
result = 31 * result + valueClass.hashCode();
result = 31 * result + (int) channelID;
return result;
}
}
}