diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index a48958023..19ad60bab 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -21,9 +21,13 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.Bytes; 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.map.MapWireHandler.Fields; -import net.openhft.chronicle.wire.*; +import net.openhft.chronicle.wire.Marshallable; +import net.openhft.chronicle.wire.ValueIn; +import net.openhft.chronicle.wire.ValueOut; +import net.openhft.chronicle.wire.Wire; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; @@ -43,19 +47,14 @@ import static net.openhft.chronicle.map.MapWireHandler.EventId.*; /** * @author Rob Austin. */ -class ClientWiredStatelessChronicleMap +class ClientWiredStatelessChronicleMap extends AbstactStatelessClient implements ChronicleMap, Cloneable, ChannelFactory { private static final Logger LOG = LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class); - private static final WriteMarshallable EMPTY = wire -> { - // nothing - }; - public static final Consumer VOID_PARAMETERS = out -> out.marshallable(EMPTY); - - private final ClientWiredStatelessTcpConnectionHub hub; + public static final Consumer VOID_PARAMETERS = out -> out.marshallable(AbstactStatelessClient.EMPTY); protected Class kClass; protected Class vClass; @@ -65,7 +64,6 @@ class ClientWiredStatelessChronicleMap // used with toString() private static final int MAX_NUM_ENTRIES = 20; - private String csp; public ClientWiredStatelessChronicleMap( @NotNull final ClientWiredChronicleMapStatelessBuilder config, @@ -73,9 +71,8 @@ class ClientWiredStatelessChronicleMap @NotNull final Class vClass, @NotNull final String channelName, ClientWiredStatelessTcpConnectionHub hub) { + super(channelName, hub); - this.csp = "//" + channelName + "#MAP"; - this.hub = hub; this.putReturnsNull = config.putReturnsNull(); this.removeReturnsNull = config.removeReturnsNull(); this.kClass = kClass; @@ -422,15 +419,6 @@ class ClientWiredStatelessChronicleMap return usingCollection; } - private void checkIsData(Wire wireIn) { - int datalen = wireIn.bytes().readVolatileInt(); - - // if (Wires.isData(datalen)) - // throw new IllegalStateException("expecting a data blob, from ->" + Bytes.toDebugString - // (wireIn.bytes(), 0, wireIn.bytes().limit())); - // - } - @NotNull public Set> entrySet() { return toMap().entrySet(); @@ -535,54 +523,6 @@ class ClientWiredStatelessChronicleMap } - private long readLong(long tid, long startTime) { - assert !hub.outBytesLock().isHeldByCurrentThread(); - final long timeoutTime = startTime + hub.timeoutMs; - - // receive - hub.inBytesLock().lock(); - try { - final Wire wire = hub.proxyReply(timeoutTime, tid); - checkIsData(wire); - return wire.read(MapWireHandler.Fields.reply).int64(); - } finally { - hub.inBytesLock().unlock(); - } - } - - - private void writeField(ValueOut wireOut, Object value) { - writeField(value, wireOut); - } - - - private void writeField(Object value, ValueOut valueOut) { - - assert hub.outBytesLock().isHeldByCurrentThread(); - assert !hub.inBytesLock().isHeldByCurrentThread(); - - - if (value instanceof Byte) - valueOut.int8((Byte) value); - else if (value instanceof Character) - valueOut.text(value.toString()); - else if (value instanceof Short) - valueOut.int16((Short) value); - else if (value instanceof Integer) - valueOut.int32((Integer) value); - else if (value instanceof Long) - valueOut.int64((Long) value); - else if (value instanceof CharSequence) { - valueOut.text((CharSequence) value); - } else if (value instanceof Marshallable) { - valueOut.marshallable((Marshallable) value); - } else { - throw new IllegalStateException("type=" + value.getClass() + - " is unsupported, it must either be of type Marshallable or CharSequence"); - } - } - - private V readValue(Fields argName, long tid, long startTime, final V usingValue) { assert !hub.outBytesLock().isHeldByCurrentThread(); long timeoutTime = startTime + hub.timeoutMs; @@ -676,22 +616,6 @@ class ClientWiredStatelessChronicleMap } - private boolean readBoolean(long tid, long startTime) { - assert !hub.outBytesLock().isHeldByCurrentThread(); - - long timeoutTime = startTime + hub.timeoutMs; - - // receive - hub.inBytesLock().lock(); - try { - final Wire wireIn = hub.proxyReply(timeoutTime, tid); - checkIsData(wireIn); - return wireIn.read(MapWireHandler.Fields.reply).bool(); - } finally { - hub.inBytesLock().unlock(); - } - } - private int readInt(long tid, long startTime) { assert !hub.outBytesLock().isHeldByCurrentThread(); @@ -709,15 +633,6 @@ class ClientWiredStatelessChronicleMap } - @SuppressWarnings("SameParameterValue") - private boolean proxyReturnBoolean( - @NotNull final EventId eventId, Object... args) { - final long startTime = System.currentTimeMillis(); - final long tid = sendEvent(startTime, eventId, toParameters(eventId, args)); - return readBoolean(tid, startTime); - - } - @SuppressWarnings("SameParameterValue") private boolean proxyReturnBooleanV(@NotNull final EventId eventId, V value) { final long startTime = System.currentTimeMillis(); @@ -731,59 +646,6 @@ class ClientWiredStatelessChronicleMap return readBoolean(sendEvent(startTime, eventId, out -> writeField(out, key)), startTime); } - @SuppressWarnings("SameParameterValue") - private long proxyReturnLong(@NotNull final EventId eventId) { - final long startTime = System.currentTimeMillis(); - long tid = sendEvent(startTime, eventId, null); - return readLong(tid, startTime); - } - - - @SuppressWarnings("SameParameterValue") - private void proxyReturnVoid(@NotNull final EventId eventId, - @Nullable final Consumer consumer) { - final long startTime = System.currentTimeMillis(); - long tid = sendEvent(startTime, eventId, consumer); - readVoid(tid, startTime); - } - - - @SuppressWarnings("SameParameterValue") - private void proxyReturnVoid(@NotNull final EventId eventId) { - proxyReturnVoid(eventId, null); - } - - - private long sendEvent(final long startTime, - @NotNull final EventId eventId, - @Nullable final Consumer consumer) { - long tid; - hub.outBytesLock().lock(); - try { - - tid = writeHeader(startTime); - hub.outWire().writeDocument(false, wireOut -> { - - final ValueOut valueOut = wireOut.writeEventName(eventId); - - if (consumer == null) - valueOut.marshallable(EMPTY); - else - consumer.accept(valueOut); - - }); - - hub.writeSocket(hub.outWire()); - - } finally { - hub.outBytesLock().unlock(); - } - return tid; - } - - private long writeHeader(long startTime) { - return hub.writeHeader(startTime, hub.outWire(), csp); - } @SuppressWarnings("SameParameterValue") private boolean proxyReturnBoolean(@NotNull final EventId eventId) { @@ -799,33 +661,6 @@ class ClientWiredStatelessChronicleMap } - private Consumer toParameters(@NotNull final EventId eventId, Object... args) { - - return out -> { - final MapWireHandler.Params[] paramNames = eventId.params(); - - if (paramNames.length == 1) { - writeField(out, args[0]); - return; - } - - assert args.length == paramNames.length : - "methodName=" + eventId + - ", args.length=" + args.length + - ", paramNames.length=" + paramNames.length; - - out.marshallable(m -> { - - for (int i = 0; i < paramNames.length; i++) { - final ValueOut vo = m.write(paramNames[i]); - ClientWiredStatelessChronicleMap.this.writeField(vo, args[i]); - } - - }); - - }; - } - @Nullable private R proxyReturnTypedObject( @NotNull final EventId eventId, Class resultType, Object... args) { @@ -910,17 +745,7 @@ class ClientWiredStatelessChronicleMap } - private void readVoid(long tid, long startTime) { - long timeoutTime = startTime + hub.timeoutMs; - // receive - hub.inBytesLock().lock(); - try { - hub.proxyReply(timeoutTime, tid); - } finally { - hub.inBytesLock().unlock(); - } - } class Entry implements Map.Entry { @@ -975,5 +800,33 @@ class ClientWiredStatelessChronicleMap return getKey() + "=" + getValue(); } } + + protected Consumer toParameters(@NotNull ParameterizeWireKey eventId, Object... args) { + + return out -> { + final MapWireHandler.Params[] paramNames = eventId.params(); + + if (paramNames.length == 1) { + writeField(out, args[0]); + return; + } + + assert args.length == paramNames.length : + "methodName=" + eventId + + ", args.length=" + args.length + + ", paramNames.length=" + paramNames.length; + + out.marshallable(m -> { + + for (int i = 0; i < paramNames.length; i++) { + final ValueOut vo = m.write(paramNames[i]); + this.writeField(vo, args[i]); + } + + }); + + }; + } + } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index f5eca0a81..57f8e85f6 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -24,6 +24,7 @@ package net.openhft.chronicle.map; 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.ChronicleHashInstanceBuilder; import net.openhft.chronicle.hash.impl.util.BuildVersion; import net.openhft.chronicle.hash.replication.ReplicationHub; @@ -323,7 +324,7 @@ public class MapWireHandler implements WireHandler, Consumer valueIn.marshallable(wire -> { - final Params[] params = putIfAbsent.params; + final Params[] params = putIfAbsent.params(); final byte[] key = wire.read(params[0]).bytes(); final byte[] value = wire.read(params[1]).bytes(); @@ -418,7 +419,7 @@ public class MapWireHandler implements WireHandler, Consumer valueIn.marshallable(wire -> { - final Params[] params = getAndPut.params; + final Params[] params = getAndPut.params(); final byte[] key1 = wire.read(params[0]).bytes(); final byte[] value1 = wire.read(params[1]).bytes(); @@ -453,7 +454,7 @@ public class MapWireHandler implements WireHandler, Consumer // todo bytesMap.replace(reader, reader, timestamp, identifier()); valueIn.marshallable(wire -> { - final Params[] params = replace.params; + final Params[] params = replace.params(); final byte[] key = wire.read(params[0]).bytes(); final byte[] value = wire.read(params[1]).bytes(); @@ -468,7 +469,7 @@ public class MapWireHandler implements WireHandler, Consumer if (replaceWithOldAndNewValue.contentEquals(eventName)) { write(bytesMap -> { final net.openhft.lang.io.Bytes reader = toReader(valueIn, - replaceWithOldAndNewValue.params); + replaceWithOldAndNewValue.params()); boolean result = bytesMap.replace(reader, reader, reader); outWire.write(Fields.reply).bool(result); }); @@ -478,7 +479,7 @@ public class MapWireHandler implements WireHandler, Consumer if (putIfAbsent.contentEquals(eventName)) { valueIn.marshallable(wire -> { - final Params[] params = putIfAbsent.params; + final Params[] params = putIfAbsent.params(); final byte[] key = wire.read(params[0]).bytes(); final byte[] value = wire.read(params[1]).bytes(); @@ -491,7 +492,7 @@ public class MapWireHandler implements WireHandler, Consumer if (removeWithValue.contentEquals(eventName)) { write(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(valueIn, removeWithValue.params); + final net.openhft.lang.io.Bytes reader = toReader(valueIn, removeWithValue.params()); // todo call outWire.write(result) // .bool(bytesMap.remove(reader, reader, timestamp, identifier())); outWire.write(reply).bool(bytesMap.remove(reader, reader)); @@ -911,7 +912,7 @@ public class MapWireHandler implements WireHandler, Consumer newValue } - enum EventId implements WireKey { + enum EventId implements ParameterizeWireKey { longSize, size, isEmpty, @@ -944,18 +945,14 @@ public class MapWireHandler implements WireHandler, Consumer createChannel, remoteIdentifier; - private final Params[] params; + private final WireKey[] params; - EventId(Params... params) { +

EventId(P... params) { this.params = params; } - public boolean contentEquals(CharSequence c) { - return this.toString().contentEquals(c); - } - - public Params[] params() { - return this.params; + public

P[] params() { + return (P[]) this.params; } }