diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java new file mode 100644 index 000000000..d83c2e2f1 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleEntrySet.java @@ -0,0 +1,144 @@ +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.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 static net.openhft.chronicle.map.ClientWiredStatelessChronicleEntrySet.EventId.isEmpty; +import static net.openhft.chronicle.map.ClientWiredStatelessChronicleEntrySet.EventId.size; + + +class ClientWiredStatelessChronicleEntrySet extends AbstactStatelessClient + implements Set> { + + public ClientWiredStatelessChronicleEntrySet(@NotNull final String channelName, + @NotNull final ClientWiredStatelessTcpConnectionHub hub, + @NotNull final String type, + final long cid) { + super(channelName, hub, type, cid); + } + + @Override + protected Consumer toParameters(@NotNull ParameterizeWireKey eventId, Object... args) { + + return out -> { + final 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]); + } + + }); + + }; + } + + @Override + public int size() { + return proxyReturnInt(size); + } + + @Override + public boolean isEmpty() { + return proxyReturnBoolean(isEmpty); + } + + @Override + public boolean contains(Object o) { + return false; + } + + @NotNull + @Override + public Iterator> iterator() { + return null; + } + + @NotNull + @Override + public Object[] toArray() { + return new Object[0]; + } + + @NotNull + @Override + public T[] toArray(T[] a) { + return null; + } + + @Override + public boolean add(Map.Entry kvEntry) { + return false; + } + + @Override + public boolean remove(Object o) { + return false; + } + + @Override + public boolean containsAll(Collection c) { + return false; + } + + @Override + public boolean addAll(Collection> c) { + return false; + } + + @Override + public boolean retainAll(Collection c) { + return false; + } + + @Override + public boolean removeAll(Collection c) { + return false; + } + + @Override + public void clear() { + + } + + enum Params implements WireKey { + + } + + enum EventId implements ParameterizeWireKey { + size, + isEmpty; + + private final WireKey[] params; + +

EventId(P... params) { + this.params = params; + } + + public

P[] params() { + return (P[]) this.params; + + } + } +} \ No newline at end of file diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index 19ad60bab..f1fa151c1 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -18,16 +18,11 @@ 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.Marshallable; -import net.openhft.chronicle.wire.ValueIn; -import net.openhft.chronicle.wire.ValueOut; -import net.openhft.chronicle.wire.Wire; +import net.openhft.chronicle.wire.*; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; @@ -35,11 +30,16 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; -import java.util.*; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; import java.util.function.Consumer; 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.*; @@ -71,7 +71,7 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient extends AbstactStatelessClient MAX_NUM_ENTRIES) { + /*if (size() > MAX_NUM_ENTRIES) { StringBuilder s = new StringBuilder(toMap(MAX_NUM_ENTRIES).toString()); s.deleteCharAt(s.length() - 1).append(", ..."); return s.toString(); } else { return toMap().toString(); - } + }*/ + return ""; } @NotNull public String serverPersistedDataVersion() { - - return hub.proxyReturnString(persistedDataVersion, csp); + return hub.proxyReturnString(persistedDataVersion, csp, 0); } public boolean isEmpty() { @@ -287,7 +287,7 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient map) { - final long startTime = System.currentTimeMillis(); + /* final long startTime = System.currentTimeMillis(); long tid = hub.nextUniqueTransaction(startTime); Set> entries = map.entrySet(); @@ -340,6 +340,7 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient extends AbstactStatelessClient keySet() { + return null; + } @NotNull + @Override + public Collection values() { + return null; + } + + + /* @NotNull public Set keySet() { final Set usingCollection = new HashSet(); readChunked(kClass, keySet, usingCollection); @@ -379,9 +392,9 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient values() { final List usingCollection = new ArrayList<>(); return readChunked(vClass, values, usingCollection); - } + }*/ - private > Collection readChunked( + /*private > Collection readChunked( Class vClass1, EventId values, A usingCollection) { final long startTime = System.currentTimeMillis(); @@ -417,14 +430,36 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient cidToCsp = new HashMap<>(); @NotNull public Set> entrySet() { - return toMap().entrySet(); + + final long[] cidRef = new long[1]; + + proxyReturnWireConsumer(entrySet, (WireIn wireIn) -> { + final StringBuilder type = Wires.acquireStringBuilder(); + final ValueIn read = wireIn.read(reply); + read.type(type); + read.marshallable(w -> { + + final String csp1 = w.read(CoreFields.csp).text(); + final long cid = w.read(CoreFields.cid).int64(); + cidToCsp.put(cid, csp1); + cidRef[0] = cid; + + }); + }); + + final long cid = cidRef[0]; + return new ClientWiredStatelessChronicleEntrySet(channelName, hub, "entrySet", cid); } - private Map toMap() { + + /* private Map toMap() { final Map result = new HashMap(); final long startTime = System.currentTimeMillis(); // send @@ -451,10 +486,10 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient 0) { + *//* if (bytes.remaining() > 0) { // todo process the exception // boolean isException = wireIn.read(Fields.isException).bool(); - }*/ + }*//* } } finally { @@ -462,7 +497,7 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient extends AbstactStatelessClient toMap(int maxNumEntries) { + /* private Map toMap(int maxNumEntries) { final Map result = new HashMap(); final long startTime = System.currentTimeMillis(); @@ -513,7 +548,7 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient extends AbstactStatelessClient extends AbstactStatelessClient E readObject(Fields argName, Wire wireIn, E usingValue, Class clazz) { + private E readObject(WireKey argName, Wire wireIn, E usingValue, Class clazz) { final ValueIn valueIn = wireIn.read(argName); if (valueIn.isNull()) @@ -626,7 +661,7 @@ class ClientWiredStatelessChronicleMap extends AbstactStatelessClient extends AbstactStatelessClient extends AbstactStatelessClient extends AbstactStatelessClient implements WireHandler, Consumer @NotNull private final Supplier>> mapFactory; + private final Map cidToCsp; + private final Map cspToCid = new HashMap<>(); private Wire inWire = null; private Wire outWire = null; @@ -84,12 +90,28 @@ public class MapWireHandler implements WireHandler, Consumer } }; + + private void setCspTextFromCid(long cid) { + cspText.setLength(0); + cspText.append(cidToCsp.get(cid)); + } + private final Consumer metaDataConsumer = new Consumer() { @Override public void accept(WireIn wireIn) { - wireIn.read(csp).text(cspText); + StringBuilder key = Wires.acquireStringBuilder(); + final ValueIn read = wireIn.read(key); + + + if (csp.contentEquals(key)) { + read.text(cspText); + } else { + if (CoreFields.cid.contentEquals(key)) { + setCspTextFromCid(read.int64()); + } + } final int i = cspText.lastIndexOf("/"); @@ -110,7 +132,7 @@ public class MapWireHandler implements WireHandler, Consumer }; - private final Consumer writeEntry = new Consumer>>() { + /* private final Consumer writeEntry = new Consumer>>() { @Override public void accept(Iterator> iterator) { @@ -125,7 +147,7 @@ public class MapWireHandler implements WireHandler, Consumer Bytes bytes1 = outWire.bytes(); bytes1.write(entry.getValue()); } - }; + };*/ private WireHandlers publishLater; private byte localIdentifier; @@ -133,7 +155,7 @@ public class MapWireHandler implements WireHandler, Consumer private int channelId; private Map channels; private ReplicationHub hub; - private byte remoteIdentifier; + private ChronicleMap channelNameToId; public MapWireHandler( @@ -141,8 +163,9 @@ public class MapWireHandler implements WireHandler, Consumer @NotNull final Supplier>> channelNameToIdFactory, @NotNull final ReplicationHub hub, byte localIdentifier, - @NotNull final Map channels) { - this(mapFactory, channelNameToIdFactory, hub); + @NotNull final Map channels, + @NotNull final Map cidToCsp) { + this(mapFactory, channelNameToIdFactory, hub, cidToCsp); this.channels = channels; this.localIdentifier = localIdentifier; @@ -152,10 +175,12 @@ public class MapWireHandler implements WireHandler, Consumer @NotNull final Supplier>> mapFactory, @NotNull final Supplier>> channelNameToIdFactory, - @NotNull final ReplicationHub hub) { + @NotNull final ReplicationHub hub, + @NotNull final Map cidToCsp) { this.mapFactory = mapFactory; this.hub = hub; + this.cidToCsp = cidToCsp; try { channelNameToId = (ChronicleMap) channelNameToIdFactory.get() @@ -171,11 +196,11 @@ public class MapWireHandler implements WireHandler, Consumer } - private void writeChunked( +/* 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 @@ -183,7 +208,7 @@ public class MapWireHandler implements WireHandler, Consumer * @param maxEntries the maximum number of items that can be written * @throws StreamCorruptedException */ - private void writeChunked( + /*private void writeChunked( @NotNull final Function function, @NotNull final Consumer c, long maxEntries) throws StreamCorruptedException { @@ -208,9 +233,8 @@ public class MapWireHandler implements WireHandler, Consumer outWire.write(Fields.hasNext).bool(hasNext); - if (hasNext) { + if (hasNext) c.accept(iterator); - } }); @@ -231,7 +255,7 @@ public class MapWireHandler implements WireHandler, Consumer that.process(inWire, outWire); - } + }*/ @Override @@ -299,12 +323,35 @@ public class MapWireHandler implements WireHandler, Consumer public long tid; + private final AtomicLong cid = new AtomicLong(); + + + /** + * create a new cid if one does not already exist for this csp + * + * @param csp the csp we wish to check for a cid + * @return the cid for this csp + */ + private long createCid(CharSequence csp) { + + final long newCid = cid.incrementAndGet(); + final Long aLong = cspToCid.putIfAbsent(csp, newCid); + + if (aLong != null) + return aLong; + + cidToCsp.put(newCid, csp.toString()); + return newCid; + + } + + private final Consumer dataConsumer = new Consumer() { @Override public void accept(WireIn wireIn) { - final StringBuilder eventName = Wires.acquireStringBuilder(); + final StringBuilder eventName = acquireStringBuilder(); final ValueIn valueIn = inWire.readEventName(eventName); if (!incompleteWork.isEmpty()) { @@ -318,6 +365,8 @@ 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)); + try { if (put.contentEquals(eventName)) { @@ -335,54 +384,9 @@ public class MapWireHandler implements WireHandler, Consumer return; } - - if (keySet.contentEquals(eventName)) { - writeChunked(map -> map.keySet().iterator(), writeElement); - return; - } - - if (values.contentEquals(eventName)) { - writeChunked(map -> map.delegate.values().iterator(), writeElement); - return; - } - - if (entrySet.contentEquals(eventName)) { - writeChunked(new Function() { - @Override - public Iterator apply(BytesChronicleMap m) { - return m.delegate.entrySet().iterator(); - } - }, writeEntry); - 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; - } - - outWire.writeDocument(true, wire -> outWire.write(CoreFields.tid).int64(tid)); - - // write the transaction id - 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())); + // -- THESE METHODS ARE USED BOTH MY MAP AND ENTRY-SET + if (size.contentEquals(eventName)) { + write(b -> outWire.write(reply).int32(b.size())); return; } @@ -391,6 +395,67 @@ public class MapWireHandler implements WireHandler, Consumer return; } + if (endsWith(cspText, "#entrySet")) + throw new IllegalStateException("unsupported event=" + eventName); + + if (keySet.contentEquals(eventName)) { + throw new UnsupportedOperationException("todo"); + } + + if (values.contentEquals(eventName)) { + throw new UnsupportedOperationException("todo"); + } + + + if (entrySet.contentEquals(eventName)) { + write(b -> outWire.write(reply).type("set-proxy").writeValue() + + .marshallable(w -> { + + CharSequence root = cspText.subSequence(0, cspText + .length() - "#map".length()); + + final StringBuilder csp = acquireStringBuilder() + .append(root) + .append("#entrySet"); + + w.write(CoreFields.csp).text(csp); + w.write(CoreFields.cid).int64(createCid(csp)); + } + + )); + 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; + } + + if (containsKey.contentEquals(eventName)) { // todo remove the toByteArray(..) write(b -> outWire.write(reply) @@ -435,9 +500,7 @@ public class MapWireHandler implements WireHandler, Consumer } if (remove.contentEquals(eventName)) { - - MapWireHandler.this.writeValue(b -> b.remove(toByteArray(valueIn))); - + writeValue(b -> b.remove(toByteArray(valueIn))); return; } @@ -471,7 +534,7 @@ public class MapWireHandler implements WireHandler, Consumer final net.openhft.lang.io.Bytes reader = toReader(valueIn, replaceWithOldAndNewValue.params()); boolean result = bytesMap.replace(reader, reader, reader); - outWire.write(Fields.reply).bool(result); + outWire.write(reply).bool(result); }); return; @@ -571,6 +634,8 @@ public class MapWireHandler implements WireHandler, Consumer private void putAll(long tid) { + // todo +/* final BytesChronicleMap bytesMap = bytesMap(MapWireHandler.this.channelId); if (bytesMap == null) @@ -627,7 +692,7 @@ public class MapWireHandler implements WireHandler, Consumer }; incompleteWork.put(tid, runnable); - runnable.run(); + runnable.run();*/ } @@ -650,11 +715,6 @@ public class MapWireHandler implements WireHandler, Consumer }); } - private byte identifier() { - // if the client provides a remote identifier then we will use that otherwise we will use - // the local identifier - return remoteIdentifier == 0 ? localIdentifier : remoteIdentifier; - } @NotNull private CharSequence persistedDataVersion() { @@ -682,7 +742,6 @@ public class MapWireHandler implements WireHandler, Consumer final net.openhft.lang.io.Bytes bytes = DirectStore.allocate(inSize + 1024).bytes(); - valueIn.marshallable(wire -> { for (Params p : params) { @@ -703,11 +762,8 @@ public class MapWireHandler implements WireHandler, Consumer private byte[] toByteArray(ValueIn valueIn) { - final String text = valueIn.text(); return text.getBytes(); - - } /** @@ -726,7 +782,6 @@ public class MapWireHandler implements WireHandler, Consumer if (map != null) return map; - throw new IllegalStateException(); } @@ -774,7 +829,7 @@ public class MapWireHandler implements WireHandler, Consumer byte[] result = f.apply((ChronicleMap) b.delegate); boolean isNull = result == null || result.length == 0; - outWire.write(Fields.reply); + outWire.write(reply); final Bytes bytes = outWire.bytes(); if (isNull) @@ -802,16 +857,16 @@ public class MapWireHandler implements WireHandler, Consumer // bytes.mark(); // outWire.write(isException).bool(false); - outWire.writeDocument(false, out -> { + long position = 0; try { - outWire.bytes().mark(); + position = outWire.bytes().position(); c.accept(bytesMap); } catch (Exception e) { // bytes.reset(); // the idea of wire is that is platform independent, // so we wil have to send the exception as a String - outWire.bytes().reset(); + outWire.bytes().position(position); final WireOut o = out.write(reply) .type(e.getClass().getSimpleName()); @@ -958,22 +1013,5 @@ public class MapWireHandler implements WireHandler, Consumer } - // note : peter has asked for these to be in camel case - enum Fields implements WireKey { - hasNext, - timeStamp, - channelId, - eventName, - csp, - reply, - resultKey, - resultValue, - arg1, - arg2, - arg3, - isException, - exception, - resultIsNull - } }