diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java index d83c2e2f1..6db8641c4 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java @@ -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 extends AbstactStatelessClient +class ClientWiredStatelessChronicleEntrySet extends MapStatelessClient implements Set> { 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 vClass) { + + super(channelName, hub, "entrySet", cid, vClass); } + @Override - protected Consumer toParameters(@NotNull ParameterizeWireKey eventId, Object... args) { + protected Consumer 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 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 extends AbstactStatelessClient @Override public boolean remove(Object o) { - return false; + return proxyReturnBoolean(remove, o); } @Override @@ -122,13 +125,20 @@ class ClientWiredStatelessChronicleEntrySet 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; diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index f1fa151c1..39de810e5 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -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 extends AbstactStatelessClient +class ClientWiredStatelessChronicleMap extends MapStatelessClient implements ChronicleMap, Cloneable, ChannelFactory { private static final Logger LOG = @@ -57,7 +55,6 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient VOID_PARAMETERS = out -> out.marshallable(AbstactStatelessClient.EMPTY); protected Class kClass; - protected Class vClass; private boolean putReturnsNull; private boolean removeReturnsNull; @@ -71,12 +68,11 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient extends AbstactStatelessClient extends AbstactStatelessClient extends AbstactStatelessClient extends AbstactStatelessClient extends AbstactStatelessClient extends AbstactStatelessClient extends AbstactStatelessClient(channelName, hub, "entrySet", cid); + return new ClientWiredStatelessChronicleEntrySet(channelName, hub, cid, vClass); } @@ -558,96 +556,8 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient E readObject(WireKey argName, Wire wireIn, E usingValue, Class 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 extends AbstactStatelessClient R proxyReturnTypedObject( - @NotNull final EventId eventId, Class 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 proxyReturnObject( - Class 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 extends AbstactStatelessClient extends AbstactStatelessClient { final K key; @@ -836,7 +710,8 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient toParameters(@NotNull ParameterizeWireKey eventId, Object... args) { + @Override + protected Consumer toParameters(@NotNull EventId eventId, Object... args) { return out -> { final MapWireHandler.Params[] paramNames = eventId.params(); @@ -863,5 +738,6 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient extends + AbstactStatelessClient { + + protected 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 + */ + 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 readObject(WireKey argName, Wire wireIn, E usingValue, Class 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 proxyReturnTypedObject( + @NotNull final E eventId, + @NotNull final Class 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 proxyReturnObject(Class 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); + + +} diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index db6fe4fdb..810610bc2 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -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 implements WireHandler, Consumer private Wire inWire = null; private Wire outWire = null; - private final Consumer writeElement = new Consumer>() { +/* private final Consumer writeElement = new Consumer>() { @Override public void accept(Iterator iterator) { @@ -88,7 +87,7 @@ public class MapWireHandler implements WireHandler, Consumer Bytes bytes = outWire.bytes(); bytes.write(iterator.next()); } - }; + };*/ private void setCspTextFromCid(long cid) { @@ -113,10 +112,12 @@ public class MapWireHandler implements WireHandler, Consumer } } - 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 implements WireHandler, Consumer } -/* private void writeChunked( - @NotNull final Function function, - @NotNull final Consumer 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 function, - @NotNull final Consumer 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 implements WireHandler, Consumer } 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 implements WireHandler, Consumer } 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 implements WireHandler, Consumer 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 implements WireHandler, Consumer } 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"); @@ -595,7 +536,7 @@ public class MapWireHandler implements WireHandler, Consumer System.out.println("--------------------------------------------\n" + "server writes:\n\n" + - Wires.fromSizePrefixedBlobs(bytes, SIZE_OF_SIZE, len)); + Wires.fromSizePrefixedBlobs(outBytes, SIZE_OF_SIZE, len)); } }