From d16929ae8871928639458797bf66b1b7e429dab2 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 23 Apr 2015 12:24:07 +0100 Subject: [PATCH] CE-23 further improved the support for wire maps to be accessed via engine. --- .../chronicle/map/MapWireConnectionHub.java | 17 +- .../openhft/chronicle/map/MapWireHandler.java | 2 +- .../net/openhft/chronicle/map/WireMap.java | 297 ++++++++++++++++-- 3 files changed, 284 insertions(+), 32 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java index ae6f3a148..2bfbd65c7 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java @@ -8,6 +8,7 @@ import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.Closeable; import java.io.IOException; import java.util.ArrayList; import java.util.Map; @@ -19,30 +20,27 @@ import static net.openhft.chronicle.map.ChronicleMapBuilder.of; /** * Created by Rob Austin */ -public class MapWireConnectionHub implements Cloneable{ +public class MapWireConnectionHub implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class); public static final int MAP_SERVICE = 3; - protected ChronicleMap channelNameToId; - private Supplier>> mapFactory; private final Map channelMap; private final ReplicationHub hub; private final ArrayList bytesChronicleMaps = new ArrayList<>(); private final ChannelProvider provider; + private final Supplier>> mapFactory; public MapWireConnectionHub( - @NotNull final Supplier>> mapFactory, - @NotNull final Supplier>> - channelNameToIdFactory, byte localIdentifier, int serverPort) throws IOException { - - this.mapFactory = mapFactory; + // this is used to hold the name to channel id + final Supplier>> + channelNameToIdFactory = () -> of(String.class, Integer.class).instance(); final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(serverPort) @@ -56,6 +54,9 @@ public class MapWireConnectionHub implements Cloneable{ provider = ChannelProvider.getProvider(hub); channelMap = provider.chronicleChannelMap(); + mapFactory = () -> of(byte[].class, byte[].class).instance(); + + } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index c3fd82368..81cba95b4 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -198,7 +198,7 @@ public class MapWireHandler implements WireHandler, Consumer final byte[] key = wire.read(params[0]).bytes(); final byte[] value = wire.read(params[1]).bytes(); - writeVoid(b -> ((Map) b.delegate).putIfAbsent(key, value)); + writeValueFromBytes(b -> ((Map) b.delegate).putIfAbsent(key, value)); }); diff --git a/src/main/java/net/openhft/chronicle/map/WireMap.java b/src/main/java/net/openhft/chronicle/map/WireMap.java index e30cf31c1..83c35d652 100644 --- a/src/main/java/net/openhft/chronicle/map/WireMap.java +++ b/src/main/java/net/openhft/chronicle/map/WireMap.java @@ -1,29 +1,32 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.Bytes; +import net.openhft.chronicle.hash.function.SerializableFunction; import net.openhft.chronicle.wire.TextWire; import net.openhft.chronicle.wire.Wire; import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; -import java.util.Collection; -import java.util.Map; -import java.util.Set; +import java.util.*; +import java.util.function.Consumer; +import java.util.function.Predicate; import java.util.function.Supplier; /** * Created by Rob Austin */ -public class WireMap implements Map { +public class WireMap implements ChronicleMap { + + private static final Logger LOG = LoggerFactory.getLogger(WireMap.class); private final Class kClass; private final Class vClass; private final MapWireConnectionHub mapWireConnectionHub; - - - Bytes buffer = Bytes.elasticByteBuffer(); - + private Bytes buffer = Bytes.elasticByteBuffer(); private final Map map; @@ -94,17 +97,65 @@ public class WireMap implements Map { @Override public V get(@NotNull final Object key) { - return toObject(vClass, new Supplier() { - @Override - public byte[] get() { + return toObject(vClass, () -> { - final byte[] bytes = bytes(key); - final byte[] bytes1 = map.get(bytes); - return bytes1; - } + final byte[] bytes = bytes(key); + final byte[] bytes1 = map.get(bytes); + return bytes1; }); } + @Override + public V getUsing(K key, V usingValue) { + throw new UnsupportedOperationException("todo"); + } + + @Override + public V acquireUsing(@NotNull K key, V usingValue) { + throw new UnsupportedOperationException("todo"); + } + + @NotNull + @Override + public MapKeyContext acquireContext(@NotNull K key, @NotNull V usingValue) { + throw new UnsupportedOperationException("todo"); + } + + @Override + public R getMapped(K key, @NotNull SerializableFunction function) { + throw new UnsupportedOperationException("todo"); + } + + @Override + public V putMapped(K key, @NotNull UnaryOperator unaryOperator) { + throw new UnsupportedOperationException("todo"); + } + + @Override + public void getAll(File toFile) throws IOException { + throw new UnsupportedOperationException("todo"); + } + + @Override + public void putAll(File fromFile) throws IOException { + throw new UnsupportedOperationException("todo"); + } + + @Override + public V newValueInstance() { + throw new UnsupportedOperationException("todo"); + } + + @Override + public K newKeyInstance() { + return null; + } + + @Override + public Class valueClass() { + return null; + } + private E toObject(Class eClass, Supplier b) { final byte[] bytes = b.get(); if (bytes == null) @@ -126,7 +177,11 @@ public class WireMap implements Map { @Override public void putAll(Map m) { - throw new UnsupportedOperationException("todo"); + + for (final Map.Entry e : m.entrySet()) { + map.put(bytes(e.getKey()), bytes(e.getValue())); + } + } @Override @@ -134,22 +189,218 @@ public class WireMap implements Map { map.clear(); } - @NotNull - @Override - public Set keySet() { - throw new UnsupportedOperationException("todo"); - } @NotNull @Override - public Collection values() { - throw new UnsupportedOperationException("todo"); + public Set keySet() { + + return new AbstractSet() { + + public Iterator iterator() { + return new Iterator() { + private Iterator> i = entrySet().iterator(); + + public boolean hasNext() { + return i.hasNext(); + } + + public K next() { + return i.next().getKey(); + } + + public void remove() { + i.remove(); + } + }; + } + + public int size() { + return WireMap.this.size(); + } + + public boolean isEmpty() { + return WireMap.this.isEmpty(); + } + + public void clear() { + WireMap.this.clear(); + } + + public boolean contains(Object k) { + return WireMap.this.containsKey(bytes(k)); + } + }; + } + + public Collection values() { + return new AbstractCollection() { + public Iterator iterator() { + return new Iterator() { + private Iterator> i = entrySet().iterator(); + + public boolean hasNext() { + return i.hasNext(); + } + + public V next() { + return i.next().getValue(); + } + + public void remove() { + i.remove(); + } + }; + } + + public int size() { + return WireMap.this.size(); + } + + public boolean isEmpty() { + return WireMap.this.isEmpty(); + } + + public void clear() { + WireMap.this.clear(); + } + + public boolean contains(Object v) { + return WireMap.this.containsValue(v); + } + }; + } + + @NotNull @Override public Set> entrySet() { + + return new AbstractSet>() { + + + @NotNull + @Override + public Iterator> iterator() { + final Iterator> iterator = map.entrySet().iterator(); + return new Iterator>() { + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public Entry next() { + + final Entry next = iterator.next(); + + return new Entry() { + + @Override + public K getKey() { + return toObject(kClass, () -> next.getKey()); + } + + @Override + public V getValue() { + return toObject(vClass, () -> next.getValue()); + } + + @Override + public V setValue(V value) { + throw new UnsupportedOperationException("todo"); + } + }; + + + } + }; + } + + + public int size() { + return WireMap.this.size(); + } + + public boolean isEmpty() { + return WireMap.this.isEmpty(); + } + + public void clear() { + WireMap.this.clear(); + } + + public boolean contains(Object v) { + return WireMap.this.containsValue(v); + } + + + @Override + public boolean remove(Object o) { + return WireMap.this.remove(o) != null; + } + }; + } + + + @Override + public V putIfAbsent(K key, V value) { throw new UnsupportedOperationException("todo"); } + @Override + public boolean remove(Object key, Object value) { + throw new UnsupportedOperationException("todo"); + } + + @Override + public boolean replace(K key, V oldValue, V newValue) { + throw new UnsupportedOperationException("todo"); + } + + @Override + public V replace(K key, V value) { + throw new UnsupportedOperationException("todo"); + } + + @Override + public File file() { + throw new UnsupportedOperationException("todo"); + } + + @Override + public long longSize() { + return map.size(); + } + + @Override + public MapKeyContext context(K key) { + throw new UnsupportedOperationException("todo"); + } + + @Override + public Class keyClass() { + throw new UnsupportedOperationException("todo"); + } + + @Override + public boolean forEachEntryWhile(Predicate> predicate) { + throw new UnsupportedOperationException("todo"); + } + + @Override + public void forEachEntry(Consumer> action) { + throw new UnsupportedOperationException("todo"); + } + + @Override + public void close() { + try { + mapWireConnectionHub.close(); + } catch (IOException e) { + LOG.error("", e); + } + } }