diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 14c888f0f..73072dccb 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -545,13 +545,13 @@ final class ReplicatedChronicleMap channelList, byte localIdentifier) { this.channelList = channelList; @@ -153,7 +154,7 @@ class StatelessWiredConnector" + inWire.bytes().remaining()); - while (inWire.bytes().remaining() > 4) { final long limit = inWire.bytes().limit(); @@ -184,18 +182,8 @@ class StatelessWiredConnector" + outWire.bytes().limit()); assert inWire.bytes().remaining() < 999; inWire.bytes().limit(limit); assert inWire.bytes().remaining() < 999; @@ -203,7 +191,6 @@ class StatelessWiredConnector "TRANSACTION_ID").int64(); timestamp = inWire.read(() -> "TIME_STAMP").int64(); channelId = inWire.read(() -> "CHANNEL_ID").int16(); inWire.read(() -> "METHOD_NAME").text(methodName); @@ -314,13 +298,13 @@ class StatelessWiredConnector Integer.MAX_VALUE) throw new IllegalStateException("position too large"); // write the size - textWire.bytes().writeInt(0, (int) textWire.bytes().position()); + textWire.bytes().writeInt(start, (int) (textWire.bytes().position() - start)); } /** @@ -351,6 +335,7 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(true); wire.write(() -> "EXCEPTION").text(sw.toString()); return null; } @@ -405,8 +390,8 @@ class StatelessWiredConnector "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); } else { outWire.write(() -> "RESULT_IS_NULL").bool(true); } - key.interestOps(OP_WRITE | OP_READ); + nofityDataWritten(); } catch (Throwable e) { - // todo remove - e.printStackTrace(); - - // move back to just after the size - outWire.bytes().position(4); - + // move back to just after where the size + outWire.bytes().position(locationOfSize + 4); return sendException(outWire, e); } finally { bytesMap.output = null; - writeLength(outWire); + writeLength(outWire, locationOfSize); } 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 BytesChronicleMap bytesMap = bytesMap(channelId); + + try { + + 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); + } else { + outWire.write(() -> "RESULT_IS_NULL").bool(true); + } + + nofityDataWritten(); + + } catch (Throwable e) { + + // move back to just after where the size + outWire.bytes().position(locationOfSize + 4); + return sendException(outWire, e); + } finally { + bytesMap.output = null; + writeLength(outWire, locationOfSize); + } + + return null; + } + + + private void nofityDataWritten() { + key.interestOps(OP_WRITE | OP_READ); + } + + } diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java index 4e65e03a4..2a1c64c3d 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java @@ -330,9 +330,7 @@ class WiredStatelessChronicleMap implements ChronicleMap, Cloneable assert !hub.inBytesLock().isHeldByCurrentThread(); if (value instanceof CharSequence) { - hub.outWire().write(() -> fieldName).text((CharSequence) value); - System.out.println(hub.outWire().bytes().position() + "fieldName=" + fieldName + ",value=" + value); } else if (value instanceof Marshallable) { hub.outWire().write(() -> fieldName).marshallable((Marshallable) value); } else { @@ -353,15 +351,19 @@ class WiredStatelessChronicleMap implements ChronicleMap, Cloneable if (usingValue == null) usingValue = newValueInstance(); - if (wireIn.read(() -> "RESULT_IS_NULL").bool()) - return null; + if (wireIn.read(() -> "IS_EXCEPTION").bool()) + throw new RuntimeException(wireIn.read(() -> "EXCEPTION").text()); - if (usingValue instanceof StringBuilder) { + 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) { wireIn.read(() -> "RESULT").marshallable((Marshallable) usingValue); return usingValue; + } else if (CharSequence.class.isAssignableFrom(vClass)) { + return (V) wireIn.read(() -> "RESULT").text(); } else { throw new IllegalStateException("unsupported type"); } @@ -427,7 +429,6 @@ class WiredStatelessChronicleMap implements ChronicleMap, Cloneable } - @SuppressWarnings("SameParameterValue") private boolean proxyReturnBoolean(@NotNull final String methodName, K key, V value1, V value2) { final long startTime = System.currentTimeMillis(); diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java index 823ce7b67..932427f16 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java @@ -377,7 +377,6 @@ public class WiredStatelessClientTcpConnectionHub { if (position > Integer.MAX_VALUE || position < Integer.MIN_VALUE) throw new IllegalStateException("message too large"); - System.out.println("writeLength len=" + position); outWire.bytes().writeInt(sizeMark, (int) position); } @@ -431,12 +430,10 @@ public class WiredStatelessClientTcpConnectionHub { intWire.bytes().skip(4); intWire.bytes().limit(messageSize); - System.out.println("trans="+Bytes.toHex(intWire.bytes())); - long trans = intWire.read(() -> "TRANSACTION_ID").int64(); - System.out.println(trans); - long transactionId0 = 0; - System.out.println("transactionId0=" + transactionId0); + long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64(); + + // check the transaction id is reasonable /* assert transactionId0 > 1410000000000L * TcpReplicator.TIMESTAMP_FACTOR : "TransactionId too small " + transactionId0 + " messageSize " + messageSize; diff --git a/src/test/java/net/openhft/chronicle/wire/TestWire.java b/src/test/java/net/openhft/chronicle/wire/TestWire.java index ceed64b65..8a05975be 100644 --- a/src/test/java/net/openhft/chronicle/wire/TestWire.java +++ b/src/test/java/net/openhft/chronicle/wire/TestWire.java @@ -18,19 +18,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 net.openhft.lang.io.ByteBufferBytes; -import net.openhft.lang.io.Bytes; import org.jetbrains.annotations.NotNull; import org.junit.Test; import java.io.IOException; import java.net.InetSocketAddress; -import java.nio.ByteBuffer; import static java.nio.ByteBuffer.allocate; import static java.util.concurrent.TimeUnit.SECONDS; @@ -55,6 +53,7 @@ public class TestWire { builder.identifier(identifier); return builder.create(); } + private ChronicleMap map2a; private ChronicleMap map3a; @@ -63,7 +62,7 @@ public class TestWire { @Test - public void test() throws IOException { + public void testPutStringWithChannels() throws IOException { { final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig @@ -75,34 +74,35 @@ 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) 2)).create(); + .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); map2a = of(byte[].class, byte[].class) .entries(1000) - .instance().replicatedViaChannel(hubA.createChannel((short) 3)).create(); + .instance().replicatedViaChannel(hubA.createChannel((short) 2)).create(); map3a = of(byte[].class, byte[].class) .entries(1000) - .instance().replicatedViaChannel(hubA.createChannel((short) 4)).create(); + .instance().replicatedViaChannel(hubA.createChannel((short) 3)).create(); byte[] key = new byte[4]; - Bytes keyBuffer = new ByteBufferBytes(ByteBuffer.wrap(key)); - - short channelID = (short) 2; + short channelID = (short) 1; byte identifier = (byte) 2; try (ChronicleMap statelessMap = localClient(8086, identifier, String.class, String.class, channelID)) { for (int i = 0; i < 10; i++) { try { - statelessMap.put("Hello", "World"); + statelessMap.put("Hello" + i, "World"); } catch (Throwable t) { t.printStackTrace(); } } - assertEquals(10, statelessMap.size()); + assertEquals(10, map1a.size()); + + // check that the bytes maps is correct + assertEquals("World", new String(map1a.get("Hello1".getBytes()))); } @@ -115,6 +115,47 @@ public class TestWire { } + + @Test + public void testPutMarshanleWithChannels() throws IOException { + + { + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .of(8086) + .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(); + + 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; + + + ChronicleMap statelessMap = localClient(8086, identifier, String.class, Details.class, (byte) 1); + statelessMap.put("FirstMap", new Details(String.class, String.class, (byte) 1)); + + final Details firstMap = statelessMap.get("FirstMap"); + Assert.assertEquals(1, firstMap.channelID); + + } + + } + @Test public void testName() throws Exception {