diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index ad11dfd16..63124792f 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -31,10 +31,13 @@ import java.io.File; import java.io.IOException; import java.util.*; import java.util.function.Consumer; +import java.util.function.Function; import java.util.function.Predicate; import static java.util.Collections.emptyList; import static net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields.reply; +import static net.openhft.chronicle.map.ClientWiredStatelessChronicleSet.Params.key; +import static net.openhft.chronicle.map.ClientWiredStatelessChronicleSet.Params.value; import static net.openhft.chronicle.map.MapWireHandler.EventId; import static net.openhft.chronicle.map.MapWireHandler.EventId.*; import static net.openhft.chronicle.map.VanillaChronicleMap.newInstance; @@ -43,14 +46,14 @@ import static net.openhft.chronicle.map.VanillaChronicleMap.newInstance; /** * @author Rob Austin. */ -class ClientWiredStatelessChronicleMap extends MapStatelessClient +class ClientWiredStatelessChronicleMap extends MapStatelessClient implements ChronicleMap, Cloneable, ChannelFactory { private static final Logger LOG = LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class); public static final Consumer VOID_PARAMETERS = out -> out.marshallable(AbstactStatelessClient.EMPTY); - private final Class vClass; + private final Class vClass; protected Class kClass; private boolean putReturnsNull; private boolean removeReturnsNull; @@ -62,7 +65,7 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient vClass, @NotNull final String channelName, @NotNull final ClientWiredStatelessTcpConnectionHub hub) { - super(channelName, hub, "MAP", 0, kClass, vClass); + super(channelName, hub, "MAP", 0); this.putReturnsNull = config.putReturnsNull(); this.removeReturnsNull = config.removeReturnsNull(); @@ -203,7 +206,7 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient entrySet = entrySet(); + final ClientWiredStatelessChronicleSet> entrySet = entrySet(); final Iterator> entries = entrySet.segmentIterator(1); @@ -272,7 +275,7 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient extends MapStatelessClient extends MapStatelessClient keySet() { - return null; - } @NotNull @Override @@ -346,7 +345,7 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient cidToCsp = new HashMap<>(); @NotNull - public ClientWiredStatelessChronicleEntrySet entrySet() { + public ClientWiredStatelessChronicleSet entrySet() { long cid = proxyReturnWireConsumer(entrySet, (WireIn wireIn) -> { @@ -365,7 +364,51 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient(channelName, hub, cid, vClass, kClass); + + Function> conumer = new Function>() { + Map.Entry e = null; + + @Override + public Map.Entry apply(ValueIn valueIn) { + + valueIn.marshallable(r -> { + final K k = r.read(key).object(kClass); + final V v = r.read(value).object(vClass); + + e = new Map.Entry() { + + @Override + public K getKey() { + return k; + } + + @Override + public V getValue() { + return v; + } + + @Override + public Object setValue(Object value) { + throw new UnsupportedOperationException(); + } + }; + + } + + ); + + return e; + } + }; + + return new ClientWiredStatelessChronicleSet<>(channelName, hub, cid, conumer); + } + + + @NotNull + public ClientWiredStatelessChronicleSet keySet() { + + throw new UnsupportedOperationException("todo"); } diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleSet.java similarity index 64% rename from src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java rename to src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleSet.java index c3d06e414..956616d7b 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleSet.java @@ -6,23 +6,30 @@ import net.openhft.chronicle.wire.ValueIn; import net.openhft.chronicle.wire.WireKey; import org.jetbrains.annotations.NotNull; -import java.util.*; +import java.util.Collection; +import java.util.HashSet; +import java.util.Iterator; +import java.util.Set; +import java.util.function.Function; import static net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields.reply; -import static net.openhft.chronicle.map.ClientWiredStatelessChronicleEntrySet.EntrySetEventId.*; -import static net.openhft.chronicle.map.ClientWiredStatelessChronicleEntrySet.Params.*; +import static net.openhft.chronicle.map.ClientWiredStatelessChronicleSet.Params.key; +import static net.openhft.chronicle.map.ClientWiredStatelessChronicleSet.Params.segment; +import static net.openhft.chronicle.map.ClientWiredStatelessChronicleSet.SetEventId.*; -class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient - implements Set> { +class ClientWiredStatelessChronicleSet extends MapStatelessClient + implements Set { - public ClientWiredStatelessChronicleEntrySet(@NotNull final String channelName, - @NotNull final ClientWiredStatelessTcpConnectionHub hub, - final long cid, - @NotNull final Class vClass, - @NotNull final Class kClass) { + private final Function conumer; - super(channelName, hub, "entrySet", cid, kClass, vClass); + public ClientWiredStatelessChronicleSet(@NotNull final String channelName, + @NotNull final ClientWiredStatelessTcpConnectionHub hub, + final long cid, + @NotNull final Function wireToSet) { + + super(channelName, hub, "entrySet", cid); + this.conumer = wireToSet; } @@ -43,7 +50,7 @@ class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient> iterator() { + public Iterator iterator() { final int numberOfSegments = proxyReturnUint16(numberOfSegements); // todo itterate the segments @@ -59,9 +66,9 @@ class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient> segmentIterator(int segment) { + Iterator segmentIterator(int segment) { - final Map e = new HashMap(); + final Set e = new HashSet(); proxyReturnWireConsumerInOut(iterator, @@ -72,19 +79,14 @@ class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient s.marshallable(r -> { - final K k = r.read(key).object(kClass); - final V v = r.read(value).object(vClass); - e.put(k, v); - })); + read.sequence(v -> e.add(conumer.apply(read))); } return e; }); - return e.entrySet().iterator(); + return e.iterator(); } @NotNull @@ -100,7 +102,7 @@ class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient kvEntry) { + public boolean add(U u) { return false; } @@ -116,10 +118,11 @@ class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient> c) { + public boolean addAll(Collection c) { return false; } + @Override public boolean retainAll(Collection c) { return false; @@ -136,7 +139,7 @@ class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient extends MapStatelessClient extends MapStatelessClient EntrySetEventId(P... params) { +

SetEventId(P... params) { this.params = params; } diff --git a/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java b/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java index b7aa3bc5c..b2b87d61d 100644 --- a/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java +++ b/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java @@ -2,40 +2,27 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub; import net.openhft.chronicle.engine.client.ParameterizeWireKey; -import net.openhft.chronicle.wire.Wire; import net.openhft.chronicle.wire.WireKey; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; -import static net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields.reply; - /** * Created by Rob Austin */ -public abstract class MapStatelessClient - extends AbstactStatelessClient { - - Class kClass; - Class vClass; - +public abstract class MapStatelessClient< E extends ParameterizeWireKey> extends AbstactStatelessClient { /** * @param channelName * @param hub * @param type the type of wire handler for example "MAP" or "QUEUE" * @param cid used by proxies such as the entry-set - * @param kClass */ public MapStatelessClient(@NotNull String channelName, @NotNull ClientWiredStatelessTcpConnectionHub hub, @NotNull String type, - long cid, - @NotNull final Class kClass, - @NotNull final Class vClass) { + long cid) { super(channelName, hub, type, cid); - this.vClass = vClass; - this.kClass = kClass; } @Nullable @@ -44,61 +31,33 @@ public abstract class MapStatelessClient @NotNull final Class resultType, @Nullable Object... args) { - final long startTime = System.currentTimeMillis(); - final long tid = sendEvent(startTime, eventId, toParameters(eventId, args)); + return proxyReturnWireConsumerInOut(eventId, + toParameters(eventId, args), + f -> f.read(ClientWiredStatelessTcpConnectionHub.CoreFields.reply) + .object(resultType)); - if (eventReturnsNull(eventId)) - return null; - - if (resultType == vClass) - return (R) readValue(reply, tid, startTime, null); - - else - throw new UnsupportedOperationException("class of type class=" + resultType + - " is not supported"); - } - - @Nullable - protected R proxyReturnObject(@NotNull final Class rClass, - @NotNull final E eventId, - @NotNull final Object o) { - - final long startTime = System.currentTimeMillis(); - final long tid = sendEvent(startTime, eventId, out -> out.object(o)); - - if (eventReturnsNull(eventId)) - return null; - - if (rClass == vClass) - return (R) readValue(reply, tid, startTime, null); - else - throw new UnsupportedOperationException("class of type class=" + rClass + " is not " + - "supported"); } protected abstract boolean eventReturnsNull(@NotNull E methodName); - protected V readValue(WireKey argName, long tid, long startTime, final V usingValue) { - assert !hub.outBytesLock().isHeldByCurrentThread(); - long timeoutTime = startTime + hub.timeoutMs; - - hub.inBytesLock().lock(); - try { - - final Wire wireIn = hub.proxyReply(timeoutTime, tid); - checkIsData(wireIn); + protected E proxyReturnE(@NotNull final WireKey eventId, + @Nullable final O arg, + @NotNull final Class eClass) { - return readV(argName, wireIn, usingValue); + return proxyReturnWireConsumerInOut(eventId, + valueOut -> valueOut.object(arg), + f -> f.read(ClientWiredStatelessTcpConnectionHub.CoreFields.reply) + .object(eClass)); + } - } finally { - hub.inBytesLock().unlock(); - } + protected E proxyReturnE(@NotNull final WireKey eventId, + @NotNull final Class eClass) { + return proxyReturnWireConsumer(eventId, + f -> f.read(ClientWiredStatelessTcpConnectionHub.CoreFields.reply) + .object(eClass)); } - private V readV(WireKey argName, Wire wireIn, V usingValue) { - return readObject(argName, wireIn, usingValue, vClass); - } } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 87cc5334b..6c3525223 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -26,7 +26,7 @@ import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub; import net.openhft.chronicle.engine.client.ParameterizeWireKey; import net.openhft.chronicle.hash.impl.util.BuildVersion; -import net.openhft.chronicle.map.ClientWiredStatelessChronicleEntrySet.EntrySetEventId; +import net.openhft.chronicle.map.ClientWiredStatelessChronicleSet.SetEventId; import net.openhft.chronicle.network.WireHandler; import net.openhft.chronicle.network.event.EventGroup; import net.openhft.chronicle.network.event.WireHandlers; @@ -222,7 +222,7 @@ public class MapWireHandler implements WireHandler, Consumer // note : remove on the key-set returns a boolean and on the map returns the // old value - if (EntrySetEventId.remove.contentEquals(eventName)) { + if (SetEventId.remove.contentEquals(eventName)) { write(b -> outWire.write(reply).bool( b.delegate.remove(toByteArray(valueIn)) != null)); return; @@ -230,7 +230,7 @@ public class MapWireHandler implements WireHandler, Consumer // note : remove on the key-set returns a boolean and on the map returns the // old value - if (EntrySetEventId.iterator.contentEquals(eventName)) { + if (SetEventId.iterator.contentEquals(eventName)) { write(b -> { final ValueOut valueOut = outWire.writeEventName(CoreFields.reply); b.delegate.entrySet().forEach(e -> { @@ -262,7 +262,7 @@ public class MapWireHandler implements WireHandler, Consumer } - if (EntrySetEventId.numberOfSegements.contentEquals(eventName)) { + if (SetEventId.numberOfSegements.contentEquals(eventName)) { write(b -> outWire.write(reply).int32(1)); return; }