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-331 added the first part of this code, Ozan will complete and issue a pull request

Browse Source
This commit is contained in:
Rob Austin
2015-05-13 11:38:02 +01:00
parent 73a43e0cb3
commit 6fbe5112fa
5 changed files with 71 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java
+24 -6
View File
@@ -26,17 +26,21 @@ import java.io.Serializable;
*/
public abstract class AbstractReplication implements Serializable {
private final byte localIdentifier;
private final
@Nullable
TcpTransportAndNetworkConfig tcpConfig;
private final
private final TcpTransportAndNetworkConfig tcpConfig;
@Nullable
UdpTransportConfig udpConfig;
private final
private final UdpTransportConfig udpConfig;
@Nullable
transient RemoteNodeValidator remoteNodeValidator;
private final transient RemoteNodeValidator remoteNodeValidator;
private final boolean bootstrapOnlyLocalEntries;
@Nullable
private final transient ConnectionListener connectionListener;
// package-private to forbid subclassing from outside of the package
AbstractReplication(byte localIdentifier, Builder builder) {
this.localIdentifier = localIdentifier;
@@ -44,6 +48,7 @@ public abstract class AbstractReplication implements Serializable {
udpConfig = builder.udpConfig;
remoteNodeValidator = builder.remoteNodeValidator;
bootstrapOnlyLocalEntries = builder.bootstrapOnlyLocalEntries;
connectionListener = builder.connectionListener;
}
@Override
@@ -78,6 +83,10 @@ public abstract class AbstractReplication implements Serializable {
return bootstrapOnlyLocalEntries;
}
public ConnectionListener connectionListener() {
return this.connectionListener;
}
/**
* Builder of {@link AbstractReplication} configurations. This class and it's subclasses are
* mutable, configuration methods mutate the builder and return it back for convenient
@@ -92,6 +101,8 @@ public abstract class AbstractReplication implements Serializable {
private TcpTransportAndNetworkConfig tcpConfig = null;
private UdpTransportConfig udpConfig = null;
private RemoteNodeValidator remoteNodeValidator = null;
private ConnectionListener connectionListener = null;
private boolean bootstrapOnlyLocalEntries = false;
// package-private to forbid subclassing from outside of the package
@@ -125,6 +136,13 @@ public abstract class AbstractReplication implements Serializable {
}
@NotNull
public B connectionListener(@Nullable ConnectionListener connectionListener) {
this.connectionListener = connectionListener;
return (B) this;
}
/**
* Configures if the node, provided with replication, created by this builder, should
* replicate only local data, last updated with own identifier, or all data currently
src/main/java/net/openhft/chronicle/hash/replication/ConnectionListener.java
+25
View File
@@ -0,0 +1,25 @@
package net.openhft.chronicle.hash.replication;
import java.net.SocketAddress;
public interface ConnectionListener {
/**
* @param address the address that we are connceted to
* @param identifier the remote identifier that we are not connected to.
* @param isServer if this host accepted the connection {@code isServer} is {@code true}, if
* we made the connection to a remote host then {@code isServer} is {@code
* false}
*/
void onConnect(SocketAddress address, byte identifier, boolean isServer);
/**
* @param address the address that we have been disconnected from
* @param identifier the identifer the address that we have been disconnected from or Byte
* .MIN_VALUE if not known
*/
void onDisconnect(SocketAddress address, byte identifier);
}
src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+1 -1
View File
@@ -67,7 +67,7 @@ final class ChannelProvider implements Closeable {
final TcpReplicator tcpReplicator = new TcpReplicator(
channelProvider.asReplica,
channelProvider.asEntryExternalizable,
tcpConfig, hub.remoteNodeValidator(), null);
tcpConfig, hub.remoteNodeValidator(), null, hub.connectionListener());
channelProvider.add(tcpReplicator);
}
src/main/java/net/openhft/chronicle/map/Replicators.java
+3 -1
View File
@@ -126,7 +126,9 @@ final class Replicators {
return new TcpReplicator(replica, entryExternalizable,
tcpConfig,
replication.remoteNodeValidator(), statelessClientParameters);
replication.remoteNodeValidator(),
statelessClientParameters,
replication.connectionListener());
}
};
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+18 -3
View File
@@ -18,6 +18,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.ConnectionListener;
import net.openhft.chronicle.hash.replication.RemoteNodeValidator;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.ThrottlingConfig;
@@ -80,6 +81,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
new KeyInterestUpdater(OP_WRITE, selectionKeysStore);
private final BitSet activeKeys = new BitSet(selectionKeysStore.length);
private final long heartBeatIntervalMillis;
private final ConnectionListener connectionListener;
private long largestEntrySoFar = 128;
@NotNull
@@ -122,7 +124,8 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
@NotNull final Replica.EntryExternalizable externalizable,
@NotNull final TcpTransportAndNetworkConfig replicationConfig,
@Nullable final RemoteNodeValidator remoteNodeValidator,
@Nullable final StatelessClientParameters statelessClientParameters)
@Nullable final StatelessClientParameters statelessClientParameters,
@Nullable final ConnectionListener connectionListener)
throws IOException {
super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig());
@@ -141,7 +144,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
this.externalizable = externalizable;
this.replicationConfig = replicationConfig;
this.connectionListener = connectionListener;
this.remoteNodeValidator = remoteNodeValidator;
this.name = replicationConfig.name();
start();
@@ -322,6 +325,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
try {
heartbeatCheckHasReceived(key, approxTime);
} catch (Exception e) {
if (LOG.isDebugEnabled())
LOG.debug("", e);
}
@@ -407,6 +411,12 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
private void quietClose(@NotNull final SelectionKey key, @NotNull final Exception e) {
if (LOG.isDebugEnabled())
LOG.debug("", e);
// todo OZAN :
// null check (Attached)key.attachment();
// connectionListener.onDiconnect( ((SocketChannel)key.channel()).socket().getInetAddress(), ((Attached)key.attachment()).remoteIdentifier);
closeEarlyAndQuietly(key.channel());
}
@@ -595,6 +605,11 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
attached.remoteIdentifier = remoteIdentifier;
// todo OZAN : add call to onConnectedListener.
//connectionListener. onConnected(channel.socket().getInetAddress() , attached
// .remoteIdentifier, attached.isServer );
// we use the as iterating the activeKeys via the bitset wont create and Objects
// but if we use the selector.keys() this will.
selectionKeysStore[remoteIdentifier] = key;
@@ -1033,7 +1048,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
public byte remoteIdentifier = Byte.MIN_VALUE;
public boolean hasRemoteHeartbeatInterval;
// true if its socket is a ServerSocket
public boolean isServer; // the frequency the remote node will send a heartbeat
public boolean isServer;
public boolean handShakingComplete;
public String serverVersion;
public long remoteHeartbeatInterval = heartBeatIntervalMillis;