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

for wire - now handling the entrySet() as a genric set() proxy

Browse Source
This commit is contained in:
Rob Austin
2015-04-29 23:32:54 +01:00
parent c704abaf04
commit d6fa74f869
4 changed files with 108 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+56 -13
View File
@@ -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<K, V> extends MapStatelessClient<K, V, EventId>
class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<EventId>
implements ChronicleMap<K, V>, Cloneable, ChannelFactory {
private static final Logger LOG =
LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class);
public static final Consumer<ValueOut> VOID_PARAMETERS = out -> out.marshallable(AbstactStatelessClient.EMPTY);
private final Class vClass;
private final Class<V> vClass;
protected Class<K> kClass;
private boolean putReturnsNull;
private boolean removeReturnsNull;
@@ -62,7 +65,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev
@NotNull final Class<V> 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<K, V> extends MapStatelessClient<K, V, Ev
@NotNull
public String toString() {
final ClientWiredStatelessChronicleEntrySet<K, V> entrySet = entrySet();
final ClientWiredStatelessChronicleSet<Map.Entry<K, V>> entrySet = entrySet();
final Iterator<Map.Entry<K, V>> entries = entrySet.segmentIterator(1);
@@ -272,7 +275,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev
}
public V get(Object key) {
return (V)proxyReturnObject(vClass, get, (K) key);
return (V) this.proxyReturnTypedObject(get, vClass, key);
}
@Nullable
@@ -295,7 +298,8 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev
public V remove(Object key) {
if (key == null)
throw keyNotNullNPE();
return (V)proxyReturnObject(vClass, removeReturnsNull ? removeWithoutAcc : remove, key);
final EventId eventId = removeReturnsNull ? removeWithoutAcc : remove;
return (V) this.proxyReturnTypedObject(eventId, vClass, key);
}
@Override
@@ -330,11 +334,6 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev
proxyReturnVoid(clear);
}
@NotNull
@Override
public Set<K> keySet() {
return null;
}
@NotNull
@Override
@@ -346,7 +345,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev
private final Map<Long, String> cidToCsp = new HashMap<>();
@NotNull
public ClientWiredStatelessChronicleEntrySet entrySet() {
public ClientWiredStatelessChronicleSet entrySet() {
long cid = proxyReturnWireConsumer(entrySet, (WireIn wireIn) -> {
@@ -365,7 +364,51 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev
return cidRef[0];
});
return new ClientWiredStatelessChronicleEntrySet<K, V>(channelName, hub, cid, vClass, kClass);
Function<ValueIn, Map.Entry<K, V>> conumer = new Function<ValueIn, Map.Entry<K, V>>() {
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");
}
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java → src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleSet.java
+29 -26
View File
@@ -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<K, V> extends MapStatelessClient<K, V, ClientWiredStatelessChronicleEntrySet.EntrySetEventId>
implements Set<Map.Entry<K, V>> {
class ClientWiredStatelessChronicleSet<U> extends MapStatelessClient<ClientWiredStatelessChronicleSet.SetEventId>
implements Set<U> {
public ClientWiredStatelessChronicleEntrySet(@NotNull final String channelName,
@NotNull final ClientWiredStatelessTcpConnectionHub hub,
final long cid,
@NotNull final Class<V> vClass,
@NotNull final Class<K> kClass) {
private final Function<ValueIn, U> conumer;
super(channelName, hub, "entrySet", cid, kClass, vClass);
public ClientWiredStatelessChronicleSet(@NotNull final String channelName,
@NotNull final ClientWiredStatelessTcpConnectionHub hub,
final long cid,
@NotNull final Function<ValueIn, U> wireToSet) {
super(channelName, hub, "entrySet", cid);
this.conumer = wireToSet;
}
@@ -43,7 +50,7 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends MapStatelessClient<K,
@NotNull
@Override
public Iterator<Map.Entry<K, V>> iterator() {
public Iterator<U> iterator() {
final int numberOfSegments = proxyReturnUint16(numberOfSegements);
// todo itterate the segments
@@ -59,9 +66,9 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends MapStatelessClient<K,
* @return and iterator for the {@code segment}
*/
@NotNull
Iterator<Map.Entry<K, V>> segmentIterator(int segment) {
Iterator<U> segmentIterator(int segment) {
final Map<K, V> e = new HashMap<K, V>();
final Set<U> e = new HashSet<U>();
proxyReturnWireConsumerInOut(iterator,
@@ -72,19 +79,14 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends MapStatelessClient<K,
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);
}));
read.sequence(v -> e.add(conumer.apply(read)));
}
return e;
});
return e.entrySet().iterator();
return e.iterator();
}
@NotNull
@@ -100,7 +102,7 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends MapStatelessClient<K,
}
@Override
public boolean add(Map.Entry<K, V> kvEntry) {
public boolean add(U u) {
return false;
}
@@ -116,10 +118,11 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends MapStatelessClient<K,
}
@Override
public boolean addAll(Collection<? extends Map.Entry<K, V>> c) {
public boolean addAll(Collection<? extends U> c) {
return false;
}
@Override
public boolean retainAll(Collection<?> c) {
return false;
@@ -136,7 +139,7 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends MapStatelessClient<K,
}
@Override
protected boolean eventReturnsNull(@NotNull EntrySetEventId methodName) {
protected boolean eventReturnsNull(@NotNull SetEventId methodName) {
return false;
}
@@ -147,7 +150,7 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends MapStatelessClient<K,
segment
}
enum EntrySetEventId implements ParameterizeWireKey {
enum SetEventId implements ParameterizeWireKey {
size,
isEmpty,
remove(key),
@@ -156,7 +159,7 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends MapStatelessClient<K,
private final WireKey[] params;
<P extends WireKey> EntrySetEventId(P... params) {
<P extends WireKey> SetEventId(P... params) {
this.params = params;
}
src/main/java/net/openhft/chronicle/map/MapStatelessClient.java
+19 -60
View File
@@ -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<K, V, E extends ParameterizeWireKey>
extends AbstactStatelessClient<E> {
Class<K> kClass;
Class<V> vClass;
public abstract class MapStatelessClient< E extends ParameterizeWireKey> extends AbstactStatelessClient<E> {
/**
* @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<K> kClass,
@NotNull final Class<V> vClass) {
long cid) {
super(channelName, hub, type, cid);
this.vClass = vClass;
this.kClass = kClass;
}
@Nullable
@@ -44,61 +31,33 @@ public abstract class MapStatelessClient<K, V, E extends ParameterizeWireKey>
@NotNull final Class<R> 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> R proxyReturnObject(@NotNull final Class<R> 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, O> E proxyReturnE(@NotNull final WireKey eventId,
@Nullable final O arg,
@NotNull final Class<E> 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> E proxyReturnE(@NotNull final WireKey eventId,
@NotNull final Class<E> 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);
}
}
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+4 -4
View File
@@ -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<K, V> implements WireHandler, Consumer<WireHandlers>
// 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<K, V> implements WireHandler, Consumer<WireHandlers>
// 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<K, V> implements WireHandler, Consumer<WireHandlers>
}
if (EntrySetEventId.numberOfSegements.contentEquals(eventName)) {
if (SetEventId.numberOfSegements.contentEquals(eventName)) {
write(b -> outWire.write(reply).int32(1));
return;
}