From 56220303d034a40532e8d836403ffb121cd6ce21 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 5 Mar 2015 10:04:51 +0000 Subject: [PATCH] HCOLL-312 work in progress updating map to use wire for the stateless client. --- .../map/StatelessWiredConnector.java | 50 ++++++++++++++++--- .../map/WiredStatelessChronicleMap.java | 22 +++----- .../net/openhft/chronicle/wire/TestWire.java | 47 +++++++++++------ 3 files changed, 84 insertions(+), 35 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java index 15349c15f..b86bb8d3c 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java @@ -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 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 "RESULT").bytes(outChronBytes); + outWire.write(() -> "RESULT"); + outWire.bytes().write(outChronBytes); } else { outWire.write(() -> "EXCEPTION").bytes(outChronBytes); } @@ -560,8 +594,10 @@ class StatelessWiredConnector "RESULT"); - outWire.write(() -> "RESULT").bytes(outChronBytes); + System.out.println("get() ->" + Bytes.toDebugString(outChronBytes)); + outWire.bytes().write(outChronBytes); } else { outWire.write(() -> "EXCEPTION").bytes(outChronBytes); } diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java index 32cf229ed..767458ed5 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java @@ -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 implements ChronicleMap, 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 implements ChronicleMap, 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 implements ChronicleMap, 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"); } diff --git a/src/test/java/net/openhft/chronicle/wire/TestWire.java b/src/test/java/net/openhft/chronicle/wire/TestWire.java index 986ce82fe..77f12b908 100644 --- a/src/test/java/net/openhft/chronicle/wire/TestWire.java +++ b/src/test/java/net/openhft/chronicle/wire/TestWire.java @@ -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 statelessMap = localClient(8086, identifier, String.class, Details.class, (byte) 1); + final WiredChronicleMapStatelessClientBuilder builder = new WiredChronicleMapStatelessClientBuilder(new InetSocketAddress("localhost", 8086), String.class, Details.class, (short) (byte) 1); + builder.identifier(identifier); + builder.putReturnsNull(true); + final ChronicleMap 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; + } } }