diff --git a/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java b/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java index 683a0a805..dd4079e5d 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java @@ -56,7 +56,8 @@ public abstract class AbstractReplication { } /** - * Builder of {@link AbstractReplication} configurations. + * Builder of {@link AbstractReplication} configurations. This class and it's subclasses are + * mutable, configuration methods mutate the builder and return it back for convenient chaining. * * @param the concrete {@link AbstractReplication} subclass: {@link * SingleChronicleHashReplication} or {@link ReplicationHub} @@ -77,6 +78,14 @@ public abstract class AbstractReplication { return (B) this; } + /** + * Configures UDP transport settings, used by Replications, created by this builder. + * {@code null} means that UDP transport shouldn't being used. + * + * @param udpConfig the new UDP transport config for replications, created by this builder. + * @return this builder back, for chaining + * @see AbstractReplication#udpTransport() + */ @NotNull public B udpTransport(@Nullable UdpTransportConfig udpConfig) { this.udpConfig = udpConfig; @@ -89,10 +98,24 @@ public abstract class AbstractReplication { return (B) this; } + /** + * Creates a Replication instance with the given node (server) identifier. + * + * @param identifier the node (server) identifier of the returned replication + * @return a new Replication instance with the specified node (server) identifier + * @throws IllegalArgumentException if the given identifier is non-positive + * @throws IllegalStateException if neither {@link #tcpTransportAndNetwork( + * TcpTransportAndNetworkConfig)} nor {@link #udpTransport(UdpTransportConfig)} are + * configured to non-{@code null}. At least one of the transport-level configs should be + * specified. + */ @NotNull public abstract R createWithId(byte identifier); - void check() { + void check(byte identifier) { + if (identifier <= 0) + throw new IllegalArgumentException("Identifier must be positive, " + identifier + + " given"); if (udpConfig == null && tcpConfig == null) throw new IllegalStateException( "At least one transport method (TCP or UDP) should be configured"); diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java index 0ea18e674..aa7e23b4d 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java @@ -17,6 +17,24 @@ package net.openhft.chronicle.hash.replication; +import net.openhft.chronicle.hash.ChronicleHash; +import net.openhft.chronicle.hash.ChronicleHashBuilder; +import net.openhft.chronicle.hash.ChronicleHashInstanceConfig; +import org.jetbrains.annotations.NotNull; + +/** + * A token which should be created via {@link ReplicationHub#createChannel(short)} call and passed + * to {@link ChronicleHashInstanceConfig#replicatedViaChannel(ReplicationChannel)} method, to + * establish a channel replication. See + * + * the corresponding section in ChronicleMap manual for more information. + * + *

A {@code ReplicationChannel} could be used to replicate only one {@link ChronicleHash} + * instance. + * + * @see ReplicationHub + * @see ReplicationHub#createChannel(short) + */ public final class ReplicationChannel { private ReplicationHub hub; private final short channelId; @@ -26,10 +44,25 @@ public final class ReplicationChannel { this.channelId = channelId; } + /** + * Returns the {@link ReplicationHub} on which this {@code ReplicationChannel} was {@linkplain + * ReplicationHub#createChannel(short) created}. + * + * @return the {@code ReplicationHub} to which this {@code ReplicationChannel} belongs + * @see ReplicationHub#createChannel(short) + */ + @NotNull public ReplicationHub hub() { return hub; } + /** + * Returns the identifier of this channel, with which was {@linkplain + * ReplicationHub#createChannel(short) created}. + * + * @return the identifier of this channel + * @see ReplicationHub#createChannel(short) + */ public short channelId() { return channelId; } diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java index 41bc3f135..b841cf987 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java @@ -16,10 +16,46 @@ package net.openhft.chronicle.hash.replication; +import net.openhft.chronicle.hash.ChronicleHash; +import net.openhft.chronicle.hash.ChronicleHashBuilder; +import net.openhft.chronicle.hash.ChronicleHashInstanceConfig; +import net.openhft.chronicle.map.ChronicleMap; +import net.openhft.chronicle.set.ChronicleSet; import org.jetbrains.annotations.NotNull; +/** + * This class holds all configurations of + * + * multicontainer replication, which is usable, when you want to replicate several {@linkplain + * ChronicleHash ChronicleHashes} ({@linkplain ChronicleMap maps}, {@linkplain ChronicleSet sets}, + * etc.) between same servers. {@code ReplicationHub} allows to share TCP/UDP connection, sockets, + * buffers, worker threads for that, considerably reducing resources usage and increasing cumulative + * (for all replicated containers) replication throughput. See + * + * the corresponding section in ChronicleMap manual for more information. + * + *

Create instances of this class using this pattern: 

{@code
+ * ReplicationHub hub = ReplicationHub.builder()
+ *     .tcpTransportAndNetwork(tcpConfig)
+ *     // more configurations...
+ *     .createWithId(identifierOfThisServerWithinTheChronicleReplicationNetwork);}
+ * Then, given you prepared a {@link ChronicleHashBuilder builder} to create a {@link ChronicleHash + * }, configure it's replication via channel of this hub like this: 
{@code
+ * ChronicleMap myMap = builder.instance()
+ *     .replicatedViaChannel(hub.createChannel(myMapChannelId))
+ *     .persistedTo(myMapFile) // optional, for this example
+ *     .create();}
+ * + * @see ChronicleHashInstanceConfig#replicatedViaChannel(ReplicationChannel) + * @see ReplicationChannel + */ public final class ReplicationHub extends AbstractReplication { + /** + * Creates and returns a new {@link ReplicationHub.Builder}. + * + * @return a new {@link ReplicationHub.Builder} + */ @NotNull public static Builder builder() { return new Builder(); @@ -34,14 +70,48 @@ public final class ReplicationHub extends AbstractReplication { channels = new ReplicationChannel[builder.maxNumberOfChronicles]; } + /** + * Returns the maximum {@linkplain ChronicleHashBuilder#entrySize(int) entry size} of {@link + * ChronicleHash}es, replicated by this {@code ReplicationHub}, could have. {@code + * RuntimeException} is thrown on attempt to replicate with this {@code ReplicationHub} + * a {@code ChronicleHash} with larger entry size. + * + * @return the maximum entry size + * @see Builder#maxEntrySize(int) + */ public int maxEntrySize() { return maxEntrySize; } + /** + * Returns the maximum number of channels could be {@linkplain #createChannel(short) created} + * for this {@code ReplicationHub}. + * + * @return the maximum number of channels + * @see Builder#maxNumberOfChannels() + * @see #createChannel(short) + */ public int maxNumberOfChannels() { return channels.length; } + /** + * Creates a new {@link ReplicationChannel} in this {@code ReplicationHub} with the given + * identifier. Identifier shouldn't be lesser than zero and greater or equal to + * {@code maxNumberOfChannels() - 1}. On a {@code ReplicationHub} instance, + * {@code ReplicationChannel} could be created only once for each possible {@code channelId} + * value. + * + * @param channelId the identifier of the channel. Should be equal for replicated containers + * on different nodes (servers) + * @return a new {@code ReplicationChannel} instance, in fact just incapsulating the given + * {@code channelId} for this {@code ReplicationHub} + * @throws IllegalArgumentException if the specified {@code channelId} is out of
+ * [0, {@link #maxNumberOfChannels()}) range + * @throws IllegalStateException if {@code ReplicationChannel} with the specified + * {@code channelId} has already been acquired on this {@code ReplicationHub} + * @see ReplicationChannel + */ public synchronized ReplicationChannel createChannel(short channelId) { if (channelId < 0) throw new IllegalArgumentException("channelId should be positive"); @@ -49,25 +119,50 @@ public final class ReplicationHub extends AbstractReplication { throw new IllegalArgumentException("maxNumberOfChannels is configured (or defaulted) " + "to " + maxNumberOfChannels() + ", channelId=" + channelId + " is requested"); if (channels[channelId] != null) - throw new IllegalArgumentException("The requested channelId=" + channelId + + throw new IllegalStateException("The requested channelId=" + channelId + " is already in use"); ReplicationChannel channel = new ReplicationChannel(this, channelId); channels[channelId] = channel; return channel; } + /** + * Builder of {@link ReplicationHub}s. + */ public static final class Builder extends AbstractReplication.Builder { private int maxEntrySize = 1024; private int maxNumberOfChronicles = 128; private Builder() {} + /** + * Configures the maximum {@linkplain ChronicleHashBuilder#entrySize(int) entry size} of + * {@link ChronicleHash}es, replicated via {@code ReplicationHub}s, created by this builder. + * + *

Default value is {@code 1024}. + * + * @param maxEntrySize {@link ReplicationHub#maxEntrySize()} of {@link ReplicationHub}s, + * created by this builder + * @return this builder object back, for chaining + * @see ReplicationHub#maxEntrySize() + */ @NotNull public Builder maxEntrySize(int maxEntrySize) { this.maxEntrySize = maxEntrySize; return this; } + /** + * Configures the maximum number of channels could be {@linkplain #createChannel(short) + * created for {@code ReplicationHub}s, created by this builder. + * + *

Default value is {@code 128}. + * + * @param maxNumberOfChannels {@link ReplicationHub#maxNumberOfChannels()} of + * {@link ReplicationHub}s, created by this builder + * @return this builder object back, for chaining + * @see ReplicationHub#maxNumberOfChannels() + */ @NotNull public Builder maxNumberOfChannels(int maxNumberOfChannels) { this.maxNumberOfChronicles = maxNumberOfChannels;