Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

added support for entrySet() to textWire

Browse Source
This commit is contained in:
Rob Austin
2015-04-29 13:11:54 +01:00
parent e4e8efe295
commit a6bd9e1e5d
4 changed files with 125 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java
+53 -12
View File
@@ -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<K, V> extends MapStatelessClient<K, V, ClientWiredStatelessChronicleEntrySet.EntrySetEventId>
@@ -22,9 +19,10 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends MapStatelessClient<K,
public ClientWiredStatelessChronicleEntrySet(@NotNull final String channelName,
@NotNull final ClientWiredStatelessTcpConnectionHub hub,
final long cid,
@NotNull final Class<V> vClass) {
@NotNull final Class<V> vClass,
@NotNull final Class<K> kClass) {
super(channelName, hub, "entrySet", cid, vClass);
super(channelName, hub, "entrySet", cid, kClass, vClass);
}
@@ -46,7 +44,47 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends MapStatelessClient<K,
@NotNull
@Override
public Iterator<Map.Entry<K, V>> 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<Map.Entry<K, V>> segmentIterator(int segment) {
final Map<K, V> e = new HashMap<K, V>();
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<K, V> extends MapStatelessClient<K,
enum Params implements WireKey {
key
key,
value,
segment
}
enum EntrySetEventId implements ParameterizeWireKey {
size,
isEmpty,
remove(key),
iterator;
numberOfSegements,
iterator(segment);
private final WireKey[] params;
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+32 -10
View File
@@ -29,10 +29,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.Predicate;
@@ -61,11 +58,11 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev
public ClientWiredStatelessChronicleMap(
@NotNull final ClientWiredChronicleMapStatelessBuilder config,
@NotNull final Class kClass,
@NotNull final Class vClass,
@NotNull final Class<K> kClass,
@NotNull final Class<V> 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<K, V> extends MapStatelessClient<K, V, Ev
@NotNull
public String toString() {
return "todo";
final ClientWiredStatelessChronicleEntrySet<K, V> entrySet = entrySet();
final Iterator<Map.Entry<K, V>> entries = entrySet.segmentIterator(1);
if (!entries.hasNext())
return "{}";
StringBuilder sb = new StringBuilder();
sb.append('{');
while (entries.hasNext()) {
final Map.Entry<K, V> 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<K, V> extends MapStatelessClient<K, V, Ev
private final Map<Long, String> cidToCsp = new HashMap<>();
@NotNull
public Set<Map.Entry<K, V>> entrySet() {
public ClientWiredStatelessChronicleEntrySet entrySet() {
long cid = proxyReturnWireConsumer(entrySet, (WireIn wireIn) -> {
@@ -343,7 +365,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev
return cidRef[0];
});
return new ClientWiredStatelessChronicleEntrySet<K, V>(channelName, hub, cid, vClass);
return new ClientWiredStatelessChronicleEntrySet<K, V>(channelName, hub, cid, vClass, kClass);
}
src/main/java/net/openhft/chronicle/map/MapStatelessClient.java
+8 -2
View File
@@ -15,21 +15,27 @@ import static net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnect
public abstract class MapStatelessClient<K, V, E extends ParameterizeWireKey>
extends AbstactStatelessClient<E> {
private Class<V> vClass;
Class<K> kClass;
Class<V> 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<K> kClass,
@NotNull final Class<V> vClass) {
super(channelName, hub, type, cid);
this.vClass = vClass;
this.kClass = kClass;
}
@Nullable
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+32 -7
View File
@@ -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<K, V> implements WireHandler, Consumer<WireHandlers>
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<K, V> implements WireHandler, Consumer<WireHandlers>
// 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<ValueOut> 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<K, V> implements WireHandler, Consumer<WireHandlers>
}
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");
}