diff --git a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java index 12ecc8390..ae6f3a148 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java @@ -19,11 +19,11 @@ import static net.openhft.chronicle.map.ChronicleMapBuilder.of; /** * Created by Rob Austin */ -public class MapWireConnectionHub { +public class MapWireConnectionHub implements Cloneable{ private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class); public static final int MAP_SERVICE = 3; - private final byte localIdentifier; + protected ChronicleMap channelNameToId; private Supplier>> mapFactory; @@ -31,6 +31,7 @@ public class MapWireConnectionHub { private final ReplicationHub hub; private final ArrayList bytesChronicleMaps = new ArrayList<>(); + private final ChannelProvider provider; public MapWireConnectionHub( @@ -42,7 +43,6 @@ public class MapWireConnectionHub { this.mapFactory = mapFactory; - this.localIdentifier = localIdentifier; final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(serverPort) @@ -50,14 +50,10 @@ public class MapWireConnectionHub { hub = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig).createWithId(localIdentifier); - // this is how you add maps after the custer is created - of(byte[].class, byte[].class) - .instance().replicatedViaChannel(hub.createChannel((short) 1)).create(); - channelNameToId = (ChronicleMap) channelNameToIdFactory.get() .replicatedViaChannel(hub.createChannel(MAP_SERVICE)).create(); - ChannelProvider provider = ChannelProvider.getProvider(hub); + provider = ChannelProvider.getProvider(hub); channelMap = provider.chronicleChannelMap(); } @@ -161,4 +157,7 @@ public class MapWireConnectionHub { throw new IllegalStateException(); } + public void close() throws IOException { + provider.close(); + } } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index d270c42a6..c3fd82368 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -190,7 +190,7 @@ public class MapWireHandler implements WireHandler, Consumer try { - if (put.contentEquals(eventName)) { + if (putIfAbsent.contentEquals(eventName)) { valueIn.marshallable(wire -> { diff --git a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java b/src/main/java/net/openhft/chronicle/map/WireMap.java similarity index 64% rename from src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java rename to src/main/java/net/openhft/chronicle/map/WireMap.java index 141059f50..e30cf31c1 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java +++ b/src/main/java/net/openhft/chronicle/map/WireMap.java @@ -1,9 +1,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.Bytes; -import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields; import net.openhft.chronicle.wire.TextWire; -import net.openhft.chronicle.wire.ValueOut; import net.openhft.chronicle.wire.Wire; import org.jetbrains.annotations.NotNull; @@ -12,11 +10,12 @@ import java.nio.ByteBuffer; import java.util.Collection; import java.util.Map; import java.util.Set; +import java.util.function.Supplier; /** * Created by Rob Austin */ -public class MapWireConnectionHubByName implements Map { +public class WireMap implements Map { private final Class kClass; private final Class vClass; @@ -28,13 +27,16 @@ public class MapWireConnectionHubByName implements Map { private final Map map; - Class wireType = TextWire.class; + // todo + private final Class wireType; - public MapWireConnectionHubByName(String name, - Class vClass, - Class kClass, - MapWireConnectionHub mapWireConnectionHub) throws IOException { + public WireMap(String name, + Class vClass, + Class kClass, + MapWireConnectionHub mapWireConnectionHub, + Class wireType) throws IOException { this.mapWireConnectionHub = mapWireConnectionHub; + this.wireType = wireType; final BytesChronicleMap b = this.mapWireConnectionHub.acquireMap(name); this.map = (Map) b.delegate; this.vClass = vClass; @@ -75,41 +77,51 @@ public class MapWireConnectionHubByName implements Map { } private byte[] bytes(Object key) { - ValueOut valueOut = toWire().getValueOut(); - AbstactStatelessClient.writeField(key, valueOut); - buffer.flip(); - byte[] bytes = new byte[(int) buffer.limit()]; - buffer.read(bytes); - return bytes; + + final Wire wire = toWire(); + AbstactStatelessClient.writeField(key, wire.getValueOut()); + wire.bytes().flip(); + + return toWire().getValueIn().bytes(); } + @Override public boolean containsValue(Object value) { return map.containsValue(bytes(value)); - } @Override - public V get(Object key) { - byte[] bytes = map.get(bytes(key)); + public V get(@NotNull final Object key) { + return toObject(vClass, new Supplier() { + @Override + public byte[] get() { + + final byte[] bytes = bytes(key); + final byte[] bytes1 = map.get(bytes); + return bytes1; + } + }); + } + + private E toObject(Class eClass, Supplier b) { + final byte[] bytes = b.get(); if (bytes == null) return null; - final Wire wire = toWire(bytes); - return AbstactStatelessClient.readObject(CoreFields.reply, wire, null, vClass); + buffer.flip(); + return AbstactStatelessClient.readObject(wire.getValueIn(), null, eClass); } @Override - public V put(K key, V value) { - final Wire wire = toWire(map.put(bytes(key), bytes(value))); - return AbstactStatelessClient.readObject(CoreFields.reply, wire, null, vClass); + public V put(@NotNull final K key, @NotNull final V value) { + return toObject(vClass, () -> map.put(bytes(key), bytes(value))); } @Override public V remove(Object key) { - final Wire wire = toWire(map.remove(bytes(key))); - return AbstactStatelessClient.readObject(CoreFields.reply, wire, null, vClass); + return toObject(vClass, () -> map.remove(bytes(key))); } @Override @@ -137,7 +149,7 @@ public class MapWireConnectionHubByName implements Map { @NotNull @Override public Set> entrySet() { - return null; + throw new UnsupportedOperationException("todo"); } }