diff --git a/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java index 7eb7652fb..066e30bd4 100644 --- a/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java @@ -26,7 +26,7 @@ import java.util.Set; public class BytesChronicleMap implements AbstractChronicleMap { private final VanillaChronicleMap delegate; - TcpReplicator.TcpSocketChannelEntryWriter output; + MapIOBuffer output; public BytesChronicleMap(VanillaChronicleMap delegate) { this.delegate = delegate; diff --git a/src/main/java/net/openhft/chronicle/map/CrossBytes.java b/src/main/java/net/openhft/chronicle/map/CrossBytes.java new file mode 100644 index 000000000..a1039e27b --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/CrossBytes.java @@ -0,0 +1,12 @@ +package net.openhft.chronicle.map; + +import net.openhft.lang.io.MultiStoreBytes; + +/** + * Created by Rob Austin + */ +public interface CrossBytes { + void write(MultiStoreBytes entry, long valueOffset, long valueSize); + + void writeBoolean(boolean b); +} diff --git a/src/main/java/net/openhft/chronicle/map/MapIOBuffer.java b/src/main/java/net/openhft/chronicle/map/MapIOBuffer.java new file mode 100644 index 000000000..c6366e661 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/MapIOBuffer.java @@ -0,0 +1,13 @@ +package net.openhft.chronicle.map; + +import net.openhft.lang.io.Bytes; + +/** + * Created by Rob Austin + */ +interface MapIOBuffer { + + void ensureBufferSize(long l); + + Bytes in(); +} diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java index 9b616971f..3af82c88c 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java @@ -27,12 +27,16 @@ import net.openhft.chronicle.bytes.BytesMarshallable; import net.openhft.chronicle.wire.TextWire; import net.openhft.chronicle.wire.Wire; import net.openhft.lang.io.DirectStore; +import net.openhft.lang.io.NativeBytes; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.io.PrintWriter; import java.io.StringWriter; +import java.lang.reflect.Field; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.SocketChannel; @@ -44,6 +48,25 @@ import java.util.List; */ class StatelessWiredConnector { + private static final Logger LOG = LoggerFactory.getLogger(StatelessWiredConnector.class); + + + static Field ADDRESS; + static Field CAPACITY; + + static { + try { + ADDRESS = ByteBuffer.class.getDeclaredField("address"); + ADDRESS.setAccessible(true); + + CAPACITY = ByteBuffer.class.getDeclaredField("capacity"); + CAPACITY.setAccessible(true); + } catch (Exception e) { + LOG.error("", e); + } + } + + private final NativeBytes langBytes = new NativeBytes(0, 0); private boolean handshingComplete; // private final byte identifier; @@ -56,6 +79,34 @@ class StatelessWiredConnector Integer.MAX_VALUE) + if (textWire.bytes().position() > Integer.MAX_VALUE) throw new IllegalStateException("position too large"); // write the size @@ -234,26 +284,20 @@ class StatelessWiredConnector extends AbstractChannelReplicator implements Clo */ boolean isValidVersionNumber(String versionNumber) { - if (versionNumber.length()<=2) + if (versionNumber.length() <= 2) return false; for (char c : versionNumber.toCharArray()) { @@ -668,7 +668,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo if (!isValidVersionNumber(attached.serverVersion)) { LOG.warn("Please check that you don't have a third party system incorrectly connecting to ChronicleMap, " + - "Closing the remote connection as Chronicle can not make sense of the remote version number received from the external connection, version="+attached.serverVersion+", Chronicle is expecting the version number to only contain '.',A-Z,a-z,0-9"); + "Closing the remote connection as Chronicle can not make sense of the remote version number received from the external connection, version=" + attached.serverVersion + ", Chronicle is expecting the version number to only contain '.',A-Z,a-z,0-9"); socketChannel.close(); } @@ -1071,8 +1071,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo /** * @author Rob Austin. */ - class TcpSocketChannelEntryWriter { - + public class TcpSocketChannelEntryWriter implements MapIOBuffer { @NotNull private final EntryCallback entryCallback; @@ -1111,7 +1110,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo in().writeByte(localIdentifier); } - void ensureBufferSize(long size) { + public void ensureBufferSize(long size) { if (in().remaining() < size) { size += entryCallback.in().position(); if (size > Integer.MAX_VALUE) @@ -1137,7 +1136,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo throw e; } - Bytes in() { + public Bytes in() { return entryCallback.in(); } @@ -1377,8 +1376,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo /** * reads entries from the buffer till empty * - * @param attached * @param key the selection key - channel + * @param attached * @param key the selection key + * channel * @param key * @throws InterruptedException */ diff --git a/src/main/java/net/openhft/chronicle/map/VanillaContext.java b/src/main/java/net/openhft/chronicle/map/VanillaContext.java index 1a2f3d118..25460c987 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaContext.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaContext.java @@ -18,7 +18,6 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.impl.ContextFactory; import net.openhft.chronicle.hash.impl.HashContext; -import net.openhft.chronicle.hash.impl.hashlookup.EntryConsumer; import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.chronicle.hash.serialization.internal.BytesBytesInterop; import net.openhft.chronicle.hash.serialization.internal.DelegatingMetaBytesInterop; @@ -33,10 +32,10 @@ class VanillaContext, extends HashContext implements MapKeyContext { - + ///////////////////////////////////////////////// // Inner state & lifecycle - + enum VanillaChronicleMapContextFactory implements ContextFactory { INSTANCE; @@ -55,9 +54,10 @@ class VanillaContext, return VanillaContext.class; } } - - VanillaContext() {} - + + VanillaContext() { + } + VanillaContext(HashContext contextCache, int indexInContextCache) { super(contextCache, indexInContextCache); } @@ -75,11 +75,11 @@ class VanillaContext, ///////////////////////////////////////////////// // Map - + public VanillaChronicleMap m() { return (VanillaChronicleMap) h; } - + @Override public void closeHashDependants() { super.closeHashDependants(); @@ -87,7 +87,7 @@ class VanillaContext, closeValue(); closeValueReader(); } - + ///////////////////////////////////////////////// // Map and context locals @@ -589,7 +589,7 @@ class VanillaContext, private final MultiStoreBytes valueCopy = new MultiStoreBytes(); - TcpReplicator.TcpSocketChannelEntryWriter output; + MapIOBuffer output; void initBytesValueModel0() { valueInterop = (VI) BytesBytesInterop.INSTANCE;