diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java index d27045ff9..4b1b097a1 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java @@ -64,12 +64,13 @@ class StatelessWiredConnector bytesChronicleMaps = new ArrayList<>(); private final TextWire inWire = new TextWire(Bytes.elasticByteBuffer()); private final TextWire outWire = new TextWire(Bytes.elasticByteBuffer()); - //private final ByteBuffer resultBuffer = ByteBuffer.allocateDirect(64); + @NotNull private final ByteBuffer inLanByteBuffer = ByteBuffer.allocateDirect(64); private final StringBuilder methodName = new StringBuilder(); @@ -80,7 +81,6 @@ class StatelessWiredConnector channelList; private ReplicationHub hub; @@ -108,8 +108,6 @@ class StatelessWiredConnector channelList) { - if (channelList == null) - throw new IllegalStateException("The StatelessWiredConnector currently only support maps " + - "that are set up via the replication channel. localIdentifier=" + localIdentifier); - this.channelList = channelList; } @@ -346,7 +340,6 @@ class StatelessWiredConnector "TRANSACTION_ID").int64(transactionId); @@ -445,7 +438,6 @@ class StatelessWiredConnector outWire.write(() -> "RESULT").text(b.toString())); - return; } @@ -468,11 +460,6 @@ class StatelessWiredConnector implements this.channelID = channelID; } + public WiredChronicleMapStatelessClientBuilder(WiredStatelessClientTcpConnectionHub hub, Class keyClass, Class valueClass, short channelID) { + this.keyClass = keyClass; + this.valueClass = valueClass; + + this.channelID = channelID; + } + // This method is meaningful, despite you could call of().create(), because Java 7 doesn't // infer type parameters if case of chained calls, like this; and then you couldn't omit class // name. Then it is @@ -56,8 +63,7 @@ public final class WiredChronicleMapStatelessClientBuilder implements // -- the second is really shorter. - - private final InetSocketAddress remoteAddress; + private InetSocketAddress remoteAddress; private boolean putReturnsNull = false; private boolean removeReturnsNull = false; private long timeoutMs = TimeUnit.SECONDS.toMillis(10); @@ -67,7 +73,6 @@ public final class WiredChronicleMapStatelessClientBuilder implements private final AtomicBoolean used = new AtomicBoolean(false); - InetSocketAddress remoteAddress() { return remoteAddress; } @@ -149,7 +154,6 @@ public final class WiredChronicleMapStatelessClientBuilder implements } - public WiredStatelessClientTcpConnectionHub hub() { return hub; } diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java index ca26ef322..419093d80 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java @@ -44,7 +44,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo private static final Logger LOG = LoggerFactory.getLogger(WiredStatelessChronicleMap.class); - // todo replace with an ENUM public static final String METHOD_NAME = "METHOD_NAME"; public static final String RESULT_IS_NULL = "RESULT_IS_NULL"; @@ -53,7 +52,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo public static final String ARG_1 = "ARG_1"; public static final String ARG_2 = "ARG_2"; - private final WiredStatelessClientTcpConnectionHub hub; protected Class kClass; @@ -270,7 +268,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key); } - @NotNull public void createChannel(short channelID) { proxyReturnVoid(() -> { writeField(METHOD_NAME, "CREATE_CHANNEL"); @@ -278,7 +275,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo }); } - public void putAll(@NotNull Map map) { final long startTime = System.currentTimeMillis(); @@ -377,9 +373,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo } else break; - - // return result; - } finally { hub.inBytesLock().unlock(); } @@ -479,7 +472,17 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo assert hub.outBytesLock().isHeldByCurrentThread(); assert !hub.inBytesLock().isHeldByCurrentThread(); - if (value instanceof CharSequence) { + if (value instanceof Byte) + wire.write(() -> fieldName).int8((Byte) value); + else if (value instanceof Character) + wire.write(() -> fieldName).text(((Character) value).toString()); + else if (value instanceof Short) + wire.write(() -> fieldName).int16((Short) value); + else if (value instanceof Integer) + wire.write(() -> fieldName).int32((Integer) value); + else if (value instanceof Long) + wire.write(() -> fieldName).int64((Long) value); + else if (value instanceof CharSequence) { wire.write(() -> fieldName).text((CharSequence) value); } else if (value instanceof Marshallable) { wire.write(() -> fieldName).marshallable((Marshallable) value); @@ -722,8 +725,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo } - - @Nullable private R proxyReturnObject(@NotNull final String methodName, K key, V value, Class resultType) { diff --git a/src/test/java/net/openhft/chronicle/wire/TestWire.java b/src/test/java/net/openhft/chronicle/map/TestWire.java similarity index 85% rename from src/test/java/net/openhft/chronicle/wire/TestWire.java rename to src/test/java/net/openhft/chronicle/map/TestWire.java index 3861be50d..b53305194 100644 --- a/src/test/java/net/openhft/chronicle/wire/TestWire.java +++ b/src/test/java/net/openhft/chronicle/map/TestWire.java @@ -16,14 +16,15 @@ * limitations under the License. */ -package net.openhft.chronicle.wire; +package net.openhft.chronicle.map; 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 net.openhft.chronicle.wire.Marshallable; +import net.openhft.chronicle.wire.WireIn; +import net.openhft.chronicle.wire.WireOut; +import org.junit.Assert; import org.junit.Test; import java.io.IOException; @@ -39,7 +40,6 @@ import static org.junit.Assert.*; */ public class TestWire { - public static int SERVER_PORT = 8086; private ChronicleMap map1a; private ChronicleMap map2a; @@ -47,6 +47,66 @@ public class TestWire { private ReplicationHub hubA; + @Test + public void testUsingDetails() 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(); + + + byte identifier = (byte) 2; + + + final WiredChronicleMapStatelessClientBuilder builder = + new WiredChronicleMapStatelessClientBuilder<>( + new InetSocketAddress("localhost", SERVER_PORT), + String.class, + Details.class, + (short) 1); + + builder.identifier(identifier); + final ChronicleMap detailsMap = builder.create(); + + + Details details = detailsMap.get("Colours"); + + // todo add some sort of cross node locking + if (details == null) { + short channelID = (short) 2; + ((WiredStatelessChronicleMap) detailsMap).createChannel(channelID); + details = new Details(String.class, String.class, channelID); + detailsMap.put("Colours", details); + } + + + ChronicleMap colours = new WiredChronicleMapStatelessClientBuilder( + builder.hub(), + details.keyClass, + details.valueClass, + details.channelID).hub(builder.hub()).create(); + + colours.put("Rob", "Blue"); + + assertEquals(1, colours.size()); + + + Details result = detailsMap.get("Colours"); + Assert.assertEquals(2, result.channelID); + map1a.close(); + } + + } + + @Test public void testCreateChannel() throws IOException, InterruptedException { @@ -351,8 +411,8 @@ public class TestWire { Set> entries = statelessMap.entrySet(); assertEquals(2, values.size()); - assertTrue(entries.contains(Collections.singletonMap("Key1","Value1").entrySet().iterator().next())); - assertTrue(entries.contains(Collections.singletonMap("Key2","Value2").entrySet().iterator().next())); + assertTrue(entries.contains(Collections.singletonMap("Key1", "Value1").entrySet().iterator().next())); + assertTrue(entries.contains(Collections.singletonMap("Key2", "Value2").entrySet().iterator().next())); map1a.close(); @@ -420,6 +480,17 @@ public class TestWire { short channelID; + public Details() { + } + + + public Details(Class keyClass, Class valueClass, short channelID) { + this.keyClass = keyClass; + this.valueClass = valueClass; + this.channelID = channelID; + } + + @Override public void writeMarshallable(WireOut wire) { wire.write(() -> "keyClass").text(keyClass.getName());