HCOLL-169 reduced replication API (not finished)

src/main/java/net/openhft/chronicle/hash/ChannelProvider.java

@@ -169,8 +169,7 @@ public final class ChannelProvider implements Closeable {
             newBitSet(systemModificationIterator.length());
     private final AtomicReferenceArray<ModificationIterator> modificationIterator =
             new AtomicReferenceArray<ModificationIterator>(128);
-    private final Set<AbstractChannelReplicator> replicators =
-            new CopyOnWriteArraySet<AbstractChannelReplicator>();
+    private final Set<Closeable> replicators = new CopyOnWriteArraySet<Closeable>();
 
     ChannelProvider(ChannelProviderBuilder builder) {
         localIdentifier = builder.identifier;
@@ -195,6 +194,10 @@ public final class ChannelProvider implements Closeable {
         add((short) 0, systemMessageQueue.asReplica, systemMessageQueue.asEntryExternalizable);
     }
 
+    public int maxEntrySize() {
+        return maxEntrySize;
+    }
+
     /**
      * creates a bit set based on a number of bits
      *
@@ -271,7 +274,7 @@ public final class ChannelProvider implements Closeable {
 
     @Override
     public void close() throws IOException {
-        for (AbstractChannelReplicator replicator : replicators) {
+        for (Closeable replicator : replicators) {
             replicator.close();
         }
         // indexed reverse loop to avoid ConcurrentModificationException and remove elements from
@@ -282,7 +285,7 @@ public final class ChannelProvider implements Closeable {
         }
     }
 
-    void add(AbstractChannelReplicator replicator) {
+    void add(Closeable replicator) {
         replicators.add(replicator);
     }
 
@@ -413,16 +416,14 @@ public final class ChannelProvider implements Closeable {
         protected Closeable applyTo(AbstractChronicleMapBuilder builder,
                                     Replica map, EntryExternalizable entryExternalizable,
                                     final ChronicleMap chronicleMap) {
-            if (builder.entrySize() > maxEntrySize) {
-                throw new IllegalArgumentException("During ReplicatingChannelBuilder setup, " +
-                        "maxEntrySize=" + maxEntrySize + " was specified, but map with " +
-                        "entrySize=" + builder.entrySize() + " is attempted to apply" +
-                        "to the replicator");
-            }
             add(chronicleChannel, map, entryExternalizable);
             return this;
         }
 
+        public ChannelProvider provider() {
+            return ChannelProvider.this;
+        }
+
         @Override
         public void close() throws IOException {
             chronicleChannelList.remove(chronicleChannels[chronicleChannel]);

src/main/java/net/openhft/chronicle/hash/TcpReplicationConfig.java

@@ -80,13 +80,8 @@ public class TcpReplicationConfig implements ReplicationConfig {
         return this;
     }
 
-
-    public long heartBeatInterval() {
-        return heartBeatInterval;
-    }
-
-    public TimeUnit heartBeatIntervalUnit() {
-        return heartBeatIntervalUnit;
+    public long heartBeatInterval(TimeUnit unit) {
+        return unit.convert(heartBeatInterval, heartBeatIntervalUnit);
     }
 
     public int serverPort() {

src/main/java/net/openhft/chronicle/hash/ThrottlingConfig.java

@@ -103,7 +103,7 @@ public class ThrottlingConfig {
      * @return maximum bits per the given time unit, or {@code 0} if there is no throttling
      */
     public final long throttling(TimeUnit perUnit) {
-        return throttlePerUnit().convert(throttle(), perUnit);
+        return throttlePerUnit.convert(throttle, perUnit);
     }
 
 
@@ -115,7 +115,7 @@ public class ThrottlingConfig {
      * @return the bucketing interval for throttling in the given time units
      */
     public long bucketInterval(TimeUnit unit) {
-        return unit.convert(bucketInterval(), bucketIntervalUnit());
+        return unit.convert(bucketInterval, bucketIntervalUnit);
     }
 
     /**
@@ -129,26 +129,7 @@ public class ThrottlingConfig {
     public ThrottlingConfig bucketInterval(long throttleBucketInterval, TimeUnit unit) {
         if (throttleBucketInterval <= 0L)
             throw new IllegalArgumentException();
-        return create(throttle(), throttlePerUnit(), throttleBucketInterval, unit);
-    }
-
-
-    long throttle() {
-        return throttle;
-    }
-
-
-    TimeUnit throttlePerUnit() {
-        return throttlePerUnit;
-    }
-
-
-    public long bucketInterval() {
-        return bucketInterval;
-    }
-
-    public TimeUnit bucketIntervalUnit() {
-        return bucketIntervalUnit;
+        return create(throttle, throttlePerUnit, throttleBucketInterval, unit);
     }
 
     @Override
@@ -168,10 +149,10 @@ public class ThrottlingConfig {
         }
         if (o instanceof ThrottlingConfig) {
             ThrottlingConfig that = (ThrottlingConfig) o;
-            return (this.throttle == that.throttle())
-                    && (this.throttlePerUnit.equals(that.throttlePerUnit()))
-                    && (this.bucketInterval == that.bucketInterval())
-                    && (this.bucketIntervalUnit.equals(that.bucketIntervalUnit()));
+            return (this.throttle == that.throttle)
+                    && (this.throttlePerUnit.equals(that.throttlePerUnit))
+                    && (this.bucketInterval == that.bucketInterval)
+                    && (this.bucketIntervalUnit.equals(that.bucketIntervalUnit));
         }
         return false;
     }

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

@@ -43,7 +43,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
 /**
  * @author Rob Austin.
  */
-public abstract class AbstractChannelReplicator implements Closeable {
+abstract class AbstractChannelReplicator implements Closeable {
 
     public static final int BITS_IN_A_BYTE = 8;
 

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

@@ -265,7 +265,7 @@ public abstract class AbstractChronicleMapBuilder<K, V,
         return self();
     }
 
-    public int entrySize() {
+    int entrySize() {
         if (entrySize > 0)
             return entryAndValueAlignment().alignSize(entrySize);
         int size = metaDataBytes;
@@ -881,6 +881,17 @@ public abstract class AbstractChronicleMapBuilder<K, V,
         if (map instanceof ReplicatedChronicleMap) {
             ReplicatedChronicleMap result = (ReplicatedChronicleMap) map;
             for (Replicator replicator : replicators.values()) {
+                if (replicator instanceof ChannelProvider.ChronicleChannel) {
+                    ChannelProvider.ChronicleChannel channel =
+                            (ChannelProvider.ChronicleChannel) replicator;
+                    if (entrySize() > channel.provider().maxEntrySize()) {
+                        throw new IllegalArgumentException("During ChannelProviderBuilder setup, " +
+                                "maxEntrySize=" + channel.provider().maxEntrySize() +
+                                " was specified, but map with " +
+                                "entrySize=" + entrySize() + " is attempted to apply" +
+                                "to the replicator");
+                    }
+                }
                 Closeable token = replicator.applyTo(this, result, result, map);
                 if (replicators.size() == 1 && token.getClass() == UdpReplicator.class) {
                     LOG.warn(

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

@@ -41,6 +41,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static java.nio.channels.SelectionKey.*;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.HEARTBEAT;
 
 /**
@@ -93,11 +94,9 @@ public class TcpReplicator extends AbstractChannelReplicator implements Closeabl
         this.statelessServerConnector = statelessServerConnector;
 
         final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig();
-        long throttleBucketInterval = TimeUnit.MILLISECONDS.convert(throttlingConfig
-                .bucketInterval(), throttlingConfig.bucketIntervalUnit());
+        long throttleBucketInterval = throttlingConfig.bucketInterval(MILLISECONDS);
 
-        heartBeatIntervalMillis = TimeUnit.MILLISECONDS.convert(replicationConfig.heartBeatInterval(),
-                replicationConfig.heartBeatIntervalUnit());
+        heartBeatIntervalMillis = replicationConfig.heartBeatInterval(MILLISECONDS);
 
         selectorTimeout = Math.min(heartBeatIntervalMillis, throttleBucketInterval);