Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-163 renamed channel replicator to channel provider

Browse Source
This commit is contained in:
Rob Austin
2014-10-13 19:03:43 +01:00
parent 02072f0e8a
commit 2aff512767
10 changed files with 90 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+10 -10
View File
@@ -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<Integer, CharSequence> 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();
src/main/java/net/openhft/chronicle/map/ChannelReplicator.java → src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+4 -4
View File
@@ -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<AbstractChannelReplicator> replicators =
new CopyOnWriteArraySet<AbstractChannelReplicator>();
ChannelReplicator(ChannelReplicatorBuilder builder) {
ChannelProvider(ChannelProviderBuilder builder) {
localIdentifier = builder.identifier;
maxEntrySize = builder.maxEntrySize;
chronicleChannels = new Replica[builder.maxNumberOfChronicles];
src/main/java/net/openhft/chronicle/map/ChannelReplicatorBuilder.java → src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java
+28 -20
View File
@@ -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,
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+1 -1
View File
@@ -562,7 +562,7 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
}
public ChronicleMapBuilder<K, V> channel(ChannelReplicator.ChronicleChannel chronicleChannel) {
public ChronicleMapBuilder<K, V> channel(ChannelProvider.ChronicleChannel chronicleChannel) {
this.identifier = chronicleChannel.identifier();
this.replicators.clear();
replicators.put(chronicleChannel.getClass(), chronicleChannel);
src/main/java/net/openhft/chronicle/map/Config.java
+7
View File
@@ -0,0 +1,7 @@
package net.openhft.chronicle.map;
/**
* @author Rob Austin.
*/
public interface Config {
}
src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java
+1 -1
View File
@@ -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;
src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java
+1 -2
View File
@@ -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());
}
src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java
+15 -12
View File
@@ -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<Integer, CharSequence> map1b;
private ChronicleMap<Integer, CharSequence> 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");
src/test/java/net/openhft/chronicle/map/PostClusterMapCreationTest.java
+13 -13
View File
@@ -43,8 +43,8 @@ public class PostClusterMapCreationTest {
private ChronicleMap<Integer, CharSequence> map1b;
private ChronicleMap<Integer, CharSequence> 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) {
src/test/java/net/openhft/chronicle/map/SingleMapClusterTest.java
+10 -9
View File
@@ -41,8 +41,8 @@ public class SingleMapClusterTest {
private ChronicleMap<Integer, CharSequence> 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) {