diff --git a/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java b/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java index 4acc192f3..9e0cbd889 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java @@ -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 diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ConnectionListener.java b/src/main/java/net/openhft/chronicle/hash/replication/ConnectionListener.java new file mode 100644 index 000000000..2f15335d9 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/replication/ConnectionListener.java @@ -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); + + +} \ No newline at end of file diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index 019b7c8a0..33296bd79 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -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); } diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index fcccf1e6d..6ea8e6207 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/src/main/java/net/openhft/chronicle/map/Replicators.java @@ -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()); } }; } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index c21d22591..a044ed867 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -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 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 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 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 extends AbstractChannelReplicator impleme try { heartbeatCheckHasReceived(key, approxTime); } catch (Exception e) { + if (LOG.isDebugEnabled()) LOG.debug("", e); } @@ -413,6 +417,12 @@ public final class TcpReplicator 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 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 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;