From 50cd15826f381ef72ef3f2e1b8d93f8d089f84c6 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 29 Sep 2014 16:39:53 +0100 Subject: [PATCH] HCOLL-134 introduced the NonUniqueIdentifierListener --- .../map/NonUniqueIdentifierListener.java | 9 +++++ .../chronicle/map/TcpReplicationConfig.java | 9 ++++- .../openhft/chronicle/map/TcpReplicator.java | 36 +++++++++++-------- 3 files changed, 39 insertions(+), 15 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/NonUniqueIdentifierListener.java diff --git a/src/main/java/net/openhft/chronicle/map/NonUniqueIdentifierListener.java b/src/main/java/net/openhft/chronicle/map/NonUniqueIdentifierListener.java new file mode 100644 index 000000000..250a664ee --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/NonUniqueIdentifierListener.java @@ -0,0 +1,9 @@ +package net.openhft.chronicle.map; + +/** + * @author Rob Austin. + */ +public interface NonUniqueIdentifierListener { + + void onNonUniqueIdentifier(byte remoteIdentifier); +} diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java index 7559cec76..60238f102 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java @@ -41,12 +41,19 @@ public class TcpReplicationConfig { private ThrottlingConfig throttlingConfig = ThrottlingConfig.noThrottling(); private long heartBeatInterval = DEFAULT_HEART_BEAT_INTERVAL; private TimeUnit heartBeatIntervalUnit = DEFAULT_HEART_BEAT_INTERVAL_UNIT; - + private NonUniqueIdentifierListener nonUniqueIdentifierListener; TcpReplicationConfig() { } + public NonUniqueIdentifierListener nonUniqueIdentifierListener() { + return nonUniqueIdentifierListener; + } + + public void nonUniqueIdentifierListener(NonUniqueIdentifierListener nonUniqueIdentifierListener) { + this.nonUniqueIdentifierListener = nonUniqueIdentifierListener; + } public boolean autoReconnectedUponDroppedConnection() { return autoReconnectedUponDroppedConnection; diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index df7f1a7dc..226282be1 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -393,11 +393,12 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { /** * used to exchange identifiers and timestamps and heartbeat intervals between the server and client * - * @param key the SelectionKey relating to the this cha + * @param key the SelectionKey relating to the this cha + * @param socketChannel * @throws java.io.IOException * @throws InterruptedException */ - private void doHandShaking(@NotNull final SelectionKey key) + private void doHandShaking(@NotNull final SelectionKey key, SocketChannel socketChannel) throws IOException, InterruptedException { final Attached attached = (Attached) key.attachment(); final TcpSocketChannelEntryWriter writer = attached.entryWriter; @@ -418,15 +419,25 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { activeKeys.set(remoteIdentifier); if (LOG.isDebugEnabled()) { - LOG.debug("server-connection id={}, remoteIdentifier={}", - localIdentifier, remoteIdentifier); + LOG.debug("server-connection id={}, remoteIdentifier={}", localIdentifier, remoteIdentifier); } + // this can occur sometimes when if 2 or more remote node attempt to use node discovery at the + // same time if (remoteIdentifier == localIdentifier) { - throw new IllegalStateException("Where are connecting to a remote " + - "map with the same " + - "identifier as this map, identifier=" + localIdentifier + ", " + - "please change either this maps identifier or the remote one"); + + + final NonUniqueIdentifierListener listener = replicationConfig.nonUniqueIdentifierListener(); + + if (listener != null) + listener.onNonUniqueIdentifier(remoteIdentifier); + + // throwing this exception will cause us to disconnect, both the client and server + // will be able to detect the the remote and local identifiers are the same, + // as the identifier is send early on in the hand shaking via the connect(() and accept() + // methods + throw new IllegalStateException("dropping connection, as the remote and local-identifier " + + "are the same, identifier=" + localIdentifier); } attached.remoteModificationIterator = replica.acquireModificationIterator(remoteIdentifier, @@ -469,7 +480,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { attached.hasRemoteHeartbeatInterval = true; // now we're finished we can get on with reading the entries - attached.setHandShakingComplete(); + attached.handShakingComplete = true; attached.remoteModificationIterator.dirtyEntries(attached.remoteBootstrapTimestamp); reader.entriesFromBuffer(); } @@ -527,7 +538,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { if (attached.isHandShakingComplete()) { attached.entryReader.entriesFromBuffer(); } else { - doHandShaking(key); + doHandShaking(key, socketChannel); } } @@ -718,15 +729,12 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { // true if its socket is a ServerSocket public boolean isServer; // the frequency the remote node will send a heartbeat public long remoteHeartbeatInterval = heartBeatIntervalMillis; - private boolean handShakingComplete; + public boolean handShakingComplete; boolean isHandShakingComplete() { return handShakingComplete; } - void setHandShakingComplete() { - handShakingComplete = true; - } void clearHandShaking() { handShakingComplete = false;