added connection listener

src/main/java/net/openhft/chronicle/hash/replication/ConnectionListener.java

@@ -16,7 +16,7 @@
 
 package net.openhft.chronicle.hash.replication;
 
-import java.net.SocketAddress;
+import java.net.InetAddress;
 
 public interface ConnectionListener {
 
@@ -27,15 +27,14 @@ public interface ConnectionListener {
      *                   we made the connection to a remote host then {@code isServer} is {@code
      *                   false}
      */
-    void onConnect(SocketAddress address, byte identifier, boolean isServer);
+    void onConnect(InetAddress 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);
+    void onDisconnect(InetAddress address, byte identifier);
 
 
 }

src/main/java/net/openhft/chronicle/map/TcpReplicator.java

@@ -391,9 +391,10 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
         if (LOG.isDebugEnabled())
             LOG.debug("", e);
 
-        // todo OZAN :
         //  null check (Attached)key.attachment();
-        //      connectionListener.onDiconnect( ((SocketChannel)key.channel()).socket().getInetAddress(),         ((Attached)key.attachment()).remoteIdentifier);
+        if (key.channel() != null && key.attachment() != null)
+            connectionListener.onDisconnect(((SocketChannel) key.channel()).socket().getInetAddress(),
+                    ((Attached) key.attachment()).remoteIdentifier);
 
 
         closeEarlyAndQuietly(key.channel());
@@ -505,7 +506,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
      */
     boolean isValidVersionNumber(String versionNumber) {
 
-        if (versionNumber.length()<=2)
+        if (versionNumber.length() <= 2)
             return false;
 
         for (char c : versionNumber.toCharArray()) {
@@ -583,10 +584,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 );
+            final SocketChannel channel = (SocketChannel) key.channel();
+            if (channel != null && channel.socket() != null) {
+                connectionListener.onConnect(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.
@@ -1148,8 +1150,8 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
         /**
          * writes all the entries that have changed, to the buffer which will later be written to
          * TCP/IP
-         *  @param modificationIterator a record of which entries have modification
          *
+         * @param modificationIterator a record of which entries have modification
          */
         void entriesToBuffer(@NotNull final Replica.ModificationIterator modificationIterator) throws InterruptedException {
 
@@ -1818,7 +1820,7 @@ class StatelessServerConnector<K, V> {
 
         writeException(writer, e);
 
-        writeSizeAndFlags(sizeLocation , true, writer.in());
+        writeSizeAndFlags(sizeLocation, true, writer.in());
         return null;
     }
 

src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java

@@ -566,6 +566,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
         long hash = metaKeyInterop.hash(keyInterop, key);
         int segmentNum = getSegment(hash);
         long segmentHash = segmentHash(hash);
+
         return segments[segmentNum].update(copies, null,
                 metaKeyInterop, keyInterop, key, keySize, keyIdentity(),
                 this, value, valueIdentity(), segmentHash);
@@ -604,6 +605,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
         long hash = metaKeyInterop.hash(keyInterop, key);
         int segmentNum = getSegment(hash);
         long segmentHash = segmentHash(hash);
+
+        System.out.println("\n" + "segmentNum=" + segmentNum + ",hash=" + hash);
+
         return segments[segmentNum].put3(copies, segmentState,
                 metaKeyInterop, keyInterop, key, keySize, toKey,
                 getValueInterops, value, toValue,
@@ -774,6 +778,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
                 metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key);
         long keySize = metaKeyInterop.size(keyInterop, key);
         long hash = metaKeyInterop.hash(keyInterop, key);
+
+
         int segmentNum = getSegment(hash);
         long segmentHash = segmentHash(hash);