From c0d9099da9a31e13acb1599933eb722270582e64 Mon Sep 17 00:00:00 2001 From: leventov Date: Sat, 8 Nov 2014 05:07:13 +0300 Subject: [PATCH] Some renames in replication API, fixed potential issue with networkInterface configuration for udp multicast --- .../chronicle/hash/ChronicleHashBuilder.java | 20 +++---- .../hash/ChronicleHashInstanceConfig.java | 18 +++--- .../hash/replication/AbstractReplication.java | 16 ++--- .../hash/replication/ReplicationHub.java | 4 +- ...va => SingleChronicleHashReplication.java} | 13 ++--- ...java => TcpTransportAndNetworkConfig.java} | 56 ++++++++++-------- ...UdpConfig.java => UdpTransportConfig.java} | 38 ++++++------ .../map/AbstractChronicleMapBuilder.java | 58 +++++++++---------- .../chronicle/map/ChannelProvider.java | 8 +-- .../openhft/chronicle/map/InstanceConfig.java | 20 +++---- .../openhft/chronicle/map/NodeDiscovery.java | 14 ++--- .../openhft/chronicle/map/Replicators.java | 8 +-- .../openhft/chronicle/map/TcpReplicator.java | 6 +- .../chronicle/map/UdpChannelReplicator.java | 17 ++++-- .../openhft/chronicle/map/UdpReplicator.java | 4 +- .../chronicle/set/ChronicleSetBuilder.java | 8 +-- .../openhft/chronicle/set/InstanceConfig.java | 8 +-- .../net/openhft/chronicle/map/Builder.java | 4 +- .../chronicle/map/ChannelReplicationTest.java | 8 +-- ...tListenerTestWithTCPSocketReplication.java | 8 +-- .../map/PostChannelCreationTest.java | 6 +- .../chronicle/map/SingleMapChannelTest.java | 6 +- .../chronicle/map/StatelessClientTest.java | 24 ++++---- ...cketReplicationTest3wayPutReturnsNull.java | 4 +- .../map/TwoMapOnDifferentServers.java | 10 ++-- .../map/UDPSocketReplicationTest.java | 13 +++-- .../chronicle/map/example/ChannelExample.java | 8 +-- .../jrs166/map/StatelessChronicleMapTest.java | 6 +- .../SetForTCPSocketReplicationTest3way.java | 4 +- 29 files changed, 215 insertions(+), 202 deletions(-) rename src/main/java/net/openhft/chronicle/hash/replication/{SimpleReplication.java => SingleChronicleHashReplication.java} (71%) rename src/main/java/net/openhft/chronicle/hash/replication/{TcpConfig.java => TcpTransportAndNetworkConfig.java} (75%) rename src/main/java/net/openhft/chronicle/hash/replication/{UdpConfig.java => UdpTransportConfig.java} (75%) diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java index 7a1caecdc..458877355 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java @@ -18,8 +18,8 @@ package net.openhft.chronicle.hash; -import net.openhft.chronicle.hash.replication.SimpleReplication; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.hash.replication.TimeProvider; import net.openhft.chronicle.hash.serialization.*; import net.openhft.chronicle.map.*; @@ -237,7 +237,7 @@ public interface ChronicleHashBuilderBy default, hash containers, created by this builder doesn't replicate their data. * *

This method call overrides all previous replication configurations of this builder, made - * either by this method or {@link #replication(byte, TcpConfig)} shortcut method. + * either by this method or {@link #replication(byte, TcpTransportAndNetworkConfig)} shortcut method. * * @param replication the replication config * @return this builder back - * @see ChronicleHashInstanceConfig#replicated(SimpleReplication) - * @see #replication(byte, TcpConfig) + * @see ChronicleHashInstanceConfig#replicated(SingleChronicleHashReplication) + * @see #replication(byte, TcpTransportAndNetworkConfig) */ - B replication(SimpleReplication replication); + B replication(SingleChronicleHashReplication replication); /** * Shortcut for {@code replication(SimpleReplication.builder() @@ -392,10 +392,10 @@ public interface ChronicleHashBuilder statelessClient(InetSocketAddress remoteAddress); diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceConfig.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceConfig.java index 472bdf94f..21b01661b 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceConfig.java @@ -17,17 +17,17 @@ package net.openhft.chronicle.hash; import net.openhft.chronicle.hash.replication.ReplicationChannel; -import net.openhft.chronicle.hash.replication.SimpleReplication; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import java.io.File; import java.io.IOException; public interface ChronicleHashInstanceConfig { - ChronicleHashInstanceConfig replicated(byte identifier, TcpConfig tcpTransportAndNetwork); + ChronicleHashInstanceConfig replicated(byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork); - ChronicleHashInstanceConfig replicated(SimpleReplication replication); + ChronicleHashInstanceConfig replicated(SingleChronicleHashReplication replication); /** * Configures replication of the hash containers, created by this builder, via so called @@ -35,19 +35,19 @@ public interface ChronicleHashInstanceConfig { * the * section about Channels and ChannelProvider in ChronicleMap manual for more information. * - *

Another way to establish replication is {@link #replicated(SimpleReplication)} method - * or it's shortcut: {@link #replicated(byte, TcpConfig)}. + *

Another way to establish replication is {@link #replicated(SingleChronicleHashReplication)} method + * or it's shortcut: {@link #replicated(byte, TcpTransportAndNetworkConfig)}. * *

By default, hash containers, created by this builder doesn't replicate their data. * *

This method call overrides all previous replication configurations of this builder, made - * either by means of this method, {@link #replicated(SimpleReplication)} - * or {@link #replicated(byte, TcpConfig)} method calls. + * either by means of this method, {@link #replicated(SingleChronicleHashReplication)} + * or {@link #replicated(byte, TcpTransportAndNetworkConfig)} method calls. * * @param channel the channel responsible for gathering updates of hash containers, * created by this builder, and replicating them over network * @return this builder object back - * @see #replicated(SimpleReplication) + * @see #replicated(SingleChronicleHashReplication) */ ChronicleHashInstanceConfig replicatedViaChannel(ReplicationChannel channel); 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 53bc4f538..078df5519 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java @@ -20,8 +20,8 @@ import org.jetbrains.annotations.Nullable; class AbstractReplication { private final byte localIdentifier; - private final TcpConfig tcpConfig; - private final UdpConfig udpConfig; + private final TcpTransportAndNetworkConfig tcpConfig; + private final UdpTransportConfig udpConfig; AbstractReplication(byte localIdentifier, Builder builder) { this.localIdentifier = localIdentifier; @@ -34,25 +34,25 @@ class AbstractReplication { } @Nullable - public TcpConfig tcpTransportAndNetwork() { + public TcpTransportAndNetworkConfig tcpTransportAndNetwork() { return tcpConfig; } @Nullable - public UdpConfig udpTransport() { + public UdpTransportConfig udpTransport() { return udpConfig; } static abstract class Builder { - private TcpConfig tcpConfig = null; - private UdpConfig udpConfig = null; + private TcpTransportAndNetworkConfig tcpConfig = null; + private UdpTransportConfig udpConfig = null; - public B tcpTransportAndNetwork(TcpConfig tcpConfig) { + public B tcpTransportAndNetwork(TcpTransportAndNetworkConfig tcpConfig) { this.tcpConfig = tcpConfig; return (B) this; } - public B udpTransport(UdpConfig udpConfig) { + public B udpTransport(UdpTransportConfig udpConfig) { this.udpConfig = udpConfig; return (B) this; } 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 6b75785a6..abd316d24 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java @@ -63,12 +63,12 @@ public final class ReplicationHub extends AbstractReplication { private Builder() {} @Override - public Builder tcpTransportAndNetwork(TcpConfig tcpConfig) { + public Builder tcpTransportAndNetwork(TcpTransportAndNetworkConfig tcpConfig) { return super.tcpTransportAndNetwork(tcpConfig); } @Override - public Builder udpTransport(UdpConfig udpConfig) { + public Builder udpTransport(UdpTransportConfig udpConfig) { return super.udpTransport(udpConfig); } diff --git a/src/main/java/net/openhft/chronicle/hash/replication/SimpleReplication.java b/src/main/java/net/openhft/chronicle/hash/replication/SingleChronicleHashReplication.java similarity index 71% rename from src/main/java/net/openhft/chronicle/hash/replication/SimpleReplication.java rename to src/main/java/net/openhft/chronicle/hash/replication/SingleChronicleHashReplication.java index b32dbda8a..6dc8e058c 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/SimpleReplication.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/SingleChronicleHashReplication.java @@ -16,32 +16,31 @@ package net.openhft.chronicle.hash.replication; -//TODO better name -public final class SimpleReplication extends AbstractReplication { +public final class SingleChronicleHashReplication extends AbstractReplication { public static Builder builder() { return new Builder(); } - SimpleReplication(byte localIdentifier, Builder builder) { + SingleChronicleHashReplication(byte localIdentifier, Builder builder) { super(localIdentifier, builder); } public static final class Builder extends AbstractReplication.Builder { private Builder() {} - public SimpleReplication create(byte identifier) { + public SingleChronicleHashReplication createWithId(byte identifier) { check(); - return new SimpleReplication(identifier, this); + return new SingleChronicleHashReplication(identifier, this); } @Override - public Builder tcpTransportAndNetwork(TcpConfig tcpConfig) { + public Builder tcpTransportAndNetwork(TcpTransportAndNetworkConfig tcpConfig) { return super.tcpTransportAndNetwork(tcpConfig); } @Override - public Builder udpTransport(UdpConfig udpConfig) { + public Builder udpTransport(UdpTransportConfig udpConfig) { return super.udpTransport(udpConfig); } } diff --git a/src/main/java/net/openhft/chronicle/hash/replication/TcpConfig.java b/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java similarity index 75% rename from src/main/java/net/openhft/chronicle/hash/replication/TcpConfig.java rename to src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java index 97f97de26..bfc277e75 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/TcpConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java @@ -26,7 +26,7 @@ import static java.util.Collections.unmodifiableSet; import static java.util.concurrent.TimeUnit.SECONDS; -public final class TcpConfig { +public final class TcpTransportAndNetworkConfig { private static final int DEFAULT_PACKET_SIZE = 1024 * 8; private static final long DEFAULT_HEART_BEAT_INTERVAL = 20; @@ -41,10 +41,12 @@ public final class TcpConfig { private final TimeUnit heartBeatIntervalUnit; private final IdentifierListener identifierListener; - private TcpConfig(int serverPort, Set endpoints, int packetSize, - boolean autoReconnectedUponDroppedConnection, - ThrottlingConfig throttlingConfig, long heartBeatInterval, - TimeUnit heartBeatIntervalUnit, IdentifierListener identifierListener) { + private TcpTransportAndNetworkConfig(int serverPort, Set endpoints, + int packetSize, + boolean autoReconnectedUponDroppedConnection, + ThrottlingConfig throttlingConfig, long heartBeatInterval, + TimeUnit heartBeatIntervalUnit, + IdentifierListener identifierListener) { this.serverPort = serverPort; this.endpoints = endpoints; this.packetSize = packetSize; @@ -56,29 +58,30 @@ public final class TcpConfig { } - public static TcpConfig forSendingNode(int serverPort, InetSocketAddress... endpoints) { + public static TcpTransportAndNetworkConfig forSendingNode(int serverPort, + InetSocketAddress... endpoints) { return forSendingNode(serverPort, Arrays.asList(endpoints)); } - public static TcpConfig forSendingNode(int serverPort, + public static TcpTransportAndNetworkConfig forSendingNode(int serverPort, Collection endpoints) { if (endpoints.isEmpty()) throw new IllegalArgumentException("There should be some endpoints"); return forUnknownTopology(serverPort, endpoints); } - public static TcpConfig forReceivingOnlyNode(int serverPort) { + public static TcpTransportAndNetworkConfig forReceivingOnlyNode(int serverPort) { return forUnknownTopology(serverPort, Collections.emptyList()); } - public static TcpConfig forUnknownTopology(int serverPort, + public static TcpTransportAndNetworkConfig forUnknownTopology(int serverPort, Collection endpoints) { for (final InetSocketAddress endpoint : endpoints) { if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName())) throw new IllegalArgumentException("endpoint=" + endpoint + " can not point to the same port as the server"); } - return new TcpConfig( + return new TcpTransportAndNetworkConfig( serverPort, unmodifiableSet(new HashSet(endpoints)), DEFAULT_PACKET_SIZE, true, // autoReconnectedUponDroppedConnection @@ -92,8 +95,9 @@ public final class TcpConfig { return identifierListener; } - public TcpConfig nonUniqueIdentifierListener(final IdentifierListener identifierListener) { - return new TcpConfig(serverPort, endpoints, packetSize, + public TcpTransportAndNetworkConfig nonUniqueIdentifierListener( + final IdentifierListener identifierListener) { + return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, heartBeatIntervalUnit, identifierListener); } @@ -102,8 +106,9 @@ public final class TcpConfig { return autoReconnectedUponDroppedConnection; } - public TcpConfig autoReconnectedUponDroppedConnection(boolean autoReconnectedUponDroppedConnection) { - return new TcpConfig(serverPort, endpoints, packetSize, + public TcpTransportAndNetworkConfig autoReconnectedUponDroppedConnection( + boolean autoReconnectedUponDroppedConnection) { + return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, heartBeatIntervalUnit, identifierListener); } @@ -112,9 +117,9 @@ public final class TcpConfig { return throttlingConfig; } - public TcpConfig throttlingConfig(ThrottlingConfig throttlingConfig) { + public TcpTransportAndNetworkConfig throttlingConfig(ThrottlingConfig throttlingConfig) { ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "TCP"); - return new TcpConfig(serverPort, endpoints, packetSize, + return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, heartBeatIntervalUnit, identifierListener); } @@ -127,8 +132,8 @@ public final class TcpConfig { return serverPort; } - public TcpConfig serverPort(int serverPort) { - return new TcpConfig(serverPort, endpoints, packetSize, + public TcpTransportAndNetworkConfig serverPort(int serverPort) { + return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, heartBeatIntervalUnit, identifierListener); } @@ -137,13 +142,13 @@ public final class TcpConfig { return endpoints; } - public TcpConfig endpoints(Set endpoints) { + public TcpTransportAndNetworkConfig endpoints(Set endpoints) { for (final InetSocketAddress endpoint : endpoints) { if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName())) throw new IllegalArgumentException("endpoint=" + endpoint + " can not point to the same port as the server"); } - return new TcpConfig(serverPort, endpoints, packetSize, + return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, heartBeatIntervalUnit, identifierListener); } @@ -152,16 +157,17 @@ public final class TcpConfig { return packetSize; } - public TcpConfig packetSize(int packetSize) { + public TcpTransportAndNetworkConfig packetSize(int packetSize) { if (packetSize <= 0) throw new IllegalArgumentException(); - return new TcpConfig(serverPort, endpoints, packetSize, + return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, heartBeatIntervalUnit, identifierListener); } - public TcpConfig heartBeatInterval(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) { - return new TcpConfig(serverPort, endpoints, packetSize, + public TcpTransportAndNetworkConfig heartBeatInterval(long heartBeatInterval, + TimeUnit heartBeatIntervalUnit) { + return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, heartBeatIntervalUnit, identifierListener); } @@ -172,7 +178,7 @@ public final class TcpConfig { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; - TcpConfig that = (TcpConfig) o; + TcpTransportAndNetworkConfig that = (TcpTransportAndNetworkConfig) o; if (autoReconnectedUponDroppedConnection != that.autoReconnectedUponDroppedConnection) return false; if (heartBeatInterval != that.heartBeatInterval) return false; diff --git a/src/main/java/net/openhft/chronicle/hash/replication/UdpConfig.java b/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java similarity index 75% rename from src/main/java/net/openhft/chronicle/hash/replication/UdpConfig.java rename to src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java index e8a44433c..b86d564d5 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/UdpConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java @@ -25,16 +25,16 @@ import java.net.InetAddress; import java.net.NetworkInterface; -public final class UdpConfig { +public final class UdpTransportConfig { private final InetAddress address; private final int port; - private final NetworkInterface networkInterface; + private final @Nullable NetworkInterface networkInterface; private final ThrottlingConfig throttlingConfig; - private UdpConfig( + private UdpTransportConfig( InetAddress address, int port, - NetworkInterface networkInterface, + @Nullable NetworkInterface networkInterface, ThrottlingConfig throttlingConfig) { if (address == null) { throw new NullPointerException("Null address"); @@ -49,23 +49,16 @@ public final class UdpConfig { this.throttlingConfig = throttlingConfig; } - // TODO better name - public static UdpConfig simple(@NotNull InetAddress address, int port) { + public static UdpTransportConfig of(@NotNull InetAddress address, int port) { if (address.isMulticastAddress()) throw new IllegalArgumentException(); return create(address, port, null, ThrottlingConfig.noThrottling()); } - public static UdpConfig multiCast(@NotNull InetAddress address, int port, - @NotNull NetworkInterface networkInterface) { - if (!address.isMulticastAddress() || networkInterface == null) - throw new IllegalArgumentException(); - return create(address, port, networkInterface, ThrottlingConfig.noThrottling()); - } - - static UdpConfig create(InetAddress address, int port, NetworkInterface networkInterface, - ThrottlingConfig throttlingConfig) { - return new UdpConfig(address, port, networkInterface, throttlingConfig); + static UdpTransportConfig create(InetAddress address, int port, + @Nullable NetworkInterface networkInterface, + ThrottlingConfig throttlingConfig) { + return new UdpTransportConfig(address, port, networkInterface, throttlingConfig); } @NotNull @@ -102,8 +95,8 @@ public final class UdpConfig { if (o == this) { return true; } - if (o instanceof UdpConfig) { - UdpConfig that = (UdpConfig) o; + if (o instanceof UdpTransportConfig) { + UdpTransportConfig that = (UdpTransportConfig) o; return (this.address.equals(that.address())) && (this.port == that.port()) && (this.networkInterface.equals(that.networkInterface())) @@ -126,7 +119,14 @@ public final class UdpConfig { return h; } - public UdpConfig throttlingConfig(@NotNull ThrottlingConfig throttlingConfig) { + public UdpTransportConfig networkInterfaceForMulticast( + @Nullable NetworkInterface networkInterface) { + if (!address().isMulticastAddress()) + throw new IllegalArgumentException(); + return create(address(), port(), networkInterface, throttlingConfig()); + } + + public UdpTransportConfig throttlingConfig(@NotNull ThrottlingConfig throttlingConfig) { ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "UDP"); return create(address(), port(), networkInterface(), throttlingConfig); } diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java index 9a877ac9e..a7d8b2297 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java @@ -139,7 +139,7 @@ public abstract class AbstractChronicleMapBuilder defaultValueProvider = NullValueProvider.INSTANCE; private PrepareValueBytes prepareValueBytes = null; - private SimpleReplication simpleReplication = null; + private SingleChronicleHashReplication singleHashReplication = null; AbstractChronicleMapBuilder(Class keyClass, Class valueClass) { keyBuilder = new SerializationBuilder(keyClass, SerializationBuilder.Role.KEY); @@ -740,41 +740,41 @@ public abstract class AbstractChronicleMapBuilder> instance() { - return new InstanceConfig<>(this.clone(), simpleReplication, null, null); + return new InstanceConfig<>(this.clone(), singleHashReplication, null, null); } @Override public ChronicleMap createPersistedTo(File file) throws IOException { - return createWithFile(file, simpleReplication, null); + return createWithFile(file, singleHashReplication, null); } @Override public ChronicleMap create() throws IOException { - return createWithoutFile(simpleReplication, null); + return createWithoutFile(singleHashReplication, null); } ChronicleMap create(InstanceConfig ib) throws IOException { if (ib.file != null) { - return createWithFile(ib.file, ib.simpleReplication, ib.channel); + return createWithFile(ib.file, ib.singleHashReplication, ib.channel); } else { - return createWithoutFile(ib.simpleReplication, ib.channel); + return createWithoutFile(ib.singleHashReplication, ib.channel); } } - ChronicleMap createWithFile(File file, SimpleReplication simpleReplication, + ChronicleMap createWithFile(File file, SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) throws IOException { for (int i = 0; i < 10; i++) { if (file.exists() && file.length() > 0) { @@ -785,7 +785,7 @@ public abstract class AbstractChronicleMapBuilder) ois.readObject(); map.headerSize = roundUpMapHeaderSize(fis.getChannel().position()); map.createMappedStoreAndSegments(file); - return establishReplication(map, simpleReplication, channel); + return establishReplication(map, singleHashReplication, channel); } catch (ClassNotFoundException e) { throw new IOException(e); } finally { @@ -805,7 +805,7 @@ public abstract class AbstractChronicleMapBuilder map = newMap(simpleReplication, channel); + VanillaChronicleMap map = newMap(singleHashReplication, channel); FileOutputStream fos = new FileOutputStream(file); ObjectOutputStream oos = new ObjectOutputStream(fos); @@ -818,7 +818,7 @@ public abstract class AbstractChronicleMapBuilder createWithoutFile( - SimpleReplication simpleReplication, ReplicationChannel channel) throws IOException { - VanillaChronicleMap map = newMap(simpleReplication, channel); + SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) throws IOException { + VanillaChronicleMap map = newMap(singleHashReplication, channel); BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE, map.sizeInBytes(), true); map.createMappedStoreAndSegments(bytesStore); - return establishReplication(map, simpleReplication, channel); + return establishReplication(map, singleHashReplication, channel); } ChronicleMap createStatelessMap(StatelessMapConfig statelessBuilder) @@ -856,13 +856,13 @@ public abstract class AbstractChronicleMapBuilder newMap( - SimpleReplication simpleReplication, ReplicationChannel channel) throws IOException { - boolean replicated = simpleReplication != null || channel != null; + SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) throws IOException { + boolean replicated = singleHashReplication != null || channel != null; preMapConstruction(replicated); if (replicated) { byte identifier; - if (simpleReplication != null) { - identifier = simpleReplication.identifier(); + if (singleHashReplication != null) { + identifier = singleHashReplication.identifier(); } else { identifier = channel.hub().identifier(); } @@ -889,19 +889,19 @@ public abstract class AbstractChronicleMapBuilder establishReplication( - VanillaChronicleMap map, SimpleReplication simpleReplication, + VanillaChronicleMap map, SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) throws IOException { if (map instanceof ReplicatedChronicleMap) { - if (simpleReplication != null && channel != null) { + if (singleHashReplication != null && channel != null) { throw new AssertionError("Only one non-null replication should be passed"); } ReplicatedChronicleMap result = (ReplicatedChronicleMap) map; List replicators = new ArrayList<>(2); - if (simpleReplication != null) { - if (simpleReplication.tcpTransportAndNetwork() != null) - replicators.add(Replicators.tcp(simpleReplication.tcpTransportAndNetwork())); - if (simpleReplication.udpTransport() != null) - replicators.add(Replicators.udp(simpleReplication.udpTransport())); + if (singleHashReplication != null) { + if (singleHashReplication.tcpTransportAndNetwork() != null) + replicators.add(Replicators.tcp(singleHashReplication.tcpTransportAndNetwork())); + if (singleHashReplication.udpTransport() != null) + replicators.add(Replicators.udp(singleHashReplication.udpTransport())); } else { ReplicationHub hub = channel.hub(); int entrySize = entrySize(true); diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index 5976dc1b3..cd039e325 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -20,8 +20,8 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.hash.replication.TcpConfig; -import net.openhft.chronicle.hash.replication.UdpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.chronicle.hash.replication.UdpTransportConfig; import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.collection.SingleThreadedDirectBitSet; import net.openhft.lang.io.ByteBufferBytes; @@ -58,7 +58,7 @@ final class ChannelProvider implements Closeable { if (channelProvider != null) return channelProvider; channelProvider = new ChannelProvider(hub); - TcpConfig tcpConfig = hub.tcpTransportAndNetwork(); + TcpTransportAndNetworkConfig tcpConfig = hub.tcpTransportAndNetwork(); if (tcpConfig != null) { final TcpReplicator tcpReplicator = new TcpReplicator( channelProvider.asReplica, @@ -67,7 +67,7 @@ final class ChannelProvider implements Closeable { channelProvider.add(tcpReplicator); } - UdpConfig udpConfig = hub.udpTransport(); + UdpTransportConfig udpConfig = hub.udpTransport(); if (udpConfig != null) { final UdpReplicator udpReplicator = new UdpReplicator( diff --git a/src/main/java/net/openhft/chronicle/map/InstanceConfig.java b/src/main/java/net/openhft/chronicle/map/InstanceConfig.java index 014bf491e..e73d2778a 100644 --- a/src/main/java/net/openhft/chronicle/map/InstanceConfig.java +++ b/src/main/java/net/openhft/chronicle/map/InstanceConfig.java @@ -18,8 +18,8 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHashInstanceConfig; import net.openhft.chronicle.hash.replication.ReplicationChannel; -import net.openhft.chronicle.hash.replication.SimpleReplication; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import java.io.File; import java.io.IOException; @@ -28,30 +28,30 @@ final class InstanceConfig implements ChronicleHashInstanceConfig> { final AbstractChronicleMapBuilder mapBuilder; - final SimpleReplication simpleReplication; + final SingleChronicleHashReplication singleHashReplication; final ReplicationChannel channel; final File file; InstanceConfig(AbstractChronicleMapBuilder mapBuilder, - SimpleReplication simpleReplication, + SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel, File file) { this.mapBuilder = mapBuilder; - this.simpleReplication = simpleReplication; + this.singleHashReplication = singleHashReplication; this.channel = channel; this.file = file; } @Override public ChronicleHashInstanceConfig> replicated( - byte identifier, TcpConfig tcpTransportAndNetwork) { - return replicated(SimpleReplication.builder() - .tcpTransportAndNetwork(tcpTransportAndNetwork).create(identifier)); + byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork) { + return replicated(SingleChronicleHashReplication.builder() + .tcpTransportAndNetwork(tcpTransportAndNetwork).createWithId(identifier)); } @Override public ChronicleHashInstanceConfig> replicated( - SimpleReplication replication) { + SingleChronicleHashReplication replication) { return new InstanceConfig<>(mapBuilder, replication, null, file); } @@ -63,7 +63,7 @@ final class InstanceConfig @Override public ChronicleHashInstanceConfig> persistedTo(File file) { - return new InstanceConfig<>(mapBuilder, simpleReplication, channel, file); + return new InstanceConfig<>(mapBuilder, singleHashReplication, channel, file); } @Override diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index ecbb09cd2..bb181fd9a 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -19,8 +19,8 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.IdentifierListener; -import net.openhft.chronicle.hash.replication.TcpConfig; -import net.openhft.chronicle.hash.replication.UdpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.chronicle.hash.replication.UdpTransportConfig; import net.openhft.lang.collection.ATSDirectBitSet; import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.io.ByteBufferBytes; @@ -42,7 +42,7 @@ import java.util.concurrent.atomic.AtomicReference; import static java.nio.ByteBuffer.allocateDirect; import static java.util.concurrent.TimeUnit.SECONDS; -import static net.openhft.chronicle.hash.replication.UdpConfig.simple; +import static net.openhft.chronicle.hash.replication.UdpTransportConfig.of; import static net.openhft.chronicle.map.ConcurrentExpiryMap.getDefaultAddress; import static net.openhft.chronicle.map.DiscoveryNodeBytesMarshallable.ProposedNodes; import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.BOOTSTRAP_BYTES; @@ -56,7 +56,7 @@ import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG; public class NodeDiscovery { private final AddressAndPort ourAddressAndPort; - private final UdpConfig udpConfig; + private final UdpTransportConfig udpConfig; private final DiscoveryNodeBytesMarshallable discoveryNodeBytesMarshallable; private final AtomicReference nodeDiscoveryEventListenerAtomicReference = new AtomicReference(); @@ -71,7 +71,7 @@ public class NodeDiscovery { @NotNull InetAddress tcpAddress, @NotNull InetAddress udpBroadcastAddress) throws IOException { this.ourAddressAndPort = new AddressAndPort(tcpAddress.getAddress(), tcpPort); - this.udpConfig = simple(udpBroadcastAddress, udpBroadcastPort); + this.udpConfig = of(udpBroadcastAddress, udpBroadcastPort); knownNodes = new KnownNodes(); discoveryNodeBytesMarshallable = new DiscoveryNodeBytesMarshallable @@ -221,7 +221,7 @@ public class NodeDiscovery { // add our identifier and host:port to the list of known identifiers knownNodes.add(ourAddressAndPort, identifier); - final TcpConfig tcpConfig = TcpConfig + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forUnknownTopology(ourAddressAndPort.getPort(), toInetSocketCollection(knownHostPorts)) .heartBeatInterval(1, SECONDS) @@ -360,7 +360,7 @@ class NodeDiscoveryBroadcaster extends UdpChannelReplicator { * @throws java.io.IOException */ NodeDiscoveryBroadcaster( - final UdpConfig replicationConfig, + final UdpTransportConfig replicationConfig, final int serializedEntrySize, final BytesMarshallable externalizable) throws IOException { diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index b62683c44..14c76b778 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/src/main/java/net/openhft/chronicle/map/Replicators.java @@ -16,8 +16,8 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TcpConfig; -import net.openhft.chronicle.hash.replication.UdpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.chronicle.hash.replication.UdpTransportConfig; import org.jetbrains.annotations.NotNull; import java.io.Closeable; @@ -36,7 +36,7 @@ final class Replicators { private Replicators() { } - static Replicator tcp(final TcpConfig replicationConfig) { + static Replicator tcp(final TcpTransportAndNetworkConfig replicationConfig) { return new Replicator() { @Override @@ -60,7 +60,7 @@ final class Replicators { } static Replicator udp( - final UdpConfig replicationConfig) { + final UdpTransportConfig replicationConfig) { return new Replicator() { @Override diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index ab4d0ed7f..d57070057 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -19,7 +19,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.IdentifierListener; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.hash.replication.ThrottlingConfig; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; @@ -74,7 +74,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private final byte localIdentifier; private final int maxEntrySizeBytes; private final Replica.EntryExternalizable externalizable; - private final TcpConfig replicationConfig; + private final TcpTransportAndNetworkConfig replicationConfig; private final StatelessServerConnector statelessServerConnector; private long selectorTimeout; @@ -89,7 +89,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable */ public TcpReplicator(@NotNull final Replica replica, @NotNull final Replica.EntryExternalizable externalizable, - @NotNull final TcpConfig replicationConfig, + @NotNull final TcpTransportAndNetworkConfig replicationConfig, final int maxEntrySizeBytes, @Nullable StatelessServerConnector statelessServerConnector) throws IOException { diff --git a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java index 2896832d2..0a1b25946 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.UdpConfig; +import net.openhft.chronicle.hash.replication.UdpTransportConfig; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -69,7 +69,7 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. * @param localIdentifier just used for logging * @throws IOException */ - UdpChannelReplicator(@NotNull final UdpConfig replicationConfig, + UdpChannelReplicator(@NotNull final UdpTransportConfig replicationConfig, final int maxPayloadSize, final byte localIdentifier) throws IOException { @@ -178,10 +178,17 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. final InetSocketAddress hostAddress = new InetSocketAddress(port); client.configureBlocking(false); if (address.isMulticastAddress()) { - client.setOption(IP_MULTICAST_IF, networkInterface); client.setOption(SO_REUSEADDR, true); client.bind(hostAddress); - client.join(address, networkInterface); + if (networkInterface != null) { + // This is probably not needed, because client socket doesn't send datagrams, + // but since EVERYBODY on the internet configures this for any channels, and + // I don't see any harm this config could make, I leave it here + client.setOption(IP_MULTICAST_IF, networkInterface); + client.join(address, networkInterface); + } else { + client.join(address, NetworkInterface.getByInetAddress(hostAddress.getAddress())); + } if (LOG.isDebugEnabled()) LOG.debug("Connecting via multicast, group=" + address); } else { @@ -264,6 +271,8 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. .setOption(StandardSocketOptions.IP_MULTICAST_LOOP, false) .setOption(StandardSocketOptions.SO_BROADCAST, true) .setOption(SO_REUSEADDR, true); + if (networkInterface != null) + server.setOption(StandardSocketOptions.IP_MULTICAST_IF, networkInterface); // the registration has be be run on the same thread as the selector addPendingRegistration(new Runnable() { diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index 504804982..5eab17b22 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.UdpConfig; +import net.openhft.chronicle.hash.replication.UdpTransportConfig; import net.openhft.lang.io.ByteBufferBytes; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; @@ -43,7 +43,7 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi public UdpReplicator(@NotNull final Replica replica, @NotNull final Replica.EntryExternalizable entryExternalizable, - @NotNull final UdpConfig replicationConfig, + @NotNull final UdpTransportConfig replicationConfig, final int serializedEntrySize) throws IOException { diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index 3150e7b80..e91126a96 100644 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -19,8 +19,8 @@ package net.openhft.chronicle.set; import net.openhft.chronicle.hash.*; -import net.openhft.chronicle.hash.replication.SimpleReplication; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.hash.replication.TimeProvider; import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.chronicle.hash.serialization.BytesWriter; @@ -266,13 +266,13 @@ public class ChronicleSetBuilder } @Override - public ChronicleSetBuilder replication(SimpleReplication replication) { + public ChronicleSetBuilder replication(SingleChronicleHashReplication replication) { chronicleMapBuilder.replication(replication); return this; } @Override - public ChronicleSetBuilder replication(byte identifier, TcpConfig tcpTransportAndNetwork) { + public ChronicleSetBuilder replication(byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork) { chronicleMapBuilder.replication(identifier, tcpTransportAndNetwork); return this; } diff --git a/src/main/java/net/openhft/chronicle/set/InstanceConfig.java b/src/main/java/net/openhft/chronicle/set/InstanceConfig.java index 5b03265f7..e7fe620c4 100644 --- a/src/main/java/net/openhft/chronicle/set/InstanceConfig.java +++ b/src/main/java/net/openhft/chronicle/set/InstanceConfig.java @@ -18,8 +18,8 @@ package net.openhft.chronicle.set; import net.openhft.chronicle.hash.ChronicleHashInstanceConfig; import net.openhft.chronicle.hash.replication.ReplicationChannel; -import net.openhft.chronicle.hash.replication.SimpleReplication; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.map.ChronicleMap; import java.io.File; @@ -35,12 +35,12 @@ final class InstanceConfig implements ChronicleHashInstanceConfig> replicated( - byte identifier, TcpConfig tcpTransportAndNetwork) { + byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork) { return new InstanceConfig<>(mapConfig.replicated(identifier, tcpTransportAndNetwork)); } @Override - public ChronicleHashInstanceConfig> replicated(SimpleReplication replication) { + public ChronicleHashInstanceConfig> replicated(SingleChronicleHashReplication replication) { return new InstanceConfig<>(mapConfig.replicated(replication)); } diff --git a/src/test/java/net/openhft/chronicle/map/Builder.java b/src/test/java/net/openhft/chronicle/map/Builder.java index 8b5fd4afd..73ffce84b 100644 --- a/src/test/java/net/openhft/chronicle/map/Builder.java +++ b/src/test/java/net/openhft/chronicle/map/Builder.java @@ -19,7 +19,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import java.io.File; import java.io.IOException; @@ -72,7 +72,7 @@ public class Builder { final byte identifier, final int serverPort, final InetSocketAddress... endpoints) throws IOException { - TcpConfig tcpConfig = TcpConfig.forUnknownTopology(serverPort, Arrays.asList(endpoints)) + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.forUnknownTopology(serverPort, Arrays.asList(endpoints)) .heartBeatInterval(1L, TimeUnit.SECONDS); return ChronicleMapBuilder.of(kClass, vClass) .replication(identifier, tcpConfig); diff --git a/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java index 452f53657..0c4a46760 100644 --- a/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java @@ -20,7 +20,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -53,7 +53,7 @@ public class ChannelReplicationTest { @Before public void setup() throws IOException { { - TcpConfig tcpConfig = TcpConfig + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forSendingNode(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); @@ -67,8 +67,8 @@ public class ChannelReplicationTest { } { - TcpConfig tcpConfig = - TcpConfig.forReceivingOnlyNode(8087).heartBeatInterval(1, SECONDS); + TcpTransportAndNetworkConfig tcpConfig = + TcpTransportAndNetworkConfig.forReceivingOnlyNode(8087).heartBeatInterval(1, SECONDS); hubB = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig) .create((byte) 2); diff --git a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java index d2a070ef5..e4ecb2849 100644 --- a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java +++ b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.lang.io.Bytes; import org.junit.After; import org.junit.Assert; @@ -30,7 +30,7 @@ import java.net.InetSocketAddress; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; -import static net.openhft.chronicle.hash.replication.TcpConfig.forSendingNode; +import static net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig.forSendingNode; /** * Test ReplicatedChronicleMap where the Replicated is over a TCP Socket @@ -240,7 +240,7 @@ public class EventListenerTestWithTCPSocketReplication { } private ChronicleMapBuilder clientBuilder() { - TcpConfig tcpConfig = TcpConfig.forReceivingOnlyNode(8077); + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.forReceivingOnlyNode(8077); return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000L) .replication((byte) 1, tcpConfig) @@ -259,7 +259,7 @@ public class EventListenerTestWithTCPSocketReplication { } private ChronicleMapBuilder serverBuilder() { - TcpConfig tcpConfig = forSendingNode(8076, new InetSocketAddress("localhost", 8077)); + TcpTransportAndNetworkConfig tcpConfig = forSendingNode(8076, new InetSocketAddress("localhost", 8077)); return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000L) .replication((byte) 2, tcpConfig); diff --git a/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java b/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java index f555addca..2811e9e89 100644 --- a/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java +++ b/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java @@ -20,7 +20,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -62,7 +62,7 @@ public class PostChannelCreationTest { public void setup() throws IOException { { - final TcpConfig tcpConfig = TcpConfig + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forSendingNode(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); @@ -81,7 +81,7 @@ public class PostChannelCreationTest { { - final TcpConfig tcpConfig = TcpConfig + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forSendingNode(8087, new InetSocketAddress("localhost", 8086)) .heartBeatInterval(1, SECONDS); diff --git a/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java b/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java index 9d1af470e..f6be15db6 100644 --- a/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java +++ b/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java @@ -20,7 +20,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -60,7 +60,7 @@ public class SingleMapChannelTest { public void setup() throws IOException { { - final TcpConfig tcpConfig = TcpConfig + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forSendingNode(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); int maxEntrySize = 1024; @@ -77,7 +77,7 @@ public class SingleMapChannelTest { { - final TcpConfig tcpConfig = TcpConfig + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forSendingNode(8087, new InetSocketAddress("localhost", 8086)) .heartBeatInterval(1, SECONDS); diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 9806aa7c7..941fec300 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import org.junit.Assert; import org.junit.Test; @@ -56,7 +56,7 @@ public class StatelessClientTest { // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8086)).create(); } @@ -99,7 +99,7 @@ public class StatelessClientTest { // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8086)).create(); } @@ -140,7 +140,7 @@ public class StatelessClientTest { // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8086)).create(); } @@ -173,7 +173,7 @@ public class StatelessClientTest { // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8086)).create(); serverMap.put(10, "EXAMPLE-10"); } @@ -212,7 +212,7 @@ public class StatelessClientTest { // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8086)).create(); serverMap.put(10, "EXAMPLE-10"); } @@ -244,7 +244,7 @@ public class StatelessClientTest { // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8086)).create(); } @@ -287,7 +287,7 @@ public class StatelessClientTest { // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)) + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8086)) .putReturnsNull(true) .create(); } @@ -333,7 +333,7 @@ public class StatelessClientTest { // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)) + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8086)) .putReturnsNull(true) .removeReturnsNull(true) .create(); @@ -372,7 +372,7 @@ public class StatelessClientTest { // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)) + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8086)) .putReturnsNull(true) .removeReturnsNull(true) .create(); @@ -415,9 +415,9 @@ public class StatelessClientTest { // server { serverMap1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8086)).create(); serverMap2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8077)).create(); + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8077)).create(); } diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java index cbe2fb665..96da5b2ed 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -48,7 +48,7 @@ public class TCPSocketReplicationTest3wayPutReturnsNull { final byte identifier, final int serverPort, final InetSocketAddress... endpoints) throws IOException { - TcpConfig tcpConfig = TcpConfig.forUnknownTopology(serverPort, Arrays.asList(endpoints)); + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.forUnknownTopology(serverPort, Arrays.asList(endpoints)); return (T) ChronicleMapBuilder.of(Integer.class, CharSequence.class) .putReturnsNull(true) diff --git a/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java b/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java index 4fa82b1ad..b55cd3f5a 100644 --- a/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java +++ b/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java @@ -18,18 +18,16 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import org.junit.After; import org.junit.Before; import org.junit.Test; import java.io.Closeable; -import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; import static java.util.concurrent.TimeUnit.SECONDS; -import static net.openhft.chronicle.map.Builder.getPersistenceFile; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -45,8 +43,8 @@ public class TwoMapOnDifferentServers { @Before public void setup() throws IOException { - final TcpConfig tcpConfig = - TcpConfig.forSendingNode(8076, new InetSocketAddress("localhost", 8077)) + final TcpTransportAndNetworkConfig tcpConfig = + TcpTransportAndNetworkConfig.forSendingNode(8076, new InetSocketAddress("localhost", 8077)) .heartBeatInterval(1, SECONDS); @@ -56,7 +54,7 @@ public class TwoMapOnDifferentServers { map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8077) + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8077) .heartBeatInterval(1, SECONDS)).create(); } diff --git a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java index 68cb1d692..4f44e0b29 100644 --- a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java @@ -18,8 +18,8 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.SimpleReplication; -import net.openhft.chronicle.hash.replication.UdpConfig; +import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; +import net.openhft.chronicle.hash.replication.UdpTransportConfig; import org.junit.After; import org.junit.Before; @@ -41,13 +41,14 @@ public class UDPSocketReplicationTest { final int identifier, final int udpPort) throws IOException { - UdpConfig udpConfig = UdpConfig - .simple(Inet4Address.getByName("255.255.255.255"), udpPort); + UdpTransportConfig udpConfig = UdpTransportConfig + .of(Inet4Address.getByName("255.255.255.255"), udpPort); return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .replication(SimpleReplication.builder().udpTransport(udpConfig) - .create((byte) identifier)).create(); + .replication(SingleChronicleHashReplication.builder().udpTransport(udpConfig) + .createWithId((byte) identifier)) + .create(); } @Before diff --git a/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java b/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java index 96d30bb89..5936b0edd 100644 --- a/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java +++ b/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java @@ -1,7 +1,7 @@ package net.openhft.chronicle.map.example; import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; import org.junit.Assert; @@ -69,7 +69,7 @@ public class ChannelExample { byte identifier = (byte) 1; - TcpConfig tcpConfig = TcpConfig + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forSendingNode(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); @@ -104,8 +104,8 @@ public class ChannelExample { byte identifier = (byte) 2; - TcpConfig tcpConfig = - TcpConfig.forReceivingOnlyNode(8087).heartBeatInterval(1, SECONDS); + TcpTransportAndNetworkConfig tcpConfig = + TcpTransportAndNetworkConfig.forReceivingOnlyNode(8087).heartBeatInterval(1, SECONDS); hubOnServer2 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) diff --git a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java index 447bc7494..61b625979 100644 --- a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map.jrs166.map; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; import net.openhft.chronicle.map.ReadContext; @@ -192,7 +192,7 @@ public class StatelessChronicleMapTest extends JSR166TestCase { final ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, String.class) - .replication((byte) 1, TcpConfig.forReceivingOnlyNode(port)).create(); + .replication((byte) 1, TcpTransportAndNetworkConfig.forReceivingOnlyNode(port)).create(); final ChronicleMap statelessMap = ChronicleMapBuilder.of(Integer .class, String.class) @@ -206,7 +206,7 @@ public class StatelessChronicleMapTest extends JSR166TestCase { IOException { final ChronicleMap serverMap = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(port)).create(); + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(port)).create(); final ChronicleMap statelessMap = ChronicleMapBuilder.of(CharSequence diff --git a/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java b/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java index c7d7864ee..786b65cc8 100644 --- a/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java +++ b/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.set; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -52,7 +52,7 @@ public class SetForTCPSocketReplicationTest3way { final int serverPort, final InetSocketAddress... endpoints) throws IOException { return (T) ChronicleSetBuilder.of(Integer.class) - .replication(identifier, TcpConfig.forUnknownTopology(serverPort, asList(endpoints)) + .replication(identifier, TcpTransportAndNetworkConfig.forUnknownTopology(serverPort, asList(endpoints)) .heartBeatInterval(1L, SECONDS)) .createPersistedTo(getPersistenceFile()); }