diff --git a/src/main/java/net/openhft/chronicle/java8/Function.java b/src/main/java/net/openhft/chronicle/java8/Function.java new file mode 100644 index 000000000..9cb4e1ca4 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/java8/Function.java @@ -0,0 +1,27 @@ +package net.openhft.chronicle.java8; + +import java.io.Serializable; + +/** + * Represents a function that accepts one argument and produces a result. + * + *

This is a functional interface whose functional method is {@link + * #apply(Object)}. + * + * @param the type of the input to the function + * @param the type of the result of the function + * @since 1.8 + */ +@FunctionalInterface +public interface Function extends Serializable { + + /** + * Applies this function to the given argument. + * + * @param t the function argument + * @return the function result + */ + R apply(T t); + + +} diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 9c10faa0f..667d0a6ae 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -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 implements ChronicleMap, Closeable { private void writeObject(@NotNull Function function) { + long start = bytes.position(); for (; ; ) { try { - bytes.writeObject(function); return; } catch (IllegalStateException e) { @@ -480,6 +480,8 @@ class StatelessChronicleMap implements ChronicleMap, 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 implements ChronicleMap, Closeable { writeKey(key, local); - final V value = e.getValue(); final Class valueClass = value.getClass(); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index dff0d86e6..b86f16cd1 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -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 { 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 o = (Function) reader.readObject(); try (WriteContext wc = map.acquireUsingLocked(key, value)) { diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index d8e45fb3a..400b55c2e 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -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, 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(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 diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index a5334bc91..fc4b0d2d3 100755 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -1497,7 +1497,7 @@ public class ChronicleMapTest { try (final ChronicleMap map = ChronicleMapBuilder .of(CharSequence.class, StringBuilder.class) .entries(1000) - .defaultValue(new StringBuilder()) + .entrySize(40).create()) { StringBuilder value = new StringBuilder(); diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index db8a40030..884b369fd 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -19,17 +19,16 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.chronicle.java8.Function; import org.junit.Assert; import org.junit.Test; import java.io.IOException; -import java.io.Serializable; import java.net.InetSocketAddress; import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.Set; -import java.util.function.Function; /** @@ -41,7 +40,7 @@ public class StatelessClientTest { public static final int SIZE = 100000; - static class StringBuilderToStringFunction implements Function, Serializable { + static class StringBuilderToStringFunction implements Function { @Override public String apply(StringBuilder stringBuilder) { @@ -50,6 +49,27 @@ public class StatelessClientTest { } + /* @Test + public void testMapForKeyLambda() throws IOException, InterruptedException { + + try (ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, + StringBuilder.class) + .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) { + + serverMap.put(10, new StringBuilder("Hello World")); + + try (ChronicleMap statelessMap = ChronicleMapBuilder.of(Integer + .class, StringBuilder.class) + .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { + + String actual = (String) ((StatelessChronicleMap) statelessMap).mapForKey(10, b -> b.toString()); + + Assert.assertEquals("Hello World", actual); + } + } + }*/ + + @Test public void testMapForKey() throws IOException, InterruptedException {