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 
Conflicts:
	src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java
	src/main/java/net/openhft/chronicle/map/ChannelProvider.java
	src/main/java/net/openhft/chronicle/map/Replicators.java
	src/main/java/net/openhft/chronicle/map/TcpReplicator.java
This commit is contained in:
Rob Austin
2015-07-25 18:28:56 +03:00
committed by leventov
parent 9a7eac0cc9
commit 8a4291750c
5 changed files with 73 additions and 12 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,18 +26,22 @@ 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;
private final String name;
@Nullable
private final transient ConnectionListener connectionListener;
// package-private to forbid subclassing from outside of the package
AbstractReplication(byte localIdentifier, Builder builder) {
this.localIdentifier = localIdentifier;
@@ -46,6 +50,7 @@ public abstract class AbstractReplication implements Serializable {
remoteNodeValidator = builder.remoteNodeValidator;
bootstrapOnlyLocalEntries = builder.bootstrapOnlyLocalEntries;
name = builder.name;
connectionListener = builder.connectionListener;
}
@Override
@@ -85,6 +90,10 @@ public abstract class AbstractReplication implements Serializable {
return name;
}
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
@@ -99,6 +108,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;
private String name = "(unknown)";
@@ -133,6 +144,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
+2 -1
View File
@@ -64,7 +64,8 @@ public final class ChannelProvider implements Closeable {
final TcpReplicator tcpReplicator = new TcpReplicator(
channelProvider.asReplica,
channelProvider.asEntryExternalizable,
tcpConfig, hub.remoteNodeValidator(), null, hub.name());
tcpConfig, hub.remoteNodeValidator(), null, hub.name(),
hub.connectionListener());
channelProvider.add(tcpReplicator);
}
src/main/java/net/openhft/chronicle/map/Replicators.java
+4 -2
View File
@@ -126,8 +126,10 @@ final class Replicators {
return new TcpReplicator(replica, entryExternalizable,
tcpConfig,
replication.remoteNodeValidator(), statelessClientParameters,
replication.name());
replication.remoteNodeValidator(),
statelessClientParameters,
replication.name(),
replication.connectionListener());
}
};
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+18 -3
View File
@@ -20,6 +20,7 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.function.SerializableFunction;
import net.openhft.chronicle.hash.impl.util.BuildVersion;
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;
@@ -85,6 +86,7 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
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
@@ -128,7 +130,8 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
@NotNull final TcpTransportAndNetworkConfig replicationConfig,
@Nullable final RemoteNodeValidator remoteNodeValidator,
@Nullable final StatelessClientParameters statelessClientParameters,
String name)
String name,
@Nullable final ConnectionListener connectionListener)
throws IOException {
super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig());
@@ -147,7 +150,7 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
this.externalizable = externalizable;
this.replicationConfig = replicationConfig;
this.connectionListener = connectionListener;
this.remoteNodeValidator = remoteNodeValidator;
this.name = name;
start();
@@ -328,6 +331,7 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
try {
heartbeatCheckHasReceived(key, approxTime);
} catch (Exception e) {
if (LOG.isDebugEnabled())
LOG.debug("", e);
}
@@ -413,6 +417,12 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
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());
}
@@ -601,6 +611,11 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
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;
@@ -1040,7 +1055,7 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
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;