From e1fc723ae4029e9be09b31d8103a062eed907250 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 16 Mar 2015 10:39:29 +0000 Subject: [PATCH] CE-14 stateless wire map - toString() is now restricted to 100 entries --- .../map/ClientWiredStatelessChronicleMap.java | 69 ++++++++++++++++++- .../openhft/chronicle/map/MapWireHandler.java | 31 +++++++-- 2 files changed, 95 insertions(+), 5 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index ac6873685..979ebff9c 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -199,7 +199,12 @@ class ClientWiredStatelessChronicleMap @NotNull public String toString() { - return toMap().toString(); + int maxNumEntries = 100; + if (size() > 100) { + return toMap(maxNumEntries).toString() + "...."; + } else { + return toMap().toString(); + } } @NotNull @@ -438,6 +443,53 @@ class ClientWiredStatelessChronicleMap return result; } + + /** + * returns a chunk of entries + * + * @return + * @param maxNumEntries + */ + private Map toMap(int maxNumEntries) { + final Map result = new HashMap(); + final long startTime = System.currentTimeMillis(); + + // send + final long transactionId = proxySend(entrySetRestricted, maxNumEntries, startTime); + assert !hub.outBytesLock().isHeldByCurrentThread(); + final long timeoutTime = startTime + hub.timeoutMs; + + OUTER: + for (; ; ) { + + // receive + hub.inBytesLock().lock(); + try { + final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + + while (wireIn.bytes().remaining() > 0) { + boolean bool = wireIn.read(Fields.hasNext).bool(); + if (bool) { + result.put( + readK(Fields.resultKey, wireIn, null), + readV(Fields.resultValue, wireIn, null)); + } else + break OUTER; + + + if (wireIn.bytes().remaining() > 0) { + // todo process the exception + boolean isException = wireIn.read(Fields.isException).bool(); + } + } + + } finally { + hub.inBytesLock().unlock(); + } + } + return result; + } + /** * @param callback each entry is passed to the callback */ @@ -819,6 +871,21 @@ class ClientWiredStatelessChronicleMap return transactionId; } + private long proxySend(EventId methodName, long arg1, long startTime) { + long transactionId; + hub.outBytesLock().lock(); + try { + transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + hub.outWire().write(Fields.methodName).text(methodName.toString()); + hub.outWire().write(Fields.arg1).int64(arg1); + hub.writeSocket(hub.outWire()); + } finally { + hub.outBytesLock().unlock(); + } + return transactionId; + } + + private boolean eventReturnsNull(@NotNull EventId methodName) { switch (methodName) { diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 456fdc2d1..48b31d924 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -117,12 +117,19 @@ class MapWireHandler implements WireHandler, Consumer { this.publishLater = wireHandlers; } + /** + * @param transactionId the transaction id + * @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( long transactionId, @NotNull final Function function, - @NotNull final Consumer c) throws StreamCorruptedException { + @NotNull final Consumer c, long maxEntries) throws StreamCorruptedException { final BytesChronicleMap m = bytesMap(channelId); - final Iterator iterator = function.apply(m); + final Iterator iterator = function.apply(m); final WireHandler that = new WireHandler() { @@ -132,7 +139,7 @@ class MapWireHandler implements WireHandler, Consumer { outWire.write(Fields.transactionId).int64(transactionId); // this allows us to write more data than the buffer will allow - for (; ; ) { + for (int count = 0; count < maxEntries; count++) { final boolean hasNext = iterator.hasNext(); @@ -140,8 +147,10 @@ class MapWireHandler implements WireHandler, Consumer { outWire.write(Fields.hasNext).bool(hasNext); - if (hasNext) + if (hasNext) { c.accept(iterator); + } + }); if (!hasNext) @@ -162,6 +171,13 @@ class MapWireHandler implements WireHandler, Consumer { } + private void writeChunked( + long transactionId, @NotNull final Function function, + @NotNull final Consumer c) throws StreamCorruptedException { + writeChunked(transactionId, function, c, Long.MAX_VALUE); + } + + @Override public void process(Wire in, Wire out) throws StreamCorruptedException { try { @@ -218,6 +234,12 @@ class MapWireHandler implements WireHandler, Consumer { return; } + if (entrySetRestricted.contentEquals(methodName)) { + long maxEntries = inWire.read(arg1).int64(); + writeChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry,maxEntries); + return; + } + if (putAll.contentEquals(methodName)) { putAll(transactionId); return; @@ -803,6 +825,7 @@ class MapWireHandler implements WireHandler, Consumer { keySet, values, entrySet, + entrySetRestricted, replace, replaceWithOldAndNewValue, putIfAbsent,