From bd643b106fc0f32017bf28f200cfb092d8d53a57 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 22 Apr 2015 16:43:11 +0100 Subject: [PATCH 1/2] CE-23 now supports map.get() on the server using the same map that wire uses --- .../chronicle/map/MapWireConnectionHub.java | 7 +------ .../map/MapWireConnectionHubByName.java | 21 +++++++++++-------- .../openhft/chronicle/map/MapWireHandler.java | 2 +- 3 files changed, 14 insertions(+), 16 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java index 12ecc8390..6b3fdb454 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java @@ -23,7 +23,7 @@ public class MapWireConnectionHub { 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; @@ -42,7 +42,6 @@ public class MapWireConnectionHub { this.mapFactory = mapFactory; - this.localIdentifier = localIdentifier; final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(serverPort) @@ -50,10 +49,6 @@ 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(); diff --git a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java index 141059f50..15a11b44b 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java @@ -3,7 +3,6 @@ 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; @@ -75,14 +74,16 @@ 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)); @@ -91,13 +92,15 @@ public class MapWireConnectionHubByName implements Map { @Override public V get(Object key) { - byte[] bytes = map.get(bytes(key)); + final byte[] bytes1 = bytes(key); + byte[] bytes = map.get(bytes1); 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, vClass); } @Override 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 -> { From f4003520ab55904ec918f06f341e41bd3c6d782f Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 22 Apr 2015 18:22:12 +0100 Subject: [PATCH 2/2] CE-23 local and remote wire maps can now share the same replication hub --- .../chronicle/map/MapWireConnectionHub.java | 8 ++- ...eConnectionHubByName.java => WireMap.java} | 49 +++++++++++-------- 2 files changed, 35 insertions(+), 22 deletions(-) rename src/main/java/net/openhft/chronicle/map/{MapWireConnectionHubByName.java => WireMap.java} (69%) diff --git a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java index 6b3fdb454..ae6f3a148 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java @@ -19,7 +19,7 @@ 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; @@ -31,6 +31,7 @@ public class MapWireConnectionHub { private final ReplicationHub hub; private final ArrayList bytesChronicleMaps = new ArrayList<>(); + private final ChannelProvider provider; public MapWireConnectionHub( @@ -52,7 +53,7 @@ public class MapWireConnectionHub { channelNameToId = (ChronicleMap) channelNameToIdFactory.get() .replicatedViaChannel(hub.createChannel(MAP_SERVICE)).create(); - ChannelProvider provider = ChannelProvider.getProvider(hub); + provider = ChannelProvider.getProvider(hub); channelMap = provider.chronicleChannelMap(); } @@ -156,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/MapWireConnectionHubByName.java b/src/main/java/net/openhft/chronicle/map/WireMap.java similarity index 69% rename from src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java rename to src/main/java/net/openhft/chronicle/map/WireMap.java index 15a11b44b..e30cf31c1 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java +++ b/src/main/java/net/openhft/chronicle/map/WireMap.java @@ -1,7 +1,6 @@ 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.Wire; import org.jetbrains.annotations.NotNull; @@ -11,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; @@ -27,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,7 +78,6 @@ public class MapWireConnectionHubByName implements Map { private byte[] bytes(Object key) { - final Wire wire = toWire(); AbstactStatelessClient.writeField(key, wire.getValueOut()); wire.bytes().flip(); @@ -87,32 +89,39 @@ public class MapWireConnectionHubByName implements Map { @Override public boolean containsValue(Object value) { return map.containsValue(bytes(value)); - } @Override - public V get(Object key) { - final byte[] bytes1 = bytes(key); - byte[] bytes = map.get(bytes1); + 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); buffer.flip(); - return AbstactStatelessClient.readObject(wire.getValueIn(), null, vClass); + 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 @@ -140,7 +149,7 @@ public class MapWireConnectionHubByName implements Map { @NotNull @Override public Set> entrySet() { - return null; + throw new UnsupportedOperationException("todo"); } }