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 4ad1d40d4..b1be83962 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java @@ -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 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 54a9b74ef..e02f7c2df 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -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); } diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index c58dd9590..7e5280571 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/src/main/java/net/openhft/chronicle/map/Replicators.java @@ -126,7 +126,9 @@ final class Replicators { return new TcpReplicator(replica, entryExternalizable, tcpConfig, - replication.remoteNodeValidator(), statelessClientParameters); + replication.remoteNodeValidator(), + statelessClientParameters, + 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 9f34415c6..b7eadccbb 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -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 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 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 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 extends AbstractChannelReplicator implements Clo try { heartbeatCheckHasReceived(key, approxTime); } catch (Exception e) { + if (LOG.isDebugEnabled()) LOG.debug("", e); } @@ -407,6 +411,12 @@ final class TcpReplicator 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 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 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;