CE-57 adding support for engine to engine replication

src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java

@@ -16,6 +16,7 @@
 
 package net.openhft.chronicle.hash.replication;
 
+import net.openhft.chronicle.map.EngineReplicationLangBytes;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -40,6 +41,7 @@ public abstract class AbstractReplication implements Serializable {
 
     @Nullable
     private final transient ConnectionListener connectionListener;
+    private final EngineReplicationLangBytesConsumer engineReplicationLangBytesConsumer;
 
     // package-private to forbid subclassing from outside of the package
     AbstractReplication(byte localIdentifier, Builder builder) {
@@ -49,6 +51,7 @@ public abstract class AbstractReplication implements Serializable {
         remoteNodeValidator = builder.remoteNodeValidator;
         bootstrapOnlyLocalEntries = builder.bootstrapOnlyLocalEntries;
         connectionListener = builder.connectionListener;
+        engineReplicationLangBytesConsumer = builder.engineReplicationLangBytesConsumer();
     }
 
     @Override
@@ -69,6 +72,10 @@ public abstract class AbstractReplication implements Serializable {
         return tcpConfig;
     }
 
+    public EngineReplicationLangBytesConsumer engineReplicator() {
+        return engineReplicationLangBytesConsumer;
+    }
+
     @Nullable
     public UdpTransportConfig udpTransport() {
         return udpConfig;
@@ -105,10 +112,25 @@ public abstract class AbstractReplication implements Serializable {
 
         private boolean bootstrapOnlyLocalEntries = false;
 
+        private EngineReplicationLangBytesConsumer engineReplicationLangBytesConsumer;
+
         // package-private to forbid subclassing from outside of the package
         Builder() {
         }
 
+        @NotNull
+        public B engineReplication(EngineReplicationLangBytesConsumer
+                                           engineReplicationLangBytesConsumer) {
+            this.engineReplicationLangBytesConsumer = engineReplicationLangBytesConsumer;
+            return (B) this;
+        }
+
+        EngineReplicationLangBytesConsumer engineReplicationLangBytesConsumer() {
+            return engineReplicationLangBytesConsumer;
+        }
+
+
+
         @NotNull
         public B tcpTransportAndNetwork(@Nullable TcpTransportAndNetworkConfig tcpConfig) {
             this.tcpConfig = tcpConfig;
@@ -190,5 +212,6 @@ public abstract class AbstractReplication implements Serializable {
             return ", udpConfig=" + udpConfig +
                     ", remoteNodeValidator=" + remoteNodeValidator;
         }
+
     }
 }

src/main/java/net/openhft/chronicle/hash/replication/EngineReplicationLangBytesConsumer.java

@@ -0,0 +1,11 @@
+package net.openhft.chronicle.hash.replication;
+
+import net.openhft.chronicle.map.EngineReplicationLangBytes;
+
+/**
+ * Created by Rob Austin
+ */
+public interface EngineReplicationLangBytesConsumer {
+
+    void set(EngineReplicationLangBytes engineReplicationLangBytes);
+}

src/main/java/net/openhft/chronicle/hash/replication/SingleChronicleHashReplication.java

@@ -18,14 +18,14 @@ package net.openhft.chronicle.hash.replication;
 
 import org.jetbrains.annotations.NotNull;
 
-public final class SingleChronicleHashReplication extends AbstractReplication {
+public class SingleChronicleHashReplication extends AbstractReplication {
 
     @NotNull
     public static Builder builder() {
         return new Builder();
     }
 
-    SingleChronicleHashReplication(byte localIdentifier, Builder builder) {
+    public SingleChronicleHashReplication(byte localIdentifier, Builder builder) {
         super(localIdentifier, builder);
     }
 
@@ -34,7 +34,7 @@ public final class SingleChronicleHashReplication extends AbstractReplication {
         return "SingleChronicleHashReplication{" + super.toString() + "}";
     }
 
-    public static final class Builder
+    public static class Builder
             extends AbstractReplication.Builder<SingleChronicleHashReplication, Builder> {
         private Builder() {}
 

src/main/java/net/openhft/chronicle/map/ChannelProvider.java

@@ -18,6 +18,7 @@
 
 package net.openhft.chronicle.map;
 
+import net.openhft.chronicle.hash.replication.EngineReplicationLangBytesConsumer;
 import net.openhft.chronicle.hash.replication.ReplicationHub;
 import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
 import net.openhft.chronicle.hash.replication.UdpTransportConfig;
@@ -58,9 +59,10 @@ final class ChannelProvider implements Closeable {
         if (channelProvider != null)
             return channelProvider;
         channelProvider = new ChannelProvider(hub);
+
+
+
         TcpTransportAndNetworkConfig tcpConfig = hub.tcpTransportAndNetwork();
-
-
         if (tcpConfig != null) {
 
             final TcpReplicator tcpReplicator = new TcpReplicator(
@@ -158,11 +160,26 @@ final class ChannelProvider implements Closeable {
 
     private final byte localIdentifier;
     final Replica asReplica = new Replica() {
+        @Override
+        public void put(final Bytes key, final Bytes value, final byte remoteIdentifier, final long timestamp) {
+            throw new UnsupportedOperationException("todo");
+        }
+
+        @Override
+        public void remove(final Bytes key, final byte remoteIdentifier, final long timestamp) {
+            throw new UnsupportedOperationException("todo");
+        }
+
         @Override
         public byte identifier() {
             return localIdentifier;
         }
 
+        @Override
+        public EngineModificationIterator acquireEngineModificationIterator(final byte remoteIdentifier) {
+            throw new UnsupportedOperationException("todo");
+        }
+
         @Override
         public ModificationIterator acquireModificationIterator(
                 final byte remoteIdentifier) {
@@ -472,11 +489,26 @@ final class ChannelProvider implements Closeable {
     class SystemQueue {
 
         final Replica asReplica = new Replica() {
+            @Override
+            public void put(final Bytes key, final Bytes value, final byte remoteIdentifier, final long timestamp) {
+                throw new UnsupportedOperationException("todo");
+            }
+
+            @Override
+            public void remove(final Bytes key, final byte remoteIdentifier, final long timestamp) {
+                throw new UnsupportedOperationException("todo");
+            }
+
             @Override
             public byte identifier() {
                 throw new UnsupportedOperationException();
             }
 
+            @Override
+            public EngineModificationIterator acquireEngineModificationIterator(final byte remoteIdentifier) {
+                throw new UnsupportedOperationException("todo");
+            }
+
             @Override
             public ModificationIterator acquireModificationIterator(
                     final byte remoteIdentifier) {

src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java

@@ -99,7 +99,7 @@ import static net.openhft.lang.model.DataValueGenerator.firstPrimitiveFieldType;
  * @see ChronicleMap
  * @see ChronicleSetBuilder
  */
-public final class ChronicleMapBuilder<K, V> implements Cloneable,
+public class ChronicleMapBuilder<K, V> implements Cloneable,
         ChronicleHashBuilder<K, ChronicleMap<K, V>, ChronicleMapBuilder<K, V>>,
         MapBuilder<ChronicleMapBuilder<K, V>>, Serializable {
 
@@ -1454,12 +1454,14 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
                 throw new AssertionError("Only one non-null replication should be passed");
             }
             ReplicatedChronicleMap result = (ReplicatedChronicleMap) map;
-            List<Replicator> replicators = new ArrayList<>(2);
+            List<Replicator> replicators = new ArrayList<>(3);
             if (singleHashReplication != null) {
                 if (singleHashReplication.tcpTransportAndNetwork() != null)
                     replicators.add(Replicators.tcp(singleHashReplication));
                 if (singleHashReplication.udpTransport() != null)
                     replicators.add(Replicators.udp(singleHashReplication.udpTransport()));
+                if (singleHashReplication.engineReplicator() != null)
+                    replicators.add(Replicators.engineReplicaton(singleHashReplication));
             } else {
                 ReplicationHub hub = channel.hub();
 

src/main/java/net/openhft/chronicle/map/Replica.java

@@ -27,7 +27,7 @@ import java.io.Closeable;
 /**
  * @author Rob Austin.
  */
-interface Replica extends Closeable {
+interface Replica extends Closeable, EngineReplicationLangBytes {
 
     /**
      * Provides the unique Identifier associated with this map instance. <p> An identifier is used

src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java

@@ -26,6 +26,10 @@ import net.openhft.chronicle.hash.serialization.internal.BytesBytesInterop;
 import net.openhft.chronicle.hash.serialization.internal.DelegatingMetaBytesInterop;
 import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
 import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter;
+import net.openhft.chronicle.map.MultiMap.EntryConsumer;
+import net.openhft.chronicle.map.Replica.EntryExternalizable;
+import net.openhft.chronicle.map.Replica.EntryResolver;
+import net.openhft.chronicle.map.ReplicatedChronicleMap.ModificationIterator.EntryModifiableCallback;
 import net.openhft.lang.Maths;
 import net.openhft.lang.collection.ATSDirectBitSet;
 import net.openhft.lang.io.*;
@@ -91,7 +95,7 @@ import static net.openhft.lang.io.NativeBytes.wrap;
 final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
         V, VI, MVI extends MetaBytesInterop<V, ? super VI>>
         extends VanillaChronicleMap<K, KI, MKI, V, VI, MVI>
-        implements Replica, Replica.EntryExternalizable, Replica.EntryResolver<K, V>, EngineReplicationLangBytes {
+        implements Replica, EntryExternalizable, EntryResolver<K, V>, EngineReplicationLangBytes {
     // for file, jdbc and UDP replication
     public static final int RESERVED_MOD_ITER = 8;
     public static final int ADDITIONAL_ENTRY_BYTES = 10;
@@ -276,7 +280,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
         long hash = metaKeyInterop.hash(keyInterop, key1);
         int segmentNum = getSegment(hash);
         long segmentHash = segmentHash(hash);
-        ReplicatedChronicleMap.Segment segment = segment(segmentNum);
+        Segment segment = segment(segmentNum);
         segment.writeLock();
         try {
             return segment.removeWithoutLock(copies, segmentState, metaKeyInterop, keyInterop, key1,
@@ -350,7 +354,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
     private ThreadLocal<IByteBufferBytes> buffer = new ThreadLocal<IByteBufferBytes>() {
         @Override
         protected IByteBufferBytes initialValue() {
-            return ByteBufferBytes.wrap(ByteBuffer.allocate(1024));
+            return ByteBufferBytes.wrap(ByteBuffer.allocateDirect(1024));
         }
     };
 
@@ -365,7 +369,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
             if (requiredSize > Integer.MAX_VALUE)
                 throw new IllegalStateException("entry is too large");
 
-            IByteBufferBytes result = ByteBufferBytes.wrap(ByteBuffer.allocate((int) requiredSize));
+            IByteBufferBytes result = ByteBufferBytes.wrap(ByteBuffer.allocateDirect((int) requiredSize));
             buffer.set(result);
             return result;
         }
@@ -407,9 +411,10 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
         final long requiredBufferSize = 8 + keySize + 8 + valueSize + 12;
         final Bytes source = threadlocalBuffer(requiredBufferSize);
 
+        source.writeLong(timestamp);
         keySizeMarshaller.writeSize(source, keySize);
         valueSizeMarshaller.writeSize(source, keySize);
-        source.writeLong(timestamp);
+        source.writeStopBit(timestamp);
         source.writeByte(id);
         source.writeBoolean(false);
 
@@ -417,13 +422,14 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
         writeTo(source, key);
         writeTo(source, value);
 
+        source.flip();
         readExternalEntry(copies, segmentState, source);
     }
 
     private static void writeTo(Bytes destination, Bytes source) {
 
         while (destination.remaining() > 0 && source.remaining() > 0) {
-            source.writeByte(source.readByte());
+            destination.writeByte(source.readByte());
         }
 
     }
@@ -498,8 +504,10 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
                         alignment.alignPositionAddr(entry);
                         final long valuePosition = entry.position();
 
-                        return callback.onEntry(wrap(keyPosition, keySize), wrap(valuePosition,
-                                        valueSize),
+                        final NativeBytes k = wrap(entry.address() + keyPosition, keySize);
+                        final NativeBytes v = wrap(entry.address() + valuePosition, valueSize);
+
+                        return callback.onEntry(k, v,
                                 timestamp,
                                 identifier,
                                 false,
@@ -1439,14 +1447,14 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
          *                                  node will be published during a bootstrap
          */
         public void dirtyEntries(final long timeStamp,
-                                 final ModificationIterator.EntryModifiableCallback callback,
+                                 final EntryModifiableCallback callback,
                                  final boolean bootstrapOnlyLocalEntries) throws InterruptedException {
             readLock(null);
             ThreadLocalCopies copies = SegmentState.getCopies(null);
             try (SegmentState segmentState = SegmentState.get(copies)) {
                 final int index = Segment.this.getIndex();
                 final MultiStoreBytes tmpBytes = segmentState.tmpBytes;
-                hashLookup().forEach(new MultiMap.EntryConsumer() {
+                hashLookup().forEach(new EntryConsumer() {
                     @Override
                     public void accept(long hash, long pos) {
                         final Bytes entry = reuse(tmpBytes, offsetFromPos(pos));

src/main/java/net/openhft/chronicle/map/Replicators.java

@@ -19,6 +19,8 @@ package net.openhft.chronicle.map;
 import net.openhft.chronicle.hash.replication.AbstractReplication;
 import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
 import net.openhft.chronicle.hash.replication.UdpTransportConfig;
+import net.openhft.chronicle.map.Replica.EntryExternalizable;
+import net.openhft.chronicle.map.TcpReplicator.StatelessClientParameters;
 import net.openhft.lang.io.ByteBufferBytes;
 import net.openhft.lang.io.Bytes;
 import org.jetbrains.annotations.NotNull;
@@ -41,6 +43,27 @@ final class Replicators {
     private Replicators() {
     }
 
+    static Replicator engineReplicaton(final AbstractReplication replication) {
+        return new Replicator() {
+
+            @Override
+            protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder,
+                                        @NotNull final Replica replica,
+                                        @NotNull final EntryExternalizable entryExternalizable,
+                                        final ChronicleMap chronicleMap) throws IOException {
+
+                replication.engineReplicator().set(replica);
+
+                return new Closeable() {
+
+                    @Override
+                    public void close() throws IOException {
+                        // do nothing
+                    }
+                };
+            }
+        };
+    }
 
     static class OutBuffer implements BufferResizer {
 
@@ -113,13 +136,13 @@ final class Replicators {
             @Override
             protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder,
                                         @NotNull final Replica replica,
-                                        @NotNull final Replica.EntryExternalizable entryExternalizable,
+                                        @NotNull final EntryExternalizable entryExternalizable,
                                         final ChronicleMap chronicleMap) throws IOException {
 
                 TcpTransportAndNetworkConfig tcpConfig = replication.tcpTransportAndNetwork();
 
-                TcpReplicator.StatelessClientParameters statelessClientParameters =
-                        new TcpReplicator.StatelessClientParameters(
+                StatelessClientParameters statelessClientParameters =
+                        new StatelessClientParameters(
                         (VanillaChronicleMap) chronicleMap,
                         builder.keyBuilder,
                         builder.valueBuilder);
@@ -139,7 +162,7 @@ final class Replicators {
             @Override
             protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder,
                                         @NotNull final Replica map,
-                                        @NotNull final Replica.EntryExternalizable entryExternalizable,
+                                        @NotNull final EntryExternalizable entryExternalizable,
                                         final ChronicleMap chronicleMap)
                     throws IOException {
                 return new UdpReplicator(map, entryExternalizable, replicationConfig