From baf9013904948d82433492f8e966b6302ad5d0e3 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 12 Mar 2015 21:26:32 +0000 Subject: [PATCH] commits some fixes but still and issue with "wire entry set" --- ...entWiredChronicleMapStatelessBuilder.java} | 28 ++++++------- .../map/ClientWiredStatelessChronicleMap.java | 37 +++++++++++++++-- ...ClientWiredStatelessTcpConnectionHub.java} | 6 +-- .../openhft/chronicle/map/MapWireHandler.java | 41 +++++++++++++++---- 4 files changed, 85 insertions(+), 27 deletions(-) rename src/main/java/net/openhft/chronicle/map/{ClientWiredChronicleMapStatelessClientBuilder.java => ClientWiredChronicleMapStatelessBuilder.java} (72%) rename src/main/java/net/openhft/chronicle/map/{ClientWiredStatelessClientTcpConnectionHub.java => ClientWiredStatelessTcpConnectionHub.java} (98%) diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessClientBuilder.java b/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java similarity index 72% rename from src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessClientBuilder.java rename to src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java index 12481552a..f8c3fa5c3 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessClientBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java @@ -19,7 +19,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHashStatelessClientBuilder; -import net.openhft.chronicle.map.ChronicleMap;import net.openhft.chronicle.map.ClientWiredStatelessChronicleMap;import net.openhft.chronicle.map.ClientWiredStatelessClientTcpConnectionHub;import net.openhft.chronicle.map.MapBuilder;import net.openhft.lang.MemoryUnit; +import net.openhft.lang.MemoryUnit; import java.io.IOException; import java.lang.Class;import java.lang.IllegalStateException;import java.lang.Override;import java.lang.String;import java.net.InetSocketAddress; @@ -29,26 +29,26 @@ import java.util.concurrent.atomic.AtomicBoolean; /** * @author Rob Austin. */ - public final class ClientWiredChronicleMapStatelessClientBuilder implements - ChronicleHashStatelessClientBuilder, + public final class ClientWiredChronicleMapStatelessBuilder implements + ChronicleHashStatelessClientBuilder, ChronicleMap>, - MapBuilder> { + MapBuilder> { - ClientWiredStatelessClientTcpConnectionHub hub; + ClientWiredStatelessTcpConnectionHub hub; private Class keyClass; private Class valueClass; private byte localIdentifier; private short channelID; private boolean doHandShaking; - public ClientWiredChronicleMapStatelessClientBuilder(InetSocketAddress remoteAddress, Class keyClass, Class valueClass, short channelID) { + public ClientWiredChronicleMapStatelessBuilder(InetSocketAddress remoteAddress, Class keyClass, Class valueClass, short channelID) { this.keyClass = keyClass; this.valueClass = valueClass; this.remoteAddress = remoteAddress; this.channelID = channelID; } - public ClientWiredChronicleMapStatelessClientBuilder(ClientWiredStatelessClientTcpConnectionHub hub, Class keyClass, Class valueClass, short channelID) { + public ClientWiredChronicleMapStatelessBuilder(ClientWiredStatelessTcpConnectionHub hub, Class keyClass, Class valueClass, short channelID) { this.keyClass = keyClass; this.valueClass = valueClass; this.hub = hub; @@ -79,7 +79,7 @@ import java.util.concurrent.atomic.AtomicBoolean; } @Override - public ClientWiredChronicleMapStatelessClientBuilder putReturnsNull(boolean putReturnsNull) { + public ClientWiredChronicleMapStatelessBuilder putReturnsNull(boolean putReturnsNull) { this.putReturnsNull = putReturnsNull; return this; } @@ -89,7 +89,7 @@ import java.util.concurrent.atomic.AtomicBoolean; } @Override - public ClientWiredChronicleMapStatelessClientBuilder removeReturnsNull(boolean removeReturnsNull) { + public ClientWiredChronicleMapStatelessBuilder removeReturnsNull(boolean removeReturnsNull) { this.removeReturnsNull = removeReturnsNull; return this; } @@ -99,7 +99,7 @@ import java.util.concurrent.atomic.AtomicBoolean; } @Override - public ClientWiredChronicleMapStatelessClientBuilder timeout(long timeout, TimeUnit units) { + public ClientWiredChronicleMapStatelessBuilder timeout(long timeout, TimeUnit units) { this.timeoutMs = units.toMillis(timeout); return this; } @@ -109,7 +109,7 @@ import java.util.concurrent.atomic.AtomicBoolean; } @Override - public ClientWiredChronicleMapStatelessClientBuilder name(String name) { + public ClientWiredChronicleMapStatelessBuilder name(String name) { this.name = name; return this; } @@ -119,7 +119,7 @@ import java.util.concurrent.atomic.AtomicBoolean; } @Override - public ClientWiredChronicleMapStatelessClientBuilder tcpBufferSize(int tcpBufferSize) { + public ClientWiredChronicleMapStatelessBuilder tcpBufferSize(int tcpBufferSize) { this.tcpBufferSize = tcpBufferSize; return this; } @@ -133,7 +133,7 @@ import java.util.concurrent.atomic.AtomicBoolean; // todo clean this up if (hub == null) - hub = new ClientWiredStatelessClientTcpConnectionHub(this, localIdentifier, doHandShaking); + hub = new ClientWiredStatelessTcpConnectionHub(this, localIdentifier, doHandShaking); if (!used.getAndSet(true)) { return new ClientWiredStatelessChronicleMap(this, keyClass, valueClass, channelID); @@ -155,7 +155,7 @@ import java.util.concurrent.atomic.AtomicBoolean; } - public ClientWiredStatelessClientTcpConnectionHub hub() { + public ClientWiredStatelessTcpConnectionHub hub() { return hub; } diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index 643b0e30d..f33279606 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -44,7 +44,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon private static final Logger LOG = LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class); - private final ClientWiredStatelessClientTcpConnectionHub hub; + private final ClientWiredStatelessTcpConnectionHub hub; protected Class kClass; protected Class vClass; @@ -53,7 +53,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon private boolean removeReturnsNull; private short channelID; - public ClientWiredStatelessChronicleMap(@NotNull final ClientWiredChronicleMapStatelessClientBuilder config, + public ClientWiredStatelessChronicleMap(@NotNull final ClientWiredChronicleMapStatelessBuilder config, @NotNull final Class kClass, @NotNull final Class vClass, short channelID) { @@ -199,7 +199,38 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon @NotNull public String toString() { - return hub.proxyReturnString(TO_STRING.toString(), channelID); + final Map result = toMap(); + return result.toString(); + } + + private Map toMap() { + final long startTime = System.currentTimeMillis(); + final Map result = new HashMap(); + + // send + final long transactionId = proxySend("ENTRY_SET", startTime); + assert !hub.outBytesLock().isHeldByCurrentThread(); + final long timeoutTime = startTime + hub.timeoutMs; + + for (; ; ) { + + // receive + hub.inBytesLock().lock(); + try { + final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + + if (wireIn.read(Fields.HAS_NEXT).bool()) { + result.put( + readK(Fields.RESULT_KEY, wireIn, null), + readV(Fields.RESULT_VALUE, wireIn, null)); + } else + break; + + } finally { + hub.inBytesLock().unlock(); + } + } + return result; } @NotNull diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessClientTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java similarity index 98% rename from src/main/java/net/openhft/chronicle/map/ClientWiredStatelessClientTcpConnectionHub.java rename to src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index f6875e91e..ca042457e 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessClientTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java @@ -45,7 +45,7 @@ import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; /** * Created by Rob Austin */ -public class ClientWiredStatelessClientTcpConnectionHub { +public class ClientWiredStatelessTcpConnectionHub { private static final Logger LOG = LoggerFactory.getLogger(StatelessChronicleMap.class); @@ -59,7 +59,7 @@ public class ClientWiredStatelessClientTcpConnectionHub { @NotNull private final AtomicLong transactionID = new AtomicLong(0); - private final ClientWiredChronicleMapStatelessClientBuilder config; + private final ClientWiredChronicleMapStatelessBuilder config; @Nullable protected CloseablesManager closeables; @@ -84,7 +84,7 @@ public class ClientWiredStatelessClientTcpConnectionHub { private boolean doHandShaking; - public ClientWiredStatelessClientTcpConnectionHub(ClientWiredChronicleMapStatelessClientBuilder config, byte localIdentifier, boolean doHandShaking) { + public ClientWiredStatelessTcpConnectionHub(ClientWiredChronicleMapStatelessBuilder config, byte localIdentifier, boolean doHandShaking) { this.localIdentifier = localIdentifier; this.doHandShaking = doHandShaking; this.remoteAddress = config.remoteAddress(); diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index cfe97d5a9..ee34e817a 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -125,7 +125,7 @@ class MapWireHandler implements WireHandler { } private void writeChunked(long transactionId, - @NotNull final Function> function, + @NotNull final Function function, @NotNull final Consumer> c) { final BytesChronicleMap m = bytesMap(channelId); @@ -155,6 +155,37 @@ class MapWireHandler implements WireHandler { } + private void writeEntryChunked(long transactionId, + @NotNull final Function function, + @NotNull final Consumer>> c) { + + final BytesChronicleMap m = bytesMap(channelId); + final Iterator> iterator = function.apply(m); + + // this allows us to write more data than the buffer will allow + out = () -> { + + // each chunk has its own transaction-id + outWire.write(TRANSACTION_ID).int64(transactionId); + + write(map -> { + + boolean hasNext = iterator.hasNext(); + outWire.write(HAS_NEXT).bool(hasNext); + + if (hasNext) + c.accept(iterator); + else + // setting out to NULL denotes that there are no more chunks + out = null; + }); + + }; + + out.run(); + + } + @SuppressWarnings("UnusedReturnValue") void onEvent() { @@ -182,12 +213,12 @@ class MapWireHandler implements WireHandler { } if ("VALUES".contentEquals(methodName)) { - writeChunked(transactionId, map -> map.values().iterator(), writeElement); + writeChunked(transactionId, map -> map.delegate.values().iterator(), writeElement); return; } if ("ENTRY_SET".contentEquals(methodName)) { - writeChunked(transactionId, map -> map.entrySet().iterator(), writeEntry); + writeEntryChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry); return; } @@ -303,10 +334,6 @@ class MapWireHandler implements WireHandler { return; } - if ("TO_STRING".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).text(b.toString())); - return; - } if ("APPLICATION_VERSION".contentEquals(methodName)) { write(b -> outWire.write(RESULT).text(applicationVersion()));