mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
CE-23 Add entryset.remove(key) -> returns boolean
This commit is contained in:
+22
-12
@@ -12,25 +12,27 @@ import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import static net.openhft.chronicle.map.ClientWiredStatelessChronicleEntrySet.EventId.isEmpty;
|
||||
import static net.openhft.chronicle.map.ClientWiredStatelessChronicleEntrySet.EventId.size;
|
||||
import static net.openhft.chronicle.map.ClientWiredStatelessChronicleEntrySet.EventId.*;
|
||||
import static net.openhft.chronicle.map.ClientWiredStatelessChronicleEntrySet.Params.key;
|
||||
|
||||
|
||||
class ClientWiredStatelessChronicleEntrySet<K, V> extends AbstactStatelessClient
|
||||
class ClientWiredStatelessChronicleEntrySet<K, V> extends MapStatelessClient<K, V, ClientWiredStatelessChronicleEntrySet.EventId>
|
||||
implements Set<Map.Entry<K, V>> {
|
||||
|
||||
public ClientWiredStatelessChronicleEntrySet(@NotNull final String channelName,
|
||||
@NotNull final ClientWiredStatelessTcpConnectionHub hub,
|
||||
@NotNull final String type,
|
||||
final long cid) {
|
||||
super(channelName, hub, type, cid);
|
||||
final long cid,
|
||||
@NotNull final Class<V> vClass) {
|
||||
|
||||
super(channelName, hub, "entrySet", cid, vClass);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
protected Consumer<ValueOut> toParameters(@NotNull ParameterizeWireKey eventId, Object... args) {
|
||||
protected Consumer<ValueOut> toParameters(@NotNull EventId eventId, Object... args) {
|
||||
|
||||
return out -> {
|
||||
final Params[] paramNames = eventId.params();
|
||||
final WireKey[] paramNames = eventId.params();
|
||||
|
||||
if (paramNames.length == 1) {
|
||||
writeField(out, args[0]);
|
||||
@@ -38,7 +40,8 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends AbstactStatelessClient
|
||||
}
|
||||
|
||||
assert args.length == paramNames.length :
|
||||
"methodName=" + eventId +
|
||||
"looks like you are missing the paramameter in the EventId, EventId=" +
|
||||
eventId +
|
||||
", args.length=" + args.length +
|
||||
", paramNames.length=" + paramNames.length;
|
||||
|
||||
@@ -94,7 +97,7 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends AbstactStatelessClient
|
||||
|
||||
@Override
|
||||
public boolean remove(Object o) {
|
||||
return false;
|
||||
return proxyReturnBoolean(remove, o);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -122,13 +125,20 @@ class ClientWiredStatelessChronicleEntrySet<K, V> extends AbstactStatelessClient
|
||||
|
||||
}
|
||||
|
||||
enum Params implements WireKey {
|
||||
@Override
|
||||
protected boolean eventReturnsNull(@NotNull EventId methodName) {
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
enum Params implements WireKey {
|
||||
key
|
||||
}
|
||||
|
||||
enum EventId implements ParameterizeWireKey {
|
||||
size,
|
||||
isEmpty;
|
||||
isEmpty,
|
||||
remove(key);
|
||||
|
||||
private final WireKey[] params;
|
||||
|
||||
|
||||
@@ -20,7 +20,6 @@ package net.openhft.chronicle.map;
|
||||
|
||||
import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub;
|
||||
import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields;
|
||||
import net.openhft.chronicle.engine.client.ParameterizeWireKey;
|
||||
import net.openhft.chronicle.hash.function.SerializableFunction;
|
||||
import net.openhft.chronicle.wire.*;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
@@ -39,7 +38,6 @@ 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.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields.tid;
|
||||
import static net.openhft.chronicle.map.MapWireHandler.EventId;
|
||||
import static net.openhft.chronicle.map.MapWireHandler.EventId.*;
|
||||
|
||||
@@ -47,7 +45,7 @@ import static net.openhft.chronicle.map.MapWireHandler.EventId.*;
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<EventId>
|
||||
class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, EventId>
|
||||
implements ChronicleMap<K, V>, Cloneable, ChannelFactory {
|
||||
|
||||
private static final Logger LOG =
|
||||
@@ -57,7 +55,6 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
public static final Consumer<ValueOut> VOID_PARAMETERS = out -> out.marshallable(AbstactStatelessClient.EMPTY);
|
||||
|
||||
protected Class<K> kClass;
|
||||
protected Class<V> vClass;
|
||||
|
||||
private boolean putReturnsNull;
|
||||
private boolean removeReturnsNull;
|
||||
@@ -71,12 +68,11 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
@NotNull final Class vClass,
|
||||
@NotNull final String channelName,
|
||||
ClientWiredStatelessTcpConnectionHub hub) {
|
||||
super(channelName, hub, "MAP", 0);
|
||||
super(channelName, hub, "MAP", 0, vClass);
|
||||
|
||||
this.putReturnsNull = config.putReturnsNull();
|
||||
this.removeReturnsNull = config.removeReturnsNull();
|
||||
this.kClass = kClass;
|
||||
this.vClass = vClass;
|
||||
}
|
||||
|
||||
|
||||
@@ -97,7 +93,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
|
||||
@Override
|
||||
public V newValueInstance() {
|
||||
return VanillaChronicleMap.newInstance(vClass, false);
|
||||
return (V) VanillaChronicleMap.newInstance(vClass, false);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -149,7 +145,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
if (key == null || value == null)
|
||||
throw new NullPointerException();
|
||||
|
||||
return proxyReturnTypedObject(putIfAbsent, vClass, key, value);
|
||||
return (V) proxyReturnTypedObject(putIfAbsent, vClass, key, value);
|
||||
}
|
||||
|
||||
@SuppressWarnings("NullableProblems")
|
||||
@@ -171,7 +167,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
public V replace(K key, V value) {
|
||||
if (key == null || value == null)
|
||||
throw new NullPointerException();
|
||||
return proxyReturnTypedObject(replace, vClass, key, value);
|
||||
return (V) proxyReturnTypedObject(replace, vClass, key, value);
|
||||
}
|
||||
|
||||
public int size() {
|
||||
@@ -253,7 +249,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
}
|
||||
|
||||
public V get(Object key) {
|
||||
return proxyReturnObject(vClass, get, (K) key);
|
||||
return (V) proxyReturnObject(vClass, get, (K) key);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
@@ -276,7 +272,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
public V remove(Object key) {
|
||||
if (key == null)
|
||||
throw keyNotNullNPE();
|
||||
return proxyReturnObject(vClass, removeReturnsNull ? removeWithoutAcc : remove, (K) key);
|
||||
return (V) proxyReturnObject(vClass, removeReturnsNull ? removeWithoutAcc : remove, key);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -347,7 +343,9 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
public V put(K key, V value) {
|
||||
if (key == null || value == null)
|
||||
throw new NullPointerException();
|
||||
return proxyReturnTypedObject(putReturnsNull ? put : getAndPut, vClass, key,
|
||||
return (V) proxyReturnTypedObject(putReturnsNull ? put : getAndPut,
|
||||
vClass,
|
||||
key,
|
||||
value);
|
||||
}
|
||||
|
||||
@@ -455,7 +453,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
});
|
||||
|
||||
final long cid = cidRef[0];
|
||||
return new ClientWiredStatelessChronicleEntrySet<K, V>(channelName, hub, "entrySet", cid);
|
||||
return new ClientWiredStatelessChronicleEntrySet<K, V>(channelName, hub, cid, vClass);
|
||||
}
|
||||
|
||||
|
||||
@@ -558,96 +556,8 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
}
|
||||
|
||||
|
||||
private 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);
|
||||
|
||||
|
||||
return readV(argName, wireIn, usingValue);
|
||||
|
||||
} finally {
|
||||
hub.inBytesLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private V readV(WireKey argName, Wire wireIn, V usingValue) {
|
||||
return readObject(argName, wireIn, usingValue, vClass);
|
||||
}
|
||||
|
||||
private K readK(WireKey argName, Wire wireIn, K usingValue) {
|
||||
return readObject(argName, wireIn, usingValue, kClass);
|
||||
}
|
||||
|
||||
private <E> E readObject(WireKey argName, Wire wireIn, E usingValue, Class<E> clazz) {
|
||||
|
||||
final ValueIn valueIn = wireIn.read(argName);
|
||||
if (valueIn.isNull())
|
||||
return null;
|
||||
|
||||
if (StringBuilder.class.isAssignableFrom(clazz)) {
|
||||
valueIn.text((StringBuilder) usingValue);
|
||||
return usingValue;
|
||||
} else if (Marshallable.class.isAssignableFrom(clazz)) {
|
||||
|
||||
|
||||
final E v;
|
||||
if (usingValue == null)
|
||||
try {
|
||||
v = clazz.newInstance();
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
else
|
||||
v = usingValue;
|
||||
|
||||
|
||||
valueIn.marshallable((Marshallable) v);
|
||||
return v;
|
||||
|
||||
} else if (String.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) valueIn.text();
|
||||
|
||||
} else if (Long.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Long) valueIn.int64();
|
||||
} else if (Double.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Double) valueIn.float64();
|
||||
|
||||
} else if (Integer.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Integer) valueIn.int32();
|
||||
|
||||
} else if (Float.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Float) valueIn.float32();
|
||||
|
||||
} else if (Short.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Short) valueIn.int16();
|
||||
|
||||
} else if (Character.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
final String text = valueIn.text();
|
||||
if (text == null || text.length() == 0)
|
||||
return null;
|
||||
return (E) (Character) text.charAt(0);
|
||||
|
||||
} else if (Byte.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Byte) valueIn.int8();
|
||||
|
||||
|
||||
} else {
|
||||
throw new IllegalStateException("unsupported type");
|
||||
}
|
||||
return (K) readObject(argName, wireIn, usingValue, kClass);
|
||||
}
|
||||
|
||||
|
||||
@@ -696,42 +606,6 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
}
|
||||
|
||||
|
||||
@Nullable
|
||||
private <R> R proxyReturnTypedObject(
|
||||
@NotNull final EventId eventId, Class<V> resultType, Object... args) {
|
||||
|
||||
final long startTime = System.currentTimeMillis();
|
||||
final long tid = sendEvent(startTime, eventId, toParameters(eventId, args));
|
||||
|
||||
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
|
||||
private <R> R proxyReturnObject(
|
||||
Class<R> rClass, @NotNull final EventId eventId, Object key) {
|
||||
|
||||
final long startTime = System.currentTimeMillis();
|
||||
final long tid = sendEvent(startTime, eventId, out -> writeField(out, key));
|
||||
|
||||
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");
|
||||
}
|
||||
|
||||
|
||||
private long proxySend(EventId methodName, long startTime) {
|
||||
long tid;
|
||||
hub.outBytesLock().lock();
|
||||
@@ -745,6 +619,22 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
return tid;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
protected boolean eventReturnsNull(@NotNull EventId methodName) {
|
||||
|
||||
switch (methodName) {
|
||||
case putAllWithoutAcc:
|
||||
case put:
|
||||
case removeWithoutAcc:
|
||||
return true;
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
/* private long proxySend(EventId methodName, long arg1, long startTime) {
|
||||
long tid;
|
||||
hub.outBytesLock().lock();
|
||||
@@ -767,22 +657,6 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
}*/
|
||||
|
||||
|
||||
private boolean eventReturnsNull(@NotNull EventId methodName) {
|
||||
|
||||
switch (methodName) {
|
||||
case putAllWithoutAcc:
|
||||
case put:
|
||||
case removeWithoutAcc:
|
||||
return true;
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
class Entry implements Map.Entry<K, V> {
|
||||
|
||||
final K key;
|
||||
@@ -836,7 +710,8 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
}
|
||||
}
|
||||
|
||||
protected Consumer<ValueOut> toParameters(@NotNull ParameterizeWireKey eventId, Object... args) {
|
||||
@Override
|
||||
protected Consumer<ValueOut> toParameters(@NotNull EventId eventId, Object... args) {
|
||||
|
||||
return out -> {
|
||||
final MapWireHandler.Params[] paramNames = eventId.params();
|
||||
@@ -863,5 +738,6 @@ class ClientWiredStatelessChronicleMap<K, V> extends AbstactStatelessClient<Even
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub;
|
||||
import net.openhft.chronicle.engine.client.ParameterizeWireKey;
|
||||
import net.openhft.chronicle.wire.Marshallable;
|
||||
import net.openhft.chronicle.wire.ValueIn;
|
||||
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> {
|
||||
|
||||
protected 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
|
||||
*/
|
||||
public MapStatelessClient(@NotNull String channelName,
|
||||
@NotNull ClientWiredStatelessTcpConnectionHub hub,
|
||||
@NotNull String type,
|
||||
long cid, @NotNull final Class vClass) {
|
||||
super(channelName, hub, type, cid);
|
||||
this.vClass = vClass;
|
||||
}
|
||||
|
||||
protected Object 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);
|
||||
|
||||
|
||||
return readV(argName, wireIn, usingValue);
|
||||
|
||||
} finally {
|
||||
hub.inBytesLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private Object readV(WireKey argName, Wire wireIn, V usingValue) {
|
||||
return readObject(argName, wireIn, usingValue, vClass);
|
||||
}
|
||||
|
||||
protected <E> E readObject(WireKey argName, Wire wireIn, E usingValue, Class<E> clazz) {
|
||||
|
||||
final ValueIn valueIn = wireIn.read(argName);
|
||||
if (valueIn.isNull())
|
||||
return null;
|
||||
|
||||
if (StringBuilder.class.isAssignableFrom(clazz)) {
|
||||
valueIn.text((StringBuilder) usingValue);
|
||||
return usingValue;
|
||||
} else if (Marshallable.class.isAssignableFrom(clazz)) {
|
||||
|
||||
|
||||
final E v;
|
||||
if (usingValue == null)
|
||||
try {
|
||||
v = clazz.newInstance();
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
else
|
||||
v = usingValue;
|
||||
|
||||
|
||||
valueIn.marshallable((Marshallable) v);
|
||||
return v;
|
||||
|
||||
} else if (String.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) valueIn.text();
|
||||
|
||||
} else if (Long.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Long) valueIn.int64();
|
||||
} else if (Double.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Double) valueIn.float64();
|
||||
|
||||
} else if (Integer.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Integer) valueIn.int32();
|
||||
|
||||
} else if (Float.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Float) valueIn.float32();
|
||||
|
||||
} else if (Short.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Short) valueIn.int16();
|
||||
|
||||
} else if (Character.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
final String text = valueIn.text();
|
||||
if (text == null || text.length() == 0)
|
||||
return null;
|
||||
return (E) (Character) text.charAt(0);
|
||||
|
||||
} else if (Byte.class.isAssignableFrom(clazz)) {
|
||||
//noinspection unchecked
|
||||
return (E) (Byte) valueIn.int8();
|
||||
|
||||
|
||||
} else {
|
||||
throw new IllegalStateException("unsupported type");
|
||||
}
|
||||
}
|
||||
|
||||
@Nullable
|
||||
protected <R> R proxyReturnTypedObject(
|
||||
@NotNull final E eventId,
|
||||
@NotNull final Class<R> resultType,
|
||||
@Nullable Object... args) {
|
||||
|
||||
final long startTime = System.currentTimeMillis();
|
||||
final long tid = sendEvent(startTime, eventId, toParameters(eventId, args));
|
||||
|
||||
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(Class<R> rClass, @NotNull final E eventId, Object field) {
|
||||
|
||||
final long startTime = System.currentTimeMillis();
|
||||
final long tid = sendEvent(startTime, eventId, out -> writeField(out, field));
|
||||
|
||||
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);
|
||||
|
||||
|
||||
}
|
||||
@@ -45,7 +45,6 @@ import java.io.StringWriter;
|
||||
import java.nio.BufferOverflowException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
@@ -80,7 +79,7 @@ public class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers>
|
||||
private Wire inWire = null;
|
||||
private Wire outWire = null;
|
||||
|
||||
private final Consumer writeElement = new Consumer<Iterator<byte[]>>() {
|
||||
/* private final Consumer writeElement = new Consumer<Iterator<byte[]>>() {
|
||||
|
||||
@Override
|
||||
public void accept(Iterator<byte[]> iterator) {
|
||||
@@ -88,7 +87,7 @@ public class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers>
|
||||
Bytes<?> bytes = outWire.bytes();
|
||||
bytes.write(iterator.next());
|
||||
}
|
||||
};
|
||||
};*/
|
||||
|
||||
|
||||
private void setCspTextFromCid(long cid) {
|
||||
@@ -113,10 +112,12 @@ public class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers>
|
||||
}
|
||||
}
|
||||
|
||||
final int i = cspText.lastIndexOf("/");
|
||||
final int slash = cspText.lastIndexOf("/");
|
||||
final int hash = cspText.lastIndexOf("#");
|
||||
|
||||
if (i != -1 && i < (cspText.length() - 1)) {
|
||||
final String channelStr = cspText.substring(i + 1, cspText.length() - "#MAP".length());
|
||||
if (slash != -1 && slash < (cspText.length() - 1) &&
|
||||
hash != -1 && hash < (cspText.length() - 1)) {
|
||||
final String channelStr = cspText.substring(slash + 1, hash);
|
||||
try {
|
||||
channelId = MapWireHandler.this.acquireChannelId(channelStr);
|
||||
} catch (IOException e) {
|
||||
@@ -196,66 +197,6 @@ public class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers>
|
||||
}
|
||||
|
||||
|
||||
/* private void writeChunked(
|
||||
@NotNull final Function<BytesChronicleMap, Iterator> function,
|
||||
@NotNull final Consumer<Iterator> c) throws StreamCorruptedException {
|
||||
writeChunked(function, c, Long.MAX_VALUE);
|
||||
}*/
|
||||
|
||||
/**
|
||||
* @param function provides that returns items bases on a BytesChronicleMap
|
||||
* @param c an iterator that contains items
|
||||
* @param maxEntries the maximum number of items that can be written
|
||||
* @throws StreamCorruptedException
|
||||
*/
|
||||
/*private void writeChunked(
|
||||
@NotNull final Function<BytesChronicleMap, Iterator> function,
|
||||
@NotNull final Consumer<Iterator> c, long maxEntries) throws StreamCorruptedException {
|
||||
|
||||
final BytesChronicleMap m = bytesMap(channelId);
|
||||
final Iterator iterator = function.apply(m);
|
||||
|
||||
final WireHandler that = new WireHandler() {
|
||||
|
||||
@Override
|
||||
public void process(Wire in, Wire out) throws StreamCorruptedException {
|
||||
|
||||
outWire.write(CoreFields.tid).int64(tid);
|
||||
|
||||
// this allows us to write more data than the buffer will allow
|
||||
for (int count = 0; ; count++) {
|
||||
|
||||
boolean finished = count == maxEntries;
|
||||
|
||||
final boolean hasNext = iterator.hasNext() && !finished;
|
||||
|
||||
write(map -> {
|
||||
|
||||
outWire.write(Fields.hasNext).bool(hasNext);
|
||||
|
||||
if (hasNext)
|
||||
c.accept(iterator);
|
||||
|
||||
});
|
||||
|
||||
if (!hasNext)
|
||||
return;
|
||||
|
||||
// quit if we have filled the buffer
|
||||
Bytes<?> bytes = outWire.bytes();
|
||||
if (bytes.remaining() < (bytes.capacity() * 0.75)) {
|
||||
publishLater.add(this);
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
that.process(inWire, outWire);
|
||||
|
||||
}*/
|
||||
|
||||
|
||||
@Override
|
||||
@@ -363,7 +304,6 @@ public class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers>
|
||||
}
|
||||
|
||||
final Bytes<?> outBytes = outWire.bytes();
|
||||
final Bytes<?> bytes = outBytes;
|
||||
|
||||
outWire.writeDocument(true, wire -> outWire.write(CoreFields.tid).int64(tid));
|
||||
|
||||
@@ -391,12 +331,29 @@ public class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers>
|
||||
}
|
||||
|
||||
if (isEmpty.contentEquals(eventName)) {
|
||||
write(b -> outWire.write(reply).bool(b.isEmpty()));
|
||||
write(b -> {
|
||||
final boolean result = b.isEmpty();
|
||||
outWire.write(reply).bool(result);
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if (endsWith(cspText, "#entrySet"))
|
||||
// -- THESE METHODS ARE ONLY ENTRY SET METHODS
|
||||
if (endsWith(cspText, "#entrySet")) {
|
||||
|
||||
// not remove on the key-set returns a boolean and on the map returns the old
|
||||
// value
|
||||
if (remove.contentEquals(eventName)) {
|
||||
write(b -> outWire.write(reply).bool(
|
||||
b.delegate.remove(toByteArray(valueIn)) != null));
|
||||
return;
|
||||
}
|
||||
|
||||
throw new IllegalStateException("unsupported event=" + eventName);
|
||||
}
|
||||
|
||||
|
||||
// -- THESE METHODS ARE ONLY MAP METHODS
|
||||
|
||||
if (keySet.contentEquals(eventName)) {
|
||||
throw new UnsupportedOperationException("todo");
|
||||
@@ -427,29 +384,13 @@ public class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers>
|
||||
return;
|
||||
}
|
||||
|
||||
/* if (entrySetRestricted.contentEquals(eventName)) {
|
||||
long maxEntries = inWire.read(arg1).int64();
|
||||
writeChunked(m -> m.delegate.entrySet().iterator(), writeEntry, maxEntries);
|
||||
return;
|
||||
}
|
||||
*/
|
||||
|
||||
if (putAll.contentEquals(eventName)) {
|
||||
putAll(tid);
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/* if (createChannel.contentEquals(eventName)) {
|
||||
writeVoid(() -> {
|
||||
short channelId1 = valueIn.int16();
|
||||
mapFactory.get().replicatedViaChannel(hub.createChannel(channelId1)).create();
|
||||
return null;
|
||||
});
|
||||
return;
|
||||
}*/
|
||||
|
||||
|
||||
if (longSize.contentEquals(eventName)) {
|
||||
write(b -> outWire.write(reply).int64(b.longSize()));
|
||||
return;
|
||||
@@ -586,7 +527,7 @@ public class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers>
|
||||
} finally {
|
||||
|
||||
if (EventGroup.IS_DEBUG) {
|
||||
long len = bytes.position() - SIZE_OF_SIZE;
|
||||
long len = outBytes.position() - SIZE_OF_SIZE;
|
||||
if (len == 0) {
|
||||
System.out.println("--------------------------------------------\n" +
|
||||
"server writes:\n\n<EMPTY>");
|
||||
@@ -595,7 +536,7 @@ public class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers>
|
||||
|
||||
System.out.println("--------------------------------------------\n" +
|
||||
"server writes:\n\n" +
|
||||
Wires.fromSizePrefixedBlobs(bytes, SIZE_OF_SIZE, len));
|
||||
Wires.fromSizePrefixedBlobs(outBytes, SIZE_OF_SIZE, len));
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user