HCOLL-81 StatelessClient : Add a visitor pattern to Chronicle Map ( tested with lamda expressions in java8 )

This commit is contained in:
Rob Austin
2014-11-14 11:58:12 +00:00
parent b7033da88c
commit 0fd6e98bed
6 changed files with 92 additions and 10 deletions
@@ -0,0 +1,27 @@
package net.openhft.chronicle.java8;
import java.io.Serializable;
/**
* Represents a function that accepts one argument and produces a result.
*
* <p>This is a <a href="package-summary.html">functional interface</a> whose functional method is {@link
* #apply(Object)}.
*
* @param <T> the type of the input to the function
* @param <R> the type of the result of the function
* @since 1.8
*/
@FunctionalInterface
public interface Function<T, R> extends Serializable {
/**
* Applies this function to the given argument.
*
* @param t the function argument
* @return the function result
*/
R apply(T t);
}
@@ -20,6 +20,7 @@ package net.openhft.chronicle.map;
import com.sun.jdi.connect.spi.ClosedConnectionException;
import net.openhft.chronicle.hash.RemoteCallTimeoutException;
import net.openhft.chronicle.java8.Function;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
@@ -35,7 +36,6 @@ import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.function.Function;
import static java.nio.ByteBuffer.allocateDirect;
import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE;
@@ -468,9 +468,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private <R> void writeObject(@NotNull Function<V, R> function) {
long start = bytes.position();
for (; ; ) {
try {
bytes.writeObject(function);
return;
} catch (IllegalStateException e) {
@@ -480,6 +480,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
if (cause instanceof IOException && cause.getMessage().contains("Not enough available space")) {
LOG.debug("resizing buffer");
resizeBuffer(buffer.capacity() + 1024);
bytes.limit(bytes.capacity());
bytes.position(start);
} else
throw e;
@@ -557,7 +559,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
writeKey(key, local);
final V value = e.getValue();
final Class<?> valueClass = value.getClass();
@@ -21,6 +21,7 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.RemoteNodeValidator;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.ThrottlingConfig;
import net.openhft.chronicle.java8.Function;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
@@ -38,7 +39,6 @@ import java.nio.channels.*;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import static java.nio.channels.SelectionKey.*;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -1350,10 +1350,15 @@ class StatelessServerConnector<K, V> {
V value = using.getAndSet(null);
if (value == null) {
// a simple way to create a value
value = map.getUsing(key, null);
//value = map.createValueInstance();
// todo replace this
value = map.acquireUsing(key, null);
}
final Function<V, ?> o = (Function<V, ?>) reader.readObject();
try (WriteContext<K, V> wc = map.acquireUsingLocked(key, value)) {
@@ -29,7 +29,10 @@ import net.openhft.lang.collection.DirectBitSet;
import net.openhft.lang.collection.SingleThreadedDirectBitSet;
import net.openhft.lang.io.*;
import net.openhft.lang.io.serialization.JDKObjectSerializer;
import net.openhft.lang.io.serialization.impl.AllocateInstanceObjectFactory;
import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory;
import net.openhft.lang.model.Byteable;
import net.openhft.lang.model.DataValueClasses;
import net.openhft.lang.model.constraints.Nullable;
import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
@@ -349,6 +352,32 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
replaceIfPresent);
}
V createValueInstance() {
if (vClass.isAssignableFrom(Byteable.class))
return DataValueClasses.newDirectReference(vClass);
else {
try {
return vClass.newInstance();
} catch (InstantiationException e) {
try {
return (V) (new AllocateInstanceObjectFactory<V>(vClass)).create();
} catch (Exception e1) {
LOG.error("", e);
return null;
}
} catch (IllegalAccessException e) {
LOG.error("", e);
return null;
}
}
}
enum LockType {READ_LOCK, WRITE_LOCK}
@Override