diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java index 36a820af5..c3d06e414 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java @@ -2,18 +2,15 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub; import net.openhft.chronicle.engine.client.ParameterizeWireKey; -import net.openhft.chronicle.wire.ValueOut; +import net.openhft.chronicle.wire.ValueIn; import net.openhft.chronicle.wire.WireKey; import org.jetbrains.annotations.NotNull; -import java.util.Collection; -import java.util.Iterator; -import java.util.Map; -import java.util.Set; -import java.util.function.Consumer; +import java.util.*; +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.key; +import static net.openhft.chronicle.map.ClientWiredStatelessChronicleEntrySet.Params.*; class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient @@ -22,9 +19,10 @@ class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient vClass) { + @NotNull final Class vClass, + @NotNull final Class kClass) { - super(channelName, hub, "entrySet", cid, vClass); + super(channelName, hub, "entrySet", cid, kClass, vClass); } @@ -46,7 +44,47 @@ class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient> iterator() { - return null; + final int numberOfSegments = proxyReturnUint16(numberOfSegements); + + // todo itterate the segments + return segmentIterator(1); + + } + + + /** + * gets the iterator for a given segment + * + * @param segment the maps segment number + * @return and iterator for the {@code segment} + */ + @NotNull + Iterator> segmentIterator(int segment) { + + final Map e = new HashMap(); + + proxyReturnWireConsumerInOut(iterator, + + valueOut -> valueOut.uint16(segment), + + wireIn -> { + + final ValueIn read = wireIn.read(reply); + + while (read.hasNextSequenceItem()) { + + read.sequence(s -> s.marshallable(r -> { + final K k = r.read(key).object(kClass); + final V v = r.read(value).object(vClass); + e.put(k, v); + })); + } + + return e; + }); + + + return e.entrySet().iterator(); } @NotNull @@ -104,14 +142,17 @@ class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient extends MapStatelessClient kClass, + @NotNull final Class vClass, @NotNull final String channelName, @NotNull final ClientWiredStatelessTcpConnectionHub hub) { - super(channelName, hub, "MAP", 0, vClass); + super(channelName, hub, "MAP", 0, kClass, vClass); this.putReturnsNull = config.putReturnsNull(); this.removeReturnsNull = config.removeReturnsNull(); @@ -206,7 +203,32 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient entrySet = entrySet(); + + final Iterator> entries = entrySet.segmentIterator(1); + + if (!entries.hasNext()) + return "{}"; + + StringBuilder sb = new StringBuilder(); + sb.append('{'); + + + while (entries.hasNext()) { + + final Map.Entry e = entries.next(); + final K key = e.getKey(); + final V value = e.getValue(); + sb.append(key == this ? "(this Map)" : key); + sb.append('='); + sb.append(value == this ? "(this Map)" : value); + if (!entries.hasNext()) + return sb.append('}').toString(); + sb.append(',').append(' '); + } + + return sb.toString(); + } @NotNull @@ -324,7 +346,7 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient cidToCsp = new HashMap<>(); @NotNull - public Set> entrySet() { + public ClientWiredStatelessChronicleEntrySet entrySet() { long cid = proxyReturnWireConsumer(entrySet, (WireIn wireIn) -> { @@ -343,7 +365,7 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient(channelName, hub, cid, vClass); + return new ClientWiredStatelessChronicleEntrySet(channelName, hub, cid, vClass, kClass); } diff --git a/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java b/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java index 7f8b1591b..b7aa3bc5c 100644 --- a/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java +++ b/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java @@ -15,21 +15,27 @@ import static net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnect public abstract class MapStatelessClient extends AbstactStatelessClient { - private Class vClass; + Class kClass; + Class vClass; + /** * @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 vClass) { + long cid, + @NotNull final Class kClass, + @NotNull final Class vClass) { super(channelName, hub, type, cid); this.vClass = vClass; + this.kClass = kClass; } @Nullable diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 45d5d3177..87cc5334b 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -48,7 +48,6 @@ import static net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnect import static net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields.csp; import static net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields.reply; import static net.openhft.chronicle.engine.client.StringUtils.endsWith; -import static net.openhft.chronicle.map.AbstactStatelessClient.toParameters; import static net.openhft.chronicle.map.MapWireHandler.EventId.*; import static net.openhft.chronicle.map.MapWireHandler.Params.*; import static net.openhft.chronicle.wire.Wires.acquireStringBuilder; @@ -131,7 +130,6 @@ public class MapWireHandler implements WireHandler, Consumer this.publishLater = wireHandlers; } - @Override public void process(@NotNull final Wire in, @NotNull final Wire out) throws StreamCorruptedException { try { @@ -234,11 +232,29 @@ public class MapWireHandler implements WireHandler, Consumer // old value if (EntrySetEventId.iterator.contentEquals(eventName)) { write(b -> { - final ValueOut valueOut = outWire.writeEventName(() -> "entry"); - b.delegate.entrySet().forEach(e -> - valueOut.sequence(toParameters(put, - e.getKey(), - e.getValue()))); + final ValueOut valueOut = outWire.writeEventName(CoreFields.reply); + b.delegate.entrySet().forEach(e -> { + + final Consumer writer = out -> { + + out.marshallable(m -> { + + final byte[] key = (byte[]) e.getKey(); + final byte[] value = (byte[]) e.getValue(); + + m.write(Params.key); + outWire.bytes().write(key); + + m.write(Params.value); + outWire.bytes().write(value); + + }); + + }; + valueOut.sequence(writer); + } + + ); } ); @@ -246,12 +262,21 @@ public class MapWireHandler implements WireHandler, Consumer } + if (EntrySetEventId.numberOfSegements.contentEquals(eventName)) { + write(b -> outWire.write(reply).int32(1)); + return; + } + + + + throw new IllegalStateException("unsupported event=" + eventName); } // -- THESE METHODS ARE ONLY MAP METHODS + if (keySet.contentEquals(eventName)) { throw new UnsupportedOperationException("todo"); }