Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

commits some fixes but still and issue with "wire entry set"

Browse Source
This commit is contained in:
Rob Austin
2015-03-12 21:26:32 +00:00
parent 9cc5b854ae
commit baf9013904
4 changed files with 85 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessClientBuilder.java → src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java
+14 -14
View File
@@ -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<K, V> implements
ChronicleHashStatelessClientBuilder<ClientWiredChronicleMapStatelessClientBuilder<K, V>,
public final class ClientWiredChronicleMapStatelessBuilder<K, V> implements
ChronicleHashStatelessClientBuilder<ClientWiredChronicleMapStatelessBuilder<K, V>,
ChronicleMap<K, V>>,
MapBuilder<ClientWiredChronicleMapStatelessClientBuilder<K, V>> {
MapBuilder<ClientWiredChronicleMapStatelessBuilder<K, V>> {
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<K, V> putReturnsNull(boolean putReturnsNull) {
public ClientWiredChronicleMapStatelessBuilder<K, V> putReturnsNull(boolean putReturnsNull) {
this.putReturnsNull = putReturnsNull;
return this;
}
@@ -89,7 +89,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
}
@Override
public ClientWiredChronicleMapStatelessClientBuilder<K, V> removeReturnsNull(boolean removeReturnsNull) {
public ClientWiredChronicleMapStatelessBuilder<K, V> removeReturnsNull(boolean removeReturnsNull) {
this.removeReturnsNull = removeReturnsNull;
return this;
}
@@ -99,7 +99,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
}
@Override
public ClientWiredChronicleMapStatelessClientBuilder<K, V> timeout(long timeout, TimeUnit units) {
public ClientWiredChronicleMapStatelessBuilder<K, V> 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<K, V> name(String name) {
public ClientWiredChronicleMapStatelessBuilder<K, V> name(String name) {
this.name = name;
return this;
}
@@ -119,7 +119,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
}
@Override
public ClientWiredChronicleMapStatelessClientBuilder<K, V> tcpBufferSize(int tcpBufferSize) {
public ClientWiredChronicleMapStatelessBuilder<K, V> 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<K, V>(this, keyClass, valueClass, channelID);
@@ -155,7 +155,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
}
public ClientWiredStatelessClientTcpConnectionHub hub() {
public ClientWiredStatelessTcpConnectionHub hub() {
return hub;
}
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+34 -3
View File
@@ -44,7 +44,7 @@ class ClientWiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clon
private static final Logger LOG = LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class);
private final ClientWiredStatelessClientTcpConnectionHub hub;
private final ClientWiredStatelessTcpConnectionHub hub;
protected Class<K> kClass;
protected Class<V> vClass;
@@ -53,7 +53,7 @@ class ClientWiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, Clon
@NotNull
public String toString() {
return hub.proxyReturnString(TO_STRING.toString(), channelID);
final Map<K, V> result = toMap();
return result.toString();
}
private Map<K, V> toMap() {
final long startTime = System.currentTimeMillis();
final Map<K, V> result = new HashMap<K, V>();
// 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
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessClientTcpConnectionHub.java → src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java
+3 -3
View File
@@ -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();
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+34 -7
View File
@@ -125,7 +125,7 @@ class MapWireHandler<K, V> implements WireHandler {
}
private void writeChunked(long transactionId,
@NotNull final Function<Map, Iterator<byte[]>> function,
@NotNull final Function<BytesChronicleMap, Iterator> function,
@NotNull final Consumer<Iterator<byte[]>> c) {
final BytesChronicleMap m = bytesMap(channelId);
@@ -155,6 +155,37 @@ class MapWireHandler<K, V> implements WireHandler {
}
private void writeEntryChunked(long transactionId,
@NotNull final Function<BytesChronicleMap, Iterator> function,
@NotNull final Consumer<Iterator<Map.Entry<byte[], byte[]>>> c) {
final BytesChronicleMap m = bytesMap(channelId);
final Iterator<Map.Entry<byte[], byte[]>> 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<K, V> 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<K, V> 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()));