diff --git a/src/main/java/net/openhft/chronicle/map/EngineReplicationLangBytes.java b/src/main/java/net/openhft/chronicle/map/EngineReplicationLangBytes.java
new file mode 100644
index 000000000..c2c4723b7
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/EngineReplicationLangBytes.java
@@ -0,0 +1,184 @@
+/*
+ * Copyright 2014 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.map;
+
+import net.openhft.lang.io.Bytes;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.Closeable;
+
+/**
+ * A simpler form of the replication interface used by Engine
+ *
+ * *** PLEASE DON'T REFACTOR THIS WITHOUT FIRST CHANGING THE ENGINE CODE ***
+ *
+ * @author Rob Austin.
+ */
+public interface EngineReplicationLangBytes extends Closeable {
+
+ /**
+ * inserts or updates the entry
+ *
+ * @param key the key of the entry
+ * @param value the value of the entry
+ * @param remoteIdentifier the identifier of the remote server
+ * @param timestamp the timestamp send from the remote server, this time stamp was the
+ * time the entry was removed
+ */
+ void put(Bytes key, Bytes value, byte remoteIdentifier, long timestamp);
+
+ /**
+ * marks the entry as removed
+ *
+ * @param key the key of the entry
+ * @param remoteIdentifier the identifier of the remote server
+ * @param timestamp the timestamp send from the remote server, this time stamp was the
+ * time the entry was removed
+ */
+ void remove(Bytes key, byte remoteIdentifier, long timestamp);
+
+ /**
+ * Provides the unique Identifier associated with this instance. An identifier is used to
+ * determine which replicating node made the change.
If two nodes update their map at the
+ * same time with different values, we have to deterministically resolve which update wins,
+ * because of eventual consistency both nodes should end up locally holding the same data.
+ * Although it is rare two remote nodes could receive an update to their maps at exactly the
+ * same time for the same key, we have to handle this edge case, its therefore important not to
+ * rely on timestamps alone to reconcile the updates. Typically the update with the newest
+ * timestamp should win, but in this example both timestamps are the same, and the decision
+ * made to one node should be identical to the decision made to the other. We resolve this
+ * simple dilemma by using a node identifier, each node will have a unique identifier, the
+ * update from the node with the smallest identifier wins.
+ *
+ * @return the unique Identifier associated with this map instance
+ */
+ byte identifier();
+
+ /**
+ * Gets (if it does not exist, creates) an instance of ModificationIterator associated with a
+ * remote node, this weak associated is bound using the {@code identifier}.
+ *
+ * @param remoteIdentifier the identifier of the remote node
+ * @return the ModificationIterator dedicated for replication to the remote node with the given
+ * identifier
+ * @see #identifier()
+ */
+ EngineModificationIterator acquireEngineModificationIterator(byte remoteIdentifier);
+
+ /**
+ * Returns the timestamp of the last change from the specified remote node, already replicated
+ * to this host.
Used in conjunction with replication, to back fill data from a remote node.
+ * This node may have missed updates while it was not been running or connected via TCP.
+ *
+ * @param remoteIdentifier the identifier of the remote node to check last replicated update
+ * time from
+ * @return a timestamp of the last modification to an entry, or 0 if there are no entries.
+ * @see #identifier()
+ */
+ long lastModificationTime(byte remoteIdentifier);
+
+ void setLastModificationTime(byte identifier, long timestamp);
+
+ /**
+ * notifies when there is a changed to the modification iterator
+ */
+ interface EngineReplicationModificationNotifier {
+ EngineReplicationModificationNotifier NOP = new EngineReplicationModificationNotifier() {
+ @Override
+ public void onChange() {
+ }
+ };
+
+ /**
+ * called when ever there is a change applied to the modification iterator, in otherwords
+ * when {@link EngineModificationIterator#hasNext()} will return true
+ */
+ void onChange();
+ }
+
+ /**
+ * Holds a record of which entries have modification. Each remote map supported will require a
+ * corresponding ModificationIterator instance
+ */
+ interface EngineModificationIterator {
+
+ /**
+ * @return {@code true} if the is another entry to be received via {@link
+ * EngineModificationIterator#nextEntry(EngineReplicationLangBytes.EngineEntryCallback)}
+ */
+ boolean hasNext();
+
+ /**
+ * A non-blocking call that provides the entry that has changed to {@code
+ * callback.onEntry()}.
+ *
+ * @param callback a callback which will be called when a new entry becomes available.
+ * @return {@code true} if the entry was accepted by the {@code callback.onEntry()} method,
+ * {@code false} if the entry was not accepted or was not available
+ */
+ boolean nextEntry(@NotNull final EngineEntryCallback callback) throws InterruptedException;
+
+ /**
+ * Dirties all entries with a modification time equal to {@code fromTimeStamp} or newer. It
+ * means all these entries will be considered as "new" by this ModificationIterator and
+ * iterated once again no matter if they have already been.
This functionality is used
+ * to publish recently modified entries to a new remote node as it connects.
+ *
+ * @param fromTimeStamp the timestamp from which all entries should be dirty
+ */
+ void dirtyEntries(long fromTimeStamp) throws InterruptedException;
+
+ /**
+ * the {@code modificationNotifier} is called when ever there is a change applied to the
+ * modification iterator
+ *
+ * @param modificationNotifier gets notified when a change occurs
+ */
+ void setModificationNotifier(@NotNull final EngineReplicationModificationNotifier modificationNotifier);
+ }
+
+ /**
+ * Implemented typically by a replicator, This interface provides the event, which will get
+ * called whenever a put() or remove() has occurred to the map
+ */
+ interface EngineEntryCallback {
+
+ /**
+ * Called whenever a put() or remove() has occurred to a replicating map.
+ *
+ * @param key the key on the entry
+ * @param value the value of the entry, will be null if the entry {@code
+ * isDeleted == true}
+ * @param isDeleted {@code true} if remove is called otherwise {@code false}
+ * @param bootStrapTimeStamp sent to the client on every update this is the timestamp that
+ * the remote client should bootstrap from when there has been a
+ * disconnection, this time maybe later than the message time as
+ * event are not send in chronological order from the bit set.
+ * @return {@code false} if this entry should be ignored because the identifier of the
+ * source node is not from one of our changes, WARNING even though we check the identifier
+ * in the ModificationIterator the entry may have been updated.
+ */
+ boolean onEntry(@NotNull Bytes key,
+ @Nullable Bytes value, long timestamp,
+ byte identifier,
+ boolean isDeleted,
+ long bootStrapTimeStamp);
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
index 1fbb66ec4..3d3e754ec 100644
--- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
@@ -28,10 +28,7 @@ import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter;
import net.openhft.lang.Maths;
import net.openhft.lang.collection.ATSDirectBitSet;
-import net.openhft.lang.io.Bytes;
-import net.openhft.lang.io.CheckedBytes;
-import net.openhft.lang.io.MultiStoreBytes;
-import net.openhft.lang.io.NativeBytes;
+import net.openhft.lang.io.*;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -40,6 +37,7 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
@@ -50,6 +48,7 @@ import java.util.concurrent.atomic.AtomicReferenceArray;
import static net.openhft.chronicle.hash.hashing.Hasher.hash;
import static net.openhft.lang.MemoryUnit.*;
import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
+import static net.openhft.lang.io.NativeBytes.wrap;
/**
*
A Replicating Multi Master HashMap
Each remote hash map, mirrors its changes over to
@@ -92,7 +91,7 @@ import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
final class ReplicatedChronicleMap,
V, VI, MVI extends MetaBytesInterop>
extends VanillaChronicleMap
- implements Replica, Replica.EntryExternalizable, Replica.EntryResolver {
+ implements Replica, Replica.EntryExternalizable, Replica.EntryResolver, EngineReplicationLangBytes {
// for file, jdbc and UDP replication
public static final int RESERVED_MOD_ITER = 8;
public static final int ADDITIONAL_ENTRY_BYTES = 10;
@@ -348,11 +347,190 @@ final class ReplicatedChronicleMap buffer = new ThreadLocal() {
+ @Override
+ protected IByteBufferBytes initialValue() {
+ return ByteBufferBytes.wrap(ByteBuffer.allocate(1024));
+ }
+ };
+
+ private Bytes threadlocalBuffer(long requiredSize) {
+
+ final IByteBufferBytes bytes = buffer.get();
+ bytes.clear();
+ bytes.buffer().clear();
+
+ if (bytes.buffer().remaining() < requiredSize) {
+
+ if (requiredSize > Integer.MAX_VALUE)
+ throw new IllegalStateException("entry is too large");
+
+ IByteBufferBytes result = ByteBufferBytes.wrap(ByteBuffer.allocate((int) requiredSize));
+ buffer.set(result);
+ return result;
+ }
+
+ return bytes;
+
+ }
+
+ private static class EntryBuffer {
+ NativeBytes key;
+ NativeBytes value;
+
+ NativeBytes getKey() {
+ return key;
+ }
+
+ void setKey(final NativeBytes key) {
+ this.key = key;
+ }
+
+ NativeBytes getValue() {
+ return value;
+ }
+
+ void setValue(final NativeBytes value) {
+ this.value = value;
+ }
+ }
+
+ @Override
+ public void put(final Bytes key, final Bytes value, final byte id, final long timestamp) {
+
+ final ThreadLocalCopies copies = SegmentState.getCopies(null);
+ final SegmentState segmentState = SegmentState.get(copies);
+
+ long keySize = key.remaining();
+ long valueSize = key.remaining();
+
+ final long requiredBufferSize = 8 + keySize + 8 + valueSize + 12;
+ final Bytes source = threadlocalBuffer(requiredBufferSize);
+
+ keySizeMarshaller.writeSize(source, keySize);
+ valueSizeMarshaller.writeSize(source, keySize);
+ source.writeLong(timestamp);
+ source.writeByte(id);
+ source.writeBoolean(false);
+
+ // key.bytes().
+ writeTo(source, key);
+ writeTo(source, value);
+
+ readExternalEntry(copies, segmentState, source);
+ }
+
+ private static void writeTo(Bytes destination, Bytes source) {
+
+ while (destination.remaining() > 0 && source.remaining() > 0) {
+ source.writeByte(source.readByte());
+ }
+
+ }
+
+ @Override
+ public void remove(final Bytes key,
+ final byte remoteIdentifier,
+ final long timestamp) {
+
+ final ThreadLocalCopies copies = SegmentState.getCopies(null);
+
+ final SegmentState segmentState = SegmentState.get(copies);
+
+ long keySize = key.remaining();
+ long valueSize = key.remaining();
+
+ final long requiredBufferSize = 8 + keySize + 8 + valueSize + 12;
+ final Bytes source = threadlocalBuffer(requiredBufferSize);
+
+ keySizeMarshaller.writeSize(source, keySize);
+ valueSizeMarshaller.writeSize(source, keySize);
+ source.writeLong(timestamp);
+ source.writeByte(remoteIdentifier);
+ source.writeBoolean(true);
+
+ writeTo(source, key);
+
+ readExternalEntry(copies, segmentState, source);
+ }
+
+
+
@Override
public byte identifier() {
return localIdentifier;
}
+ @Override
+ public EngineModificationIterator acquireEngineModificationIterator(final byte remoteIdentifier) {
+ final ModificationIterator modificationIterator = acquireModificationIterator(remoteIdentifier);
+
+ return new EngineModificationIterator() {
+
+ @Override
+ public boolean hasNext() {
+ return modificationIterator.hasNext();
+ }
+
+ @Override
+ public boolean nextEntry(@NotNull final EngineEntryCallback callback) throws InterruptedException {
+
+ return modificationIterator.nextEntry(new EntryCallback() {
+
+ @Override
+ public boolean onEntry(@NotNull final Bytes entry, final int chronicleId, final long bootStrapTimeStamp) {
+
+ final long keySize = keySizeMarshaller.readSize(entry);
+ final long keyPosition = entry.position();
+ entry.skip(keySize);
+
+ final long timestamp = entry.readLong();
+ final byte identifier = entry.readByte();
+ final boolean isDeleted = entry.readBoolean();
+
+ if (isDeleted)
+ return callback.onEntry(wrap(keyPosition, keySize), null, timestamp, identifier,
+ true,
+ bootStrapTimeStamp);
+
+ long valueSize = valueSizeMarshaller.readSize(entry);
+
+ alignment.alignPositionAddr(entry);
+ final long valuePosition = entry.position();
+
+ return callback.onEntry(wrap(keyPosition, keySize), wrap(valuePosition,
+ valueSize),
+ timestamp,
+ identifier,
+ false,
+ bootStrapTimeStamp);
+ }
+
+ @Override
+ public boolean shouldBeIgnored(final Bytes entry, final int chronicleId) {
+ return false; // do nothing
+ }
+ }, 0);
+ }
+
+ @Override
+ public void dirtyEntries(final long fromTimeStamp) throws InterruptedException {
+ modificationIterator.dirtyEntries(fromTimeStamp);
+ }
+
+ @Override
+ public void setModificationNotifier(
+ @NotNull final EngineReplicationModificationNotifier modificationNotifier) {
+ modificationIterator.setModificationNotifier(toModificationNotifier());
+ }
+
+ private ModificationNotifier toModificationNotifier() {
+ return null;
+ }
+ };
+
+ }
+
@Override
public ModificationIterator acquireModificationIterator(byte remoteIdentifier) {
ModificationIterator modificationIterator = modificationIterators.get(remoteIdentifier);
diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
index b3062b4a2..8304cdae7 100755
--- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
@@ -1326,7 +1326,8 @@ class VanillaChronicleMap,
// do nothing
}
- static final class SegmentState implements StatefulCopyable, AutoCloseable {
+ public static final class SegmentState implements StatefulCopyable,
+ AutoCloseable {
private static final Provider segmentStateProvider =
Provider.of(SegmentState.class);
private static final SegmentState originalSegmentState = new SegmentState(0);