mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
exposing the boostrap timestamp
This commit is contained in:
@@ -238,6 +238,23 @@ final class ChannelProvider implements Closeable {
|
||||
}
|
||||
notifier0 = modificationNotifier;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @return the earliest bootstrap timestamp across all the channels
|
||||
*/
|
||||
@Override
|
||||
public long bootStrapTimeStamp() {
|
||||
long boostapTimeStamp = Long.MAX_VALUE;
|
||||
|
||||
for (int i = 0, len = chronicleChannelList.size(); i < len; i++) {
|
||||
Replica chronicleChannel = chronicleChannelList.get(i);
|
||||
boostapTimeStamp = Math.min(boostapTimeStamp, chronicleChannel
|
||||
.acquireModificationIterator(remoteIdentifier).bootStrapTimeStamp());
|
||||
}
|
||||
|
||||
return boostapTimeStamp;
|
||||
}
|
||||
};
|
||||
|
||||
modificationIterator.set((int) remoteIdentifier, result0);
|
||||
@@ -528,6 +545,11 @@ final class ChannelProvider implements Closeable {
|
||||
modificationNotifier0 = modificationNotifier;
|
||||
}
|
||||
|
||||
@Override
|
||||
public long bootStrapTimeStamp() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void addPayload(Bytes bytes) {
|
||||
if (bytes.remaining() == 0)
|
||||
|
||||
@@ -125,6 +125,13 @@ interface Replica extends Closeable, EngineReplicationLangBytes {
|
||||
* @param modificationNotifier gets notified when a change occurs
|
||||
*/
|
||||
void setModificationNotifier(@NotNull final ModificationNotifier modificationNotifier);
|
||||
|
||||
/**
|
||||
* @return 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.
|
||||
*/
|
||||
long bootStrapTimeStamp();
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -88,10 +88,9 @@ import static net.openhft.lang.io.NativeBytes.wrap;
|
||||
* dilemma by using a node identifier, each node will have a unique identifier, the update from the
|
||||
* node with the smallest identifier wins. </p>
|
||||
*
|
||||
* @param <K> the entries key type
|
||||
* @param <V> the entries value type
|
||||
* @param <K> the entries key type § * @param <V> the entries value type
|
||||
*/
|
||||
public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
|
||||
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, EntryExternalizable, EntryResolver<K, V>, EngineReplicationLangBytes {
|
||||
@@ -125,13 +124,6 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
}
|
||||
}
|
||||
|
||||
private static void writeTo(Bytes destination, Bytes source) {
|
||||
|
||||
while (destination.remaining() > 0 && source.remaining() > 0) {
|
||||
destination.writeByte(source.readByte());
|
||||
}
|
||||
}
|
||||
|
||||
private int assignedModIterBitSetSizeInBytes() {
|
||||
return (int) CACHE_LINES.align(BYTES.alignAndConvert(127 + RESERVED_MOD_ITER, BITS), BYTES);
|
||||
}
|
||||
@@ -189,10 +181,6 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
return identifierUpdatedBytes.readLong(remoteIdentifier * 8L);
|
||||
}
|
||||
|
||||
public long bootStrapTime(){
|
||||
return this.acquireModificationIterator(this.identifier()).bootStrapTimeStamp();
|
||||
}
|
||||
|
||||
@Override
|
||||
void onHeaderCreated() {
|
||||
long offset = super.getHeaderSize();
|
||||
@@ -1793,7 +1781,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
changes = ATSDirectBitSet.wrap(bytes);
|
||||
}
|
||||
|
||||
public void setModificationNotifier(ModificationNotifier modificationNotifier) {
|
||||
public void setModificationNotifier(@NotNull ModificationNotifier modificationNotifier) {
|
||||
this.modificationNotifier = modificationNotifier;
|
||||
}
|
||||
|
||||
@@ -1934,7 +1922,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
* disconnection, this time maybe later than the message time as event are not send in
|
||||
* chronological order from the bit set.
|
||||
*/
|
||||
private long bootStrapTimeStamp() {
|
||||
public long bootStrapTimeStamp() {
|
||||
final long timeStamp = bootStrapTimeStamp.get();
|
||||
long result = (timeStamp == 0) ? this.lastBootStrapTimeStamp : timeStamp;
|
||||
this.lastBootStrapTimeStamp = result;
|
||||
|
||||
@@ -247,6 +247,10 @@ public class TCPSocketReplicationBootStrapTest {
|
||||
map1 = (ReplicatedChronicleMap<Integer, ?, ?, CharSequence, ?, ?>) ChronicleMapBuilder.of(Integer.class, CharSequence.class)
|
||||
.replication((byte) 1, map1Config).create();
|
||||
|
||||
final long bootStrapTimeStamp = map1.acquireModificationIterator((byte) 1).bootStrapTimeStamp();
|
||||
System.out.println("bootStrapTimeStamp=" + bootStrapTimeStamp);
|
||||
|
||||
|
||||
File persistenceFile = getPersistenceFile();
|
||||
|
||||
TcpTransportAndNetworkConfig map2Config = TcpTransportAndNetworkConfig.of(8067)
|
||||
|
||||
Reference in New Issue
Block a user