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 extends VanillaChronicleMap extends VanillaChronicleMap> 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 extends MapEntryStages implements RemoteOperationContext 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); }