Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-163 renamed chronicle custer to chronicle channel

Browse Source
This commit is contained in:
Rob Austin
2014-10-13 18:04:38 +01:00
parent 30a1abef06
commit d01b5d8c59
7 changed files with 41 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+1 -1
View File
@@ -562,7 +562,7 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
}
public ChronicleMapBuilder<K, V> channel(ReplicatingCluster.ChronicleChannel replicator) {
public ChronicleMapBuilder<K, V> channel(ReplicatingChannel.ChronicleChannel replicator) {
this.identifier = replicator.identifier();
this.replicators.clear();
replicators.put(replicator.getClass(), replicator);
src/main/java/net/openhft/chronicle/map/Replica.java
+14 -19
View File
@@ -55,14 +55,11 @@ public interface Replica extends Closeable {
ModificationNotifier modificationNotifier);
/**
* Returns the timestamp of the last change from the specified remote node, already replicated
* to this Replica.
* <p/>
* <p>Used in conjunction with replication, to back fill data from a remote node. This node
* Returns the timestamp of the last change from the specified remote node, already replicated to this
* Replica. <p/> <p>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
* @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()
*/
@@ -102,7 +99,7 @@ public interface Replica extends Closeable {
* 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.
* @param chronicleId only assigned when clustering
* @param chronicleId only assigned when using chronicle channels
* @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
*/
@@ -111,10 +108,8 @@ public interface Replica extends Closeable {
/**
* 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.
* <p/>
* <p>This functionality is used to publish recently modified entries to a new remote node as it
* connects.
* matter if they have already been. <p/> <p>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
*/
@@ -133,7 +128,7 @@ public interface Replica extends Closeable {
* @param entry the byte location of the entry to be stored
* @param destination a buffer the entry will be written to, the segment may reject this operation and
* add zeroBytes, if the identifier in the entry did not match the maps local
* @param chronicleId used in cluster into identify the canonical map or queue
* @param chronicleId is the channel id used to identify the canonical map or queue
*/
void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination, int chronicleId);
@@ -180,20 +175,20 @@ public interface Replica extends Closeable {
}
/**
* Implemented typically by a replicator, This interface provides the event, which will get
* called whenever a put() or remove() has occurred to the map
* Implemented typically by a replicator, This interface provides the event, which will get called
* whenever a put() or remove() has occurred to the map
*/
abstract class EntryCallback {
/**
* Called whenever a put() or remove() has occurred to a replicating map.
*
* @param entry the entry you will receive, this does not have to be locked,
* as locking is already provided from the caller.
* @param entry the entry you will receive, this does not have to be locked, as locking is
* already provided from the caller.
* @param chronicleId only assigned when clustering
* @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.
* @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.
*/
public abstract boolean onEntry(final Bytes entry, final int chronicleId);
src/main/java/net/openhft/chronicle/map/ReplicatingCluster.java → src/main/java/net/openhft/chronicle/map/ReplicatingChannel.java
+6 -6
View File
@@ -45,8 +45,8 @@ import static net.openhft.chronicle.map.Replica.ModificationNotifier.NOP;
/**
* @author Rob Austin.
*/
public final class ReplicatingCluster implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(ReplicatingCluster.class.getName());
public final class ReplicatingChannel implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(ReplicatingChannel.class.getName());
private static final byte BOOTSTRAP_MESSAGE = 'B';
final EntryExternalizable asEntryExternalizable = new EntryExternalizable() {
@@ -152,7 +152,7 @@ public final class ReplicatingCluster implements Closeable {
@Override
public void close() throws IOException {
ReplicatingCluster.this.close();
ReplicatingChannel.this.close();
}
};
private final int maxEntrySize;
@@ -169,7 +169,7 @@ public final class ReplicatingCluster implements Closeable {
private final Set<AbstractChannelReplicator> replicators =
new CopyOnWriteArraySet<AbstractChannelReplicator>();
ReplicatingCluster(ReplicatingClusterBuilder builder) {
ReplicatingChannel(ReplicatingChannelBuilder builder) {
localIdentifier = builder.identifier;
maxEntrySize = builder.maxEntrySize;
chronicleChannels = new Replica[builder.maxNumberOfChronicles];
@@ -206,7 +206,7 @@ public final class ReplicatingCluster implements Closeable {
/**
* Returns a replicator, dedicated to the specified channel. Channel is basically just a number, that
* should correspond on different servers for instances of the same replicated map. Only one replicator
* per channel could be obtained from a single {@code ReplicatingCluster}. The returned replicator could
* per channel could be obtained from a single {@code ReplicatingChannel}. The returned replicator could
* be applied to a map at most once.
*
* @return a replicator, dedicated to the specified channel
@@ -409,7 +409,7 @@ public final class ReplicatingCluster implements Closeable {
protected Closeable applyTo(ChronicleMapBuilder builder,
Replica map, EntryExternalizable entryExternalizable) {
if (builder.entrySize() > maxEntrySize) {
throw new IllegalArgumentException("During ReplicatingClusterBuilder setup, " +
throw new IllegalArgumentException("During ReplicatingChannelBuilder setup, " +
"maxEntrySize=" + maxEntrySize + " was specified, but map with " +
"entrySize=" + builder.entrySize() + " is attempted to apply" +
"to the replicator");
src/main/java/net/openhft/chronicle/map/ReplicatingClusterBuilder.java → src/main/java/net/openhft/chronicle/map/ReplicatingChannelBuilder.java
+7 -7
View File
@@ -21,7 +21,7 @@ import java.io.IOException;
/**
* @author Rob Austin.
*/
public final class ReplicatingClusterBuilder {
public final class ReplicatingChannelBuilder {
final byte identifier;
final int maxEntrySize;
@@ -29,7 +29,7 @@ public final class ReplicatingClusterBuilder {
private UdpReplicationConfig udpReplicationConfig = null;
private TcpReplicationConfig tcpReplicationConfig = null;
ReplicatingClusterBuilder(byte identifier, final int maxEntrySize) {
ReplicatingChannelBuilder(byte identifier, final int maxEntrySize) {
this.identifier = identifier;
this.maxEntrySize = maxEntrySize;
if (identifier <= 0) {
@@ -38,23 +38,23 @@ public final class ReplicatingClusterBuilder {
}
}
public ReplicatingClusterBuilder udpReplication(UdpReplicationConfig replicationConfig) {
public ReplicatingChannelBuilder udpReplication(UdpReplicationConfig replicationConfig) {
this.udpReplicationConfig = replicationConfig;
return this;
}
public ReplicatingClusterBuilder tcpReplication(TcpReplicationConfig replicationConfig) {
public ReplicatingChannelBuilder tcpReplication(TcpReplicationConfig replicationConfig) {
this.tcpReplicationConfig = replicationConfig;
return this;
}
public ReplicatingClusterBuilder maxNumberOfChronicles(int maxNumberOfChronicles) {
public ReplicatingChannelBuilder maxNumberOfChronicles(int maxNumberOfChronicles) {
this.maxNumberOfChronicles = maxNumberOfChronicles;
return this;
}
public ReplicatingCluster create() throws IOException {
final ReplicatingCluster replicatingCluster = new ReplicatingCluster(this);
public ReplicatingChannel create() throws IOException {
final ReplicatingChannel replicatingCluster = new ReplicatingChannel(this);
if (tcpReplicationConfig != null) {
final TcpReplicator tcpReplicator = new TcpReplicator(
replicatingCluster.asReplica,