diff --git a/pom.xml b/pom.xml index 7a3cc0215..a7d6b963b 100755 --- a/pom.xml +++ b/pom.xml @@ -69,6 +69,11 @@ 1.0.1-alpha-SNAPSHOT + + net.openhft + chronicle-bytes + 1.0.0-alpha-SNAPSHOT + @@ -101,6 +106,11 @@ chronicle-wire + + net.openhft + chronicle-bytes + + diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java index 635e71613..72c092c8b 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java @@ -119,6 +119,7 @@ public final class ReplicationHub extends AbstractReplication { return channel; } + /** * Builder of {@link ReplicationHub}s. */ diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java index c52bfdb75..d298ce2b6 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java @@ -21,7 +21,6 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.function.SerializableFunction; import net.openhft.chronicle.wire.Marshallable; import net.openhft.chronicle.wire.Wire; -import net.openhft.lang.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; @@ -42,7 +41,7 @@ import static net.openhft.chronicle.map.WiredStatelessChronicleMap.EventId.*; /** * @author Rob Austin. */ -class WiredStatelessChronicleMap implements ChronicleMap, Cloneable { +public class WiredStatelessChronicleMap implements ChronicleMap, Cloneable { private static final Logger LOG = LoggerFactory.getLogger(WiredStatelessChronicleMap.class); private final WiredStatelessClientTcpConnectionHub hub; @@ -414,7 +413,7 @@ class WiredStatelessChronicleMap implements ChronicleMap, Cloneable } } - private String readString(long transactionId, long startTime) { +/* private String readString(long transactionId, long startTime) { assert !hub.outBytesLock().isHeldByCurrentThread(); long timeoutTime = startTime + hub.timeoutMs; @@ -427,7 +426,7 @@ class WiredStatelessChronicleMap implements ChronicleMap, Cloneable } finally { hub.inBytesLock().unlock(); } - } + }*/ @SuppressWarnings("SameParameterValue") @@ -622,7 +621,6 @@ class WiredStatelessChronicleMap implements ChronicleMap, Cloneable @Nullable private R proxyReturnObject(Class rClass, @NotNull final String methodName, Object key) { final long startTime = System.currentTimeMillis(); - ThreadLocalCopies copies; long transactionId; diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java index 514ec0f70..ccf32eb17 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java @@ -92,7 +92,7 @@ public class WiredStatelessClientTcpConnectionHub { this.name = config.name(); this.timeoutMs = config.timeoutMs(); attemptConnect(remoteAddress); -// checkVersion(config.channelID()); + checkVersion(config.channelID()); } private synchronized void attemptConnect(final InetSocketAddress remoteAddress) { @@ -264,6 +264,7 @@ public class WiredStatelessClientTcpConnectionHub { /** * sends data to the server via TCP/IP + * * @param wire the {@code wire} containing the outbound data */ void writeSocket(@NotNull final Wire wire) { @@ -375,7 +376,7 @@ public class WiredStatelessClientTcpConnectionHub { intWire.bytes().skip(4); intWire.bytes().limit(messageSize); - long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64(); + long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64(); // check the transaction id is reasonable @@ -555,24 +556,6 @@ public class WiredStatelessClientTcpConnectionHub { } - private long proxySend(@NotNull final String methodName, final long startTime, short channelID) { - - assert outBytesLock().isHeldByCurrentThread(); - assert !inBytesLock().isHeldByCurrentThread(); - - // send - outBytesLock().lock(); - try { - long transactionId = writeHeader(startTime, channelID, outWire()); - outWire.write(() -> "METHOD_NAME").text(methodName); - writeSocket(outWire); - return transactionId; - } finally { - outBytesLock().unlock(); - } - } - - private long proxySend(@NotNull final String methodName, final long startTime, short channelID, @@ -617,7 +600,7 @@ public class WiredStatelessClientTcpConnectionHub { // receive inBytesLock().lock(); try { - return proxyReply(timeoutTime, transactionId).read(() -> messageId).text(); + return proxyReply(timeoutTime, transactionId).read(() -> "RESULT").text(); } catch (Exception e) { throw new RuntimeException(e); } finally { @@ -649,6 +632,7 @@ public class WiredStatelessClientTcpConnectionHub { /** * mark the location of the outWire size + * * @param outWire */ void markSize(Wire outWire) { diff --git a/src/test/java/net/openhft/chronicle/wire/TestWire.java b/src/test/java/net/openhft/chronicle/wire/TestWire.java index 57179784a..7457f834c 100644 --- a/src/test/java/net/openhft/chronicle/wire/TestWire.java +++ b/src/test/java/net/openhft/chronicle/wire/TestWire.java @@ -23,6 +23,7 @@ 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 org.junit.Test; import java.io.IOException; @@ -80,12 +81,51 @@ public class TestWire { assertFalse(statelessMap.isEmpty()); } + map1a.close(); } } + + @Test + public void testVersion() 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(); + + short channelID = (short) 1; + byte identifier = (byte) 2; + + final WiredChronicleMapStatelessClientBuilder builder = + new WiredChronicleMapStatelessClientBuilder<>( + new InetSocketAddress("localhost", SERVER_PORT), + String.class, + String.class, + channelID); + + builder.identifier(identifier); + builder.putReturnsNull(true); + + try (ChronicleMap statelessMap = builder.create()) { + String s = ((WiredStatelessChronicleMap) statelessMap).serverApplicationVersion(); + Assert.assertNotNull(s); + + } + map1a.close(); + + } + + @Test public void testLongSize() throws IOException, InterruptedException { @@ -120,7 +160,7 @@ public class TestWire { assertEquals(1, statelessMap.longSize()); assertFalse(statelessMap.isEmpty()); } - + map1a.close(); } @@ -173,6 +213,9 @@ public class TestWire { Assert.assertEquals(expected, actual); + map1a.close(); + map2a.close(); + map3a.close(); } } @@ -208,13 +251,12 @@ public class TestWire { final String actual = statelessMap.get("FirstMap"); Assert.assertEquals(null, actual); - + map1a.close(); } } - public static class Details implements Marshallable { Class keyClass;