From e20ea011e3098b1302a9f492ce70ec14c098ef37 Mon Sep 17 00:00:00 2001
From: leventov <leventov@ya.ru>
Date: Fri, 1 Jan 2016 21:21:34 +0300
Subject: [PATCH] Add ReplicableEntry.dropChangeFor(remoteIdentifier)

---
 .../replication/ReplicableEntryDelegating.java |  5 +++++
 .../hash/replication/ReplicableEntry.java      |  8 +++++++-
 .../chronicle/map/ReplicatedChronicleMap.java  | 18 ++++++++++++++++++
 .../stage/entry/ReplicatedMapEntryStages.java  |  7 +++++++
 .../stage/replication/ReplicationUpdate.java   |  4 ++++
 5 files changed, 41 insertions(+), 1 deletion(-)

diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java
index 8d791bddc..10c3d9bef 100644
--- a/src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java
@@ -42,6 +42,11 @@ public interface ReplicableEntryDelegating extends ReplicableEntry {
         d().dropChanged();
     }
 
+    @Override
+    default void dropChangedFor(byte remoteIdentifier) {
+        d().dropChangedFor(remoteIdentifier);
+    }
+
     @Override
     default void raiseChanged() {
         d().raiseChanged();
diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java
index 26ff8d685..a300593d2 100644
--- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java
+++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java
@@ -76,13 +76,19 @@ public interface ReplicableEntry {
      */
     void dropChanged();
 
+    /**
+     * Suppress the entry, if it was scheduled to be replicated over to the remote Chronicle node
+     * with the specified identifier.
+     */
+    void dropChangedFor(byte remoteIdentifier);
+
     /**
      * Propagate the entry, schedule it to be replicated over to all remote Chronicle nodes.
      */
     void raiseChanged();
 
     /**
-     * Propagate the entry, schedule it to be replicated over to the remote Chronicle nodes with
+     * Propagate the entry, schedule it to be replicated over to the remote Chronicle node with
      * the specified identifier.
      *
      * @param remoteIdentifier the identifier of the node to replicate this entry to
diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
index ab3320f0a..0be4850e0 100644
--- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
@@ -380,6 +380,10 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
         }
     }
 
+    public void dropChangeFor(long tierIndex, long pos, byte remoteIdentifier) {
+        acquireModificationIterator(remoteIdentifier).dropChange0(tierIndex, pos);
+    }
+
     public void moveChange(long oldTierIndex, long oldPos, long newTierIndex, long newPos) {
         if (oldTierIndex <= actualSegments) {
             long oldSegmentIndex = oldTierIndex - 1;
@@ -899,5 +903,19 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
                 raiseChangeInTierBulk(bulkIndex, offsetToTierBitSet, pos);
             }
         }
+
+        void dropChange0(long tierIndex, long pos) {
+            if (tierIndex <= actualSegments) {
+                long segmentIndex = tierIndex - 1;
+                long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize;
+                dropChangeInSegment(offsetToTierBitSet, pos);
+            } else {
+                long extraTierIndex = tierIndex - 1 - actualSegments;
+                int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk);
+                long offsetToTierBitSet =
+                        (extraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize;
+                dropChangeInTierBulk(bulkIndex, offsetToTierBitSet, pos);
+            }
+        }
     }
 }
\ No newline at end of file
diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java
index a62950705..82ab0b12a 100644
--- a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java
+++ b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java
@@ -112,6 +112,13 @@ public abstract class ReplicatedMapEntryStages<K, V> extends MapEntryStages<K, V
         ru.dropChange();
     }
 
+    @Override
+    public void dropChangedFor(byte remoteIdentifier) {
+        checkOnEachPublicOperation.checkOnEachPublicOperation();
+        s.innerUpdateLock.lock();
+        ru.dropChangeFor(remoteIdentifier);
+    }
+
     @Override
     public void raiseChanged() {
         checkOnEachPublicOperation.checkOnEachPublicOperation();
diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java
index e3434ed8d..dc3db2649 100644
--- a/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java
+++ b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java
@@ -52,6 +52,10 @@ public abstract class ReplicationUpdate<K> implements RemoteOperationContext<K>
         mh.m().dropChange(s.tierIndex, e.pos);
     }
 
+    public void dropChangeFor(byte remoteIdentifier) {
+        mh.m().dropChangeFor(s.tierIndex, e.pos, remoteIdentifier);
+    }
+
     public void moveChange(long oldTierIndex, long oldPos, long newPos) {
         mh.m().moveChange(oldTierIndex, oldPos, s.tierIndex, newPos);
     }