diff --git a/README.md b/README.md index 338d01678..d0dfb17bd 100644 --- a/README.md +++ b/README.md @@ -46,7 +46,7 @@ Click here to get the [Latest Version Number](http://search.maven.org/#search%7C * [Identifier](https://github.com/OpenHFT/Chronicle-Map#identifier) * [Port](https://github.com/OpenHFT/Chronicle-Map#port) * [Heart Beat Interval](https://github.com/OpenHFT/Chronicle-Map#heart-beat-interval) -* [Channels and the Channel Replicator](https://github.com/OpenHFT/Chronicle-Map#channels-and-channelreplicator) +* [Channels and the Channel Provider](https://github.com/OpenHFT/Chronicle-Map#channels-and-channelprovider) #### Miscellaneous @@ -512,7 +512,7 @@ A heartbeat will only be send if no data is transmitted, if the maps are constan no heartbeat message is sent. If a map does not receive either data of a heartbeat the connection is dropped and re-established. -# Channels and ChannelReplicator +# Channels and ChannelProvider Chronicle Map TCP Replication lets you distribute a single Chronicle Map, to a number of servers across your network. Replication is point to point and the data transfer is bidirectional, so in the @@ -529,28 +529,28 @@ when using Chronicle Channels its the channels that are given the unique identif ``` java byte identifier = 2; int maxEntrySize = 1024; -ChannelReplicator replicator = new ChannelReplicatorBuilder(identifier, maxEntrySize) +ChannelProvider replicator = new ChannelProviderBuilder(identifier, maxEntrySize) .create(); ``` In this example above the channel is given the identifier of 2 In addition to specifying the identifier we also have to set the maximum entry size, this sets -the size of the memory buffers within the ChannelReplicator. This has to be set manually, with channels you -are able to attach additional maps to a ChannelReplicator once its up and running, so the maximum size of each +the size of the memory buffers within the ChannelProvider. This has to be set manually, with channels you +are able to attach additional maps to a ChannelProvider once its up and running, so the maximum size of each entry in the map can not be known in advance and we don’t currently support automatic resizing of buffers. -Once you have created the ChannelReplicator you should attach your tcp configuration +Once you have created the ChannelProvider you should attach your tcp configuration ``` java byte identifier = 2; int maxEntrySize =1024; -ChannelReplicator replicator = new ChannelReplicatorBuilder(identifier, 1024) +ChannelProvider replicator = new ChannelProviderBuilder(identifier, 1024) .tcpReplication(tcpConfig) .create();; ``` -Attaching ChannelReplicator replication to the map: +Attaching ChannelProvider replication to the map: ``` java ChronicleMap map = ChronicleMapBuilder.of(Integer.class, CharSequence.class) @@ -572,8 +572,8 @@ If you inadvertently got the chronicle channels around the wrong way, then chron to replicate the wrong maps data. The chronicle channels don't have to be in order but they must be unique for each map you have. -Once you have created the ChannelReplicator you may wish to hold onto the reference so that you can call close -once you have finished, this will close everything in the ChannelReplicator +Once you have created the ChannelProvider you may wish to hold onto the reference so that you can call close +once you have finished, this will close everything in the ChannelProvider ``` java replicator.close(); diff --git a/src/main/java/net/openhft/chronicle/map/ChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java similarity index 98% rename from src/main/java/net/openhft/chronicle/map/ChannelReplicator.java rename to src/main/java/net/openhft/chronicle/map/ChannelProvider.java index 4aaa95158..c51a56e8c 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -45,8 +45,8 @@ import static net.openhft.chronicle.map.Replica.ModificationNotifier.NOP; /** * @author Rob Austin. */ -public final class ChannelReplicator implements Closeable { - private static final Logger LOG = LoggerFactory.getLogger(ChannelReplicator.class.getName()); +public final class ChannelProvider implements Closeable { + private static final Logger LOG = LoggerFactory.getLogger(ChannelProvider.class.getName()); private static final byte BOOTSTRAP_MESSAGE = 'B'; final EntryExternalizable asEntryExternalizable = new EntryExternalizable() { @@ -152,7 +152,7 @@ public final class ChannelReplicator implements Closeable { @Override public void close() throws IOException { - ChannelReplicator.this.close(); + ChannelProvider.this.close(); } }; private final int maxEntrySize; @@ -169,7 +169,7 @@ public final class ChannelReplicator implements Closeable { private final Set replicators = new CopyOnWriteArraySet(); - ChannelReplicator(ChannelReplicatorBuilder builder) { + ChannelProvider(ChannelProviderBuilder builder) { localIdentifier = builder.identifier; maxEntrySize = builder.maxEntrySize; chronicleChannels = new Replica[builder.maxNumberOfChronicles]; diff --git a/src/main/java/net/openhft/chronicle/map/ChannelReplicatorBuilder.java b/src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java similarity index 67% rename from src/main/java/net/openhft/chronicle/map/ChannelReplicatorBuilder.java rename to src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java index 537095985..ca1a00f00 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelReplicatorBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java @@ -21,40 +21,48 @@ import java.io.IOException; /** * @author Rob Austin. */ -public final class ChannelReplicatorBuilder { +public final class ChannelProviderBuilder { - final byte identifier; - final int maxEntrySize; + byte identifier; + int maxEntrySize = 1024; int maxNumberOfChronicles = 128; private UdpReplicationConfig udpReplicationConfig = null; private TcpReplicationConfig tcpReplicationConfig = null; - ChannelReplicatorBuilder(byte identifier, final int maxEntrySize) { - this.identifier = identifier; + + + public int maxEntrySize() { + return maxEntrySize; + } + + public ChannelProviderBuilder maxEntrySize(int maxEntrySize) { this.maxEntrySize = maxEntrySize; - if (identifier <= 0) { - throw new IllegalArgumentException("Identifier must be positive, identifier=" + - identifier); + return this; + } + + public ChannelProviderBuilder replicators(byte identifier, Config... configs) { + this.identifier = identifier; + + + for (Config config : configs) { + + if (config instanceof TcpReplicationConfig) { + this.tcpReplicationConfig = (TcpReplicationConfig) config; + } else if (config instanceof UdpReplicationConfig) { + this.udpReplicationConfig = (UdpReplicationConfig) config; + } else + throw new UnsupportedOperationException(); } - } - - public ChannelReplicatorBuilder udpReplication(UdpReplicationConfig replicationConfig) { - this.udpReplicationConfig = replicationConfig; return this; } - public ChannelReplicatorBuilder tcpReplication(TcpReplicationConfig replicationConfig) { - this.tcpReplicationConfig = replicationConfig; - return this; - } - - public ChannelReplicatorBuilder maxNumberOfChronicles(int maxNumberOfChronicles) { + public ChannelProviderBuilder maxNumberOfChronicles(int maxNumberOfChronicles) { this.maxNumberOfChronicles = maxNumberOfChronicles; return this; } - public ChannelReplicator create() throws IOException { - final ChannelReplicator replicatingCluster = new ChannelReplicator(this); + public ChannelProvider create() throws IOException { + final ChannelProvider replicatingCluster = new ChannelProvider(this); if (tcpReplicationConfig != null) { final TcpReplicator tcpReplicator = new TcpReplicator( replicatingCluster.asReplica, diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index ff33d1e01..6582ea539 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -562,7 +562,7 @@ public class ChronicleMapBuilder implements Cloneable, } - public ChronicleMapBuilder channel(ChannelReplicator.ChronicleChannel chronicleChannel) { + public ChronicleMapBuilder channel(ChannelProvider.ChronicleChannel chronicleChannel) { this.identifier = chronicleChannel.identifier(); this.replicators.clear(); replicators.put(chronicleChannel.getClass(), chronicleChannel); diff --git a/src/main/java/net/openhft/chronicle/map/Config.java b/src/main/java/net/openhft/chronicle/map/Config.java new file mode 100644 index 000000000..6e0d9a436 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/Config.java @@ -0,0 +1,7 @@ +package net.openhft.chronicle.map; + +/** + * @author Rob Austin. + */ +public interface Config { +} diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java index b213f4455..5ea8cb1c1 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java @@ -29,7 +29,7 @@ import static java.util.Collections.unmodifiableSet; import static java.util.concurrent.TimeUnit.SECONDS; -public class TcpReplicationConfig { +public class TcpReplicationConfig implements Config { private static final int DEFAULT_PACKET_SIZE = 1024 * 8; private static final long DEFAULT_HEART_BEAT_INTERVAL = 20; diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java index 0e02d9e9d..a1e8ad0aa 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java @@ -23,7 +23,7 @@ import java.net.InetAddress; import java.net.NetworkInterface; -public abstract class UdpReplicationConfig { +public abstract class UdpReplicationConfig implements Config { /** * Package-private constructor forbids subclassing from outside of the package @@ -37,7 +37,6 @@ public abstract class UdpReplicationConfig { throw new IllegalArgumentException(); - return create(address, port, null, ThrottlingConfig.noThrottling()); } diff --git a/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java index 40344799a..086dffb86 100644 --- a/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java @@ -17,7 +17,10 @@ package net.openhft.chronicle.map; -import org.junit.*; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; import java.io.Closeable; import java.io.IOException; @@ -40,8 +43,8 @@ public class ChannelReplicationTest { private ChronicleMap map1b; private ChronicleMap map2b; - private ChannelReplicator channelReplicatorA; - private ChannelReplicator channelReplicatorB; + private ChannelProvider channelProviderA; + private ChannelProvider channelProviderB; @Before @@ -52,13 +55,13 @@ public class ChannelReplicationTest { .of(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); - channelReplicatorA = new ChannelReplicatorBuilder((byte) 1, 1024) - .tcpReplication(tcpConfig) - .create(); + channelProviderA = new ChannelProviderBuilder() + .replicators((byte) 1, tcpConfig) + .create(); map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .channel(channelReplicatorA.createChannel((short) 1)) + .channel(channelProviderA.createChannel((short) 1)) .create(getPersistenceFile()); } @@ -66,12 +69,12 @@ public class ChannelReplicationTest { TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(8087).heartBeatInterval(1, SECONDS); - channelReplicatorB = new ChannelReplicatorBuilder((byte) 2, 1024).tcpReplication(tcpConfig) + channelProviderB = new ChannelProviderBuilder().replicators((byte) 2, tcpConfig) .create(); map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .channel(channelReplicatorB.createChannel((short) 1)) + .channel(channelProviderB.createChannel((short) 1)) .create(getPersistenceFile()); } } @@ -79,7 +82,7 @@ public class ChannelReplicationTest { @After public void tearDown() throws InterruptedException { - for (final Closeable closeable : new Closeable[]{channelReplicatorA, channelReplicatorB}) { + for (final Closeable closeable : new Closeable[]{channelProviderA, channelProviderB}) { try { closeable.close(); } catch (IOException e) { @@ -94,13 +97,13 @@ public class ChannelReplicationTest { map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .channel(channelReplicatorB.createChannel((short) 2)) + .channel(channelProviderB.createChannel((short) 2)) .create(getPersistenceFile()); map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .channel(channelReplicatorA.createChannel((short) 2)) + .channel(channelProviderA.createChannel((short) 2)) .create(getPersistenceFile()); map2a.put(1, "EXAMPLE-2"); diff --git a/src/test/java/net/openhft/chronicle/map/PostClusterMapCreationTest.java b/src/test/java/net/openhft/chronicle/map/PostClusterMapCreationTest.java index 7cabde3c1..fba74f285 100644 --- a/src/test/java/net/openhft/chronicle/map/PostClusterMapCreationTest.java +++ b/src/test/java/net/openhft/chronicle/map/PostClusterMapCreationTest.java @@ -43,8 +43,8 @@ public class PostClusterMapCreationTest { private ChronicleMap map1b; private ChronicleMap map2b; - private ChannelReplicator channelReplicatorA; - private ChannelReplicator channelReplicatorB; + private ChannelProvider channelProviderA; + private ChannelProvider channelProviderB; public static File getPersistenceFile() { String TMP = System.getProperty("java.io.tmpdir"); @@ -59,43 +59,43 @@ public class PostClusterMapCreationTest { public void setup() throws IOException { { - final TcpReplicationConfig tcpReplicationConfig = TcpReplicationConfig + final TcpReplicationConfig tcpConfig = TcpReplicationConfig .of(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); - channelReplicatorA = new ChannelReplicatorBuilder((byte) 1, 1024) - .tcpReplication(tcpReplicationConfig).create(); + channelProviderA = new ChannelProviderBuilder() + .replicators((byte) 1, tcpConfig).create(); // this is how you add maps after the custer is created map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .channel(channelReplicatorA.createChannel((short) 1)) + .channel(channelProviderA.createChannel((short) 1)) .create(getPersistenceFile()); map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .channel(channelReplicatorA.createChannel((short) 2)) + .channel(channelProviderA.createChannel((short) 2)) .create(getPersistenceFile()); } { - final TcpReplicationConfig tcpReplicationConfig = TcpReplicationConfig + final TcpReplicationConfig tcpConfig = TcpReplicationConfig .of(8087, new InetSocketAddress("localhost", 8086)) .heartBeatInterval(1, SECONDS); - channelReplicatorB = new ChannelReplicatorBuilder((byte) 2, 1024) - .tcpReplication(tcpReplicationConfig).create(); + channelProviderB = new ChannelProviderBuilder() + .replicators((byte) 2, tcpConfig).create(); // this is how you add maps after the custer is created map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .channel(channelReplicatorB.createChannel((short) 1)) + .channel(channelProviderB.createChannel((short) 1)) .create(getPersistenceFile()); map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .channel(channelReplicatorB.createChannel((short) 2)) + .channel(channelProviderB.createChannel((short) 2)) .create(getPersistenceFile()); } @@ -105,7 +105,7 @@ public class PostClusterMapCreationTest { @After public void tearDown() throws InterruptedException { - for (final Closeable closeable : new Closeable[]{channelReplicatorA, channelReplicatorB}) { + for (final Closeable closeable : new Closeable[]{channelProviderA, channelProviderB}) { try { closeable.close(); } catch (IOException e) { diff --git a/src/test/java/net/openhft/chronicle/map/SingleMapClusterTest.java b/src/test/java/net/openhft/chronicle/map/SingleMapClusterTest.java index cd0e870d0..82ede45f6 100644 --- a/src/test/java/net/openhft/chronicle/map/SingleMapClusterTest.java +++ b/src/test/java/net/openhft/chronicle/map/SingleMapClusterTest.java @@ -41,8 +41,8 @@ public class SingleMapClusterTest { private ChronicleMap map1b; - private ChannelReplicator channelReplicatorA; - private ChannelReplicator channelReplicatorB; + private ChannelProvider channelProviderA; + private ChannelProvider channelProviderB; public static File getPersistenceFile() { String TMP = System.getProperty("java.io.tmpdir"); @@ -60,12 +60,12 @@ public class SingleMapClusterTest { .of(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); - channelReplicatorA = new ChannelReplicatorBuilder((byte) 1, 1024) - .tcpReplication(tcpReplicationConfig).create(); + channelProviderA = new ChannelProviderBuilder() + .replicators((byte) 1, tcpReplicationConfig).create(); // this is how you add maps after the custer is created map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .channel(channelReplicatorA.createChannel((short) 1)) + .channel(channelProviderA.createChannel((short) 1)) .create(getPersistenceFile()); } @@ -75,11 +75,12 @@ public class SingleMapClusterTest { .of(8087, new InetSocketAddress("localhost", 8086)) .heartBeatInterval(1, SECONDS); - channelReplicatorB = new ChannelReplicatorBuilder((byte) 2, 1024) - .tcpReplication(tcpReplicationConfig).create(); + channelProviderB = new ChannelProviderBuilder() + .replicators((byte) 2, tcpReplicationConfig).create(); + // this is how you add maps after the custer is created map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .channel(channelReplicatorB.createChannel((short) 1)) + .channel(channelProviderB.createChannel((short) 1)) .entries(1000) .create(getPersistenceFile()); } @@ -88,7 +89,7 @@ public class SingleMapClusterTest { @After public void tearDown() throws InterruptedException { - for (final Closeable closeable : new Closeable[]{channelReplicatorA, channelReplicatorB}) { + for (final Closeable closeable : new Closeable[]{channelProviderA, channelProviderB}) { try { closeable.close(); } catch (IOException e) {