diff --git a/README.md b/README.md index 72c1946fe..ccd4f7f02 100644 --- a/README.md +++ b/README.md @@ -679,7 +679,7 @@ information on identifiers and how they are used in replication. ) The identifier is setup on the builder as follows. ```java -TcpConfig tcpConfig = ... +TcpTransportAndNetworkConfig tcpConfig = ... map = ChronicleMapBuilder .of(Integer.class, CharSequence.class) .replication(identifier, tcpConfig) @@ -731,12 +731,12 @@ require 6 connections. In our case we are only using 2 maps, this is how we connected map1 to map 2. ``` -TcpConfig.of(8076, new InetSocketAddress("localhost", 8077)) +TcpTransportAndNetworkConfig.forSendingNode(8076, new InetSocketAddress("localhost", 8077)) .heartBeatInterval(1, SECONDS); ``` you could have put this instruction on map2 instead, like this ``` -TcpConfig.of(8077, new InetSocketAddress("localhost", 8076)) +TcpTransportAndNetworkConfig.forSendingNode(8077, new InetSocketAddress("localhost", 8076)) .heartBeatInterval(1, SECONDS); ``` even though data flows from map1 to map2 and map2 to map1 it doesn't matter which way you connected @@ -764,15 +764,16 @@ InetSocketAddress inetSocketAddress3 = new InetSocketAddress(hostServer3, server // this is to go on server 1 -TcpConfig tcpConfigServer1 = TcpConfig.of(serverPort1); +TcpTransportAndNetworkConfig tcpConfigServer1 = + TcpTransportAndNetworkConfig.forReceivingOnlyNode(serverPort1); // this is to go on server 2 -TcpConfig tcpConfigServer2 = TcpConfig.of(serverPort2, - inetSocketAddress1); +TcpTransportAndNetworkConfig tcpConfigServer2 = TcpTransportAndNetworkConfig + .forSendingNode(serverPort2, inetSocketAddress1); // this is to go on server 3 -TcpConfig tcpConfigServer3 = TcpConfig.of(serverPort3, - inetSocketAddress1,inetSocketAddress2); +TcpTransportAndNetworkConfig tcpConfigServer3 = TcpTransportAndNetworkConfig + .forSendingNode(serverPort3, inetSocketAddress1, inetSocketAddress2); ``` ### Heart Beat Interval @@ -811,7 +812,7 @@ byte identifier= 2; ReplicationHub replicationHub = ReplicationHub.builder() .maxEntrySize(maxEntrySize) .tcpTransportAndNetwork(tcpConfig) - .create(identifier); + .createWithId(identifier); ``` In this example above the channel is given the identifier of 2 @@ -829,7 +830,7 @@ byte identifier = 1; ReplicationHub replicationHub = ReplicationHub.builder() .maxEntrySize(maxEntrySize) .tcpTransportAndNetwork(tcpConfig) - .create(identifier); + .createWithId(identifier); ``` Attaching ChannelProvider replication to the map: @@ -868,14 +869,14 @@ unique for each map you have. byte identifier = (byte) 1; - TcpConfig tcpConfig = TcpConfig + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); hubOnServer1 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) - .create(identifier); + .createWithId(identifier); // this demotes favoriteColour short channel1 = (short) 1; @@ -903,12 +904,12 @@ unique for each map you have. byte identifier = (byte) 2; - TcpConfig tcpConfig = - TcpConfig.of(8087).heartBeatInterval(1, SECONDS); + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .forReceivingOnlyNode(8087).heartBeatInterval(1, SECONDS); hubOnServer2 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) - .create(identifier); + .createWithId(identifier); // this demotes favoriteColour @@ -980,7 +981,7 @@ final ChronicleMap statelessMap; { ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.of(8076)) + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8076)) .create(); serverMap.put(10, "EXAMPLE-10"); @@ -1007,7 +1008,8 @@ being used by another application. In this example we choose the port 8076 ``` java -.replication((byte) 2, TcpConfig.of(8076)) // sets the server to run on localhost:8076 +// sets the server to run on localhost:8076 +.replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8076)) ``` On the "stateless client" we connect to the server via TCP/IP on localhost:8076 : @@ -1074,7 +1076,7 @@ For the very best performance you should also set these properties on the server ``` java ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.of(8076)) + .replication((byte) 2, TcpTransportAndNetworkConfig.forReceivingOnlyNode(8076)) .putReturnsNull(true) .removeReturnsNull(true) .create(); @@ -1246,7 +1248,8 @@ public class YourClass { // we connect the maps via a TCP/IP socket connection on port 8077 - TcpConfig tcpConfig = TcpConfig.of(8076, new InetSocketAddress("localhost", 8077)) + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .forSendingNode(8076, new InetSocketAddress("localhost", 8077)) .heartBeatInterval(1L, TimeUnit.SECONDS); ChronicleMapBuilder map1Builder = ChronicleMapBuilder.of(Integer.class, CharSequence.class) @@ -1259,7 +1262,8 @@ public class YourClass { // ---------- SERVER2 on the same server as ---------- { - TcpConfig tcpConfig = TcpConfig.of(8077) + TcpTransportAndNetworkConfig tcpConfig = + TcpTransportAndNetworkConfig.forReceivingOnlyNode(8077) .heartBeatInterval(1L, TimeUnit.SECONDS); ChronicleMapBuilder map2Builder = ChronicleMapBuilder.of(Integer.class, CharSequence.class) @@ -1326,23 +1330,24 @@ public class YourClass { // we connect the maps via a TCP socket connection on port 8077 - TcpConfig tcpConfig = TcpConfig.of(8076, new InetSocketAddress("localhost", 8077)) + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .forSendingNode(8076, new InetSocketAddress("localhost", 8077)) .heartBeatInterval(1L, TimeUnit.SECONDS) // a maximum of 1024 bits per millisecond .throttlingConfig(ThrottlingConfig.throttle(1024, TimeUnit.MILLISECONDS)); - UdpConfig udpConfig = UdpConfig - .simple(Inet4Address.getByName("255.255.255.255"), udpPort); + UdpTransportConfig udpConfig = UdpTransportConfig + .of(Inet4Address.getByName("255.255.255.255"), udpPort); ChronicleMapBuilder map1Builder = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000L) - .replication(SimpleReplication.builder() + .replication(SingleChronicleHashReplication.builder() .tcpTransportAndNetwork(tcpConfig) .udpTransport(udpConfig) - .create((byte) 1)); + .createWithId((byte) 1)); map1 = map1Builder.create(); @@ -1350,20 +1355,21 @@ public class YourClass { // ---------- SERVER2 2 on the same server as ---------- { - TcpConfig tcpConfig = TcpConfig.of(8077) + TcpTransportAndNetworkConfig tcpConfig = + TcpTransportAndNetworkConfig.forReceivingOnlyNode(8077) .heartBeatInterval(1L, TimeUnit.SECONDS) .throttlingConfig(ThrottlingConfig.throttle(1024, TimeUnit.MILLISECONDS)); - UdpConfig udpConfig = UdpConfig - .simple(Inet4Address.getByName("255.255.255.255"), udpPort); + UdpTransportConfig udpConfig = UdpTransportConfig + .of(Inet4Address.getByName("255.255.255.255"), udpPort); ChronicleMapBuilder map2Builder = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000L) - .replication(SimpleReplication.builder() + .replication(SingleChronicleHashReplication.builder() .tcpTransportAndNetwork(tcpConfig) .udpTransport(udpConfig) - .create((byte) 2)); + .createWithId((byte) 2)); map2 = map2Builder.create(); diff --git a/pom.xml b/pom.xml old mode 100755 new mode 100644 index 50e60012e..f1c471919 --- a/pom.xml +++ b/pom.xml @@ -28,7 +28,7 @@ 4.0.0 chronicle-map - 2.0.8a-SNAPSHOT + 2.0.9a-SNAPSHOT OpenHFT/Chronicle-Map Chronicle-Map diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java index 4f86d0980..924b95a5f 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java @@ -18,9 +18,10 @@ 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.*; import net.openhft.chronicle.set.ChronicleSet; import net.openhft.chronicle.set.ChronicleSetBuilder; @@ -212,7 +213,7 @@ public interface ChronicleHashBuilderDefault lock time out is 2 seconds. @@ -236,7 +237,7 @@ public interface ChronicleHashBuilder keyMarshaller); + /** + * Configures the marshallers, used to serialize/deserialize keys to/from off-heap + * memory in hash containers, created by this builder. See + * the section about + * serialization in ChronicleMap manual for more information. + * + *

Configuring marshalling this way results to a little bit more compact in-memory layout + * of the map, comparing to a single interface configuration: + * {@link #keyMarshaller(BytesMarshaller)}. + * + *

Passing {@link BytesInterop} (which is a subinterface of {@link BytesWriter}) as the + * first argument is supported, and even more advantageous from performance perspective. + * + * @param keyWriter the new key object → {@link Bytes} writer (interop) strategy + * @param keyReader the new {@link Bytes} → key object reader strategy + * @return this builder back + * @see #keyMarshaller(BytesMarshaller) + */ + B keyMarshallers(@NotNull BytesWriter keyWriter, @NotNull BytesReader keyReader); + + /** + * Configures the marshaller used to serialize actual key sizes to off-heap memory + * in hash containers, created by this builder. + * + *

Default key size marshaller is so-called {@linkplain SizeMarshallers#stopBit() + * stop bit encoding marshalling}. If {@linkplain #constantKeySizeBySample(Object) constant key + * size} is configured, or defaulted if the key type is always constant and {@code + * ChronicleHashBuilder} implementation knows about it, this configuration takes no effect, + * because a special {@link SizeMarshaller} implementation, which doesn't actually do any + * marshalling, and just returns the known constant size on {@link SizeMarshaller#readSize( + * Bytes)} calls, is used instead of any {@code SizeMarshaller} configured using this method. + * + * @param keySizeMarshaller the new marshaller, used to serialize actual key sizes to off-heap + * memory + * @return this builder back + */ + B keySizeMarshaller(@NotNull SizeMarshaller keySizeMarshaller); + /** * Configures factory which is used to create a new key instance, if key class is either * {@link Byteable}, {@link BytesMarshallable} or {@link Externalizable} subclass in maps, @@ -336,26 +376,26 @@ 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() - * .tcpTransportAndNetwork(tcpTransportAndNetwork).create(identifier))}. + * .tcpTransportAndNetwork(tcpTransportAndNetwork).createWithId(identifier))}. * * @param identifier the network-wide identifier of the containers, created by this builder * @param tcpTransportAndNetwork configuration of tcp connection and network * @return this builder back - * @see #replication(SimpleReplication) - * @see ChronicleHashInstanceConfig#replicated(byte, TcpConfig) + * @see #replication(SingleChronicleHashReplication) + * @see ChronicleHashInstanceConfig#replicated(byte, TcpTransportAndNetworkConfig) */ - B replication(byte identifier, TcpConfig tcpTransportAndNetwork); + B replication(byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork); StatelessClientConfig 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/exceptions/TimeoutRuntimeException.java b/src/main/java/net/openhft/chronicle/hash/RemoteCallTimeoutException.java similarity index 76% rename from src/main/java/net/openhft/chronicle/hash/exceptions/TimeoutRuntimeException.java rename to src/main/java/net/openhft/chronicle/hash/RemoteCallTimeoutException.java index 477e5b56e..eedb9a276 100644 --- a/src/main/java/net/openhft/chronicle/hash/exceptions/TimeoutRuntimeException.java +++ b/src/main/java/net/openhft/chronicle/hash/RemoteCallTimeoutException.java @@ -16,19 +16,20 @@ * limitations under the License. */ -package net.openhft.chronicle.hash.exceptions; +package net.openhft.chronicle.hash; /** + * TODO mention in docs * @author Rob Austin. */ -public class TimeoutRuntimeException extends RuntimeException { +public class RemoteCallTimeoutException extends RuntimeException { - public TimeoutRuntimeException(String s) { + public RemoteCallTimeoutException(String s) { super(s); } - public TimeoutRuntimeException() { + public RemoteCallTimeoutException() { super(); } } 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 ac269165b..078df5519 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java @@ -19,14 +19,14 @@ package net.openhft.chronicle.hash.replication; import org.jetbrains.annotations.Nullable; class AbstractReplication { - protected final byte localIdentifier; - protected final UdpConfig udpConfig; - protected final TcpConfig tcpConfig; + private final byte localIdentifier; + private final TcpTransportAndNetworkConfig tcpConfig; + private final UdpTransportConfig udpConfig; AbstractReplication(byte localIdentifier, Builder builder) { - udpConfig = builder.udpConfig; this.localIdentifier = localIdentifier; tcpConfig = builder.tcpConfig; + udpConfig = builder.udpConfig; } public byte identifier() { @@ -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 { - protected UdpConfig udpConfig = null; - protected TcpConfig tcpConfig = 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 6c833c8f1..eb137ac1b 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java @@ -62,6 +62,16 @@ public final class ReplicationHub extends AbstractReplication { private Builder() {} + @Override + public Builder tcpTransportAndNetwork(TcpTransportAndNetworkConfig tcpConfig) { + return super.tcpTransportAndNetwork(tcpConfig); + } + + @Override + public Builder udpTransport(UdpTransportConfig udpConfig) { + return super.udpTransport(udpConfig); + } + public Builder maxEntrySize(int maxEntrySize) { this.maxEntrySize = maxEntrySize; return this; @@ -72,7 +82,7 @@ public final class ReplicationHub extends AbstractReplication { return this; } - public ReplicationHub create(byte identifier) throws IOException { + public ReplicationHub createWithId(byte identifier) throws IOException { check(); return new ReplicationHub(identifier, this); } 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 60% 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 0f716083f..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,23 +16,32 @@ 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(TcpTransportAndNetworkConfig tcpConfig) { + return super.tcpTransportAndNetwork(tcpConfig); + } + + @Override + 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 72% 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 66332f21e..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 unknownTopology(serverPort, endpoints); + return forUnknownTopology(serverPort, endpoints); } - public static TcpConfig forReceivingOnlyNode(int serverPort) { - return unknownTopology(serverPort, Collections.emptyList()); + public static TcpTransportAndNetworkConfig forReceivingOnlyNode(int serverPort) { + return forUnknownTopology(serverPort, Collections.emptyList()); } - public static TcpConfig unknownTopology(int serverPort, - Collection endpoints) { + 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 7b77c10ac..f555e49b5 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java @@ -20,6 +20,9 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.*; import net.openhft.chronicle.hash.replication.*; +import net.openhft.chronicle.hash.serialization.BytesReader; +import net.openhft.chronicle.hash.serialization.BytesWriter; +import net.openhft.chronicle.hash.serialization.SizeMarshaller; import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter; import net.openhft.chronicle.hash.serialization.internal.MetaProvider; @@ -136,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); @@ -435,10 +438,6 @@ public abstract class AbstractChronicleMapBuilder (long) minSegments() << 15; - } - public B lockTimeOut(long lockTimeOut, TimeUnit unit) { this.lockTimeOut = lockTimeOut; lockTimeOutUnit = unit; @@ -606,7 +605,20 @@ public abstract class AbstractChronicleMapBuilder keyMarshaller) { - keyBuilder.marshaller(keyMarshaller, null); + keyBuilder.marshaller(keyMarshaller); + return self(); + } + + @Override + public B keyMarshallers(@NotNull BytesWriter keyWriter, @NotNull BytesReader keyReader) { + keyBuilder.writer(keyWriter); + keyBuilder.reader(keyReader); + return self(); + } + + @Override + public B keySizeMarshaller(@NotNull SizeMarshaller keySizeMarshaller) { + keyBuilder.sizeMarshaller(keySizeMarshaller); return self(); } @@ -728,56 +740,59 @@ 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); + // clone() to make this builder instance thread-safe, because createWithFile() method + // computes some state based on configurations, but doesn't synchronize on configuration + // changes. + return clone().createWithFile(file, singleHashReplication, null); } @Override public ChronicleMap create() throws IOException { - return createWithoutFile(simpleReplication, null); + // clone() to make this builder instance thread-safe, because createWithoutFile() method + // computes some state based on configurations, but doesn't synchronize on configuration + // changes. + return clone().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) { - FileInputStream fis = new FileInputStream(file); - ObjectInputStream ois = new ObjectInputStream(fis); - try { + try (FileInputStream fis = new FileInputStream(file); + ObjectInputStream ois = new ObjectInputStream(fis)) { VanillaChronicleMap map = (VanillaChronicleMap) 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 { - ois.close(); } } if (file.createNewFile() || file.length() == 0) { @@ -793,29 +808,27 @@ public abstract class AbstractChronicleMapBuilder map = newMap(simpleReplication, channel); + VanillaChronicleMap map = newMap(singleHashReplication, channel); - FileOutputStream fos = new FileOutputStream(file); - ObjectOutputStream oos = new ObjectOutputStream(fos); - try { + try (FileOutputStream fos = new FileOutputStream(file); + ObjectOutputStream oos = new ObjectOutputStream(fos)) { oos.writeObject(map); oos.flush(); map.headerSize = roundUpMapHeaderSize(fos.getChannel().position()); map.createMappedStoreAndSegments(file); - } finally { - oos.close(); } - return establishReplication(map, simpleReplication, channel); + return establishReplication(map, singleHashReplication, channel); } /** - * Only for testing. + * Only for testing. Shouldn't be called concurrently on a single builder instance. * * @param identifier id * @return map */ ChronicleMap createReplicated(byte identifier) throws IOException { + preMapConstruction(true); VanillaChronicleMap map = new ReplicatedChronicleMap, V, Object, MetaBytesWriter>(this, identifier); @@ -825,13 +838,13 @@ public abstract class AbstractChronicleMapBuilder createWithoutFile( - SimpleReplication simpleReplication, ReplicationChannel channel) throws IOException { - VanillaChronicleMap map = newMap(simpleReplication, channel); + ChronicleMap createWithoutFile( + 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) @@ -843,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(); } @@ -876,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/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 65828fc97..c501004d6 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -17,8 +17,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHashBuilder; -import net.openhft.chronicle.hash.serialization.BytesReader; -import net.openhft.chronicle.hash.serialization.BytesWriter; +import net.openhft.chronicle.hash.serialization.*; import net.openhft.lang.io.Bytes; import net.openhft.lang.io.serialization.BytesMarshallable; import net.openhft.lang.io.serialization.BytesMarshaller; @@ -218,16 +217,68 @@ public final class ChronicleMapBuilder return super.defaultValueProvider(defaultValueProvider); } + /** + * Configures the {@code BytesMarshaller} used to serialize/deserialize values to/from off-heap + * memory in maps, created by this builder. See + * the + * section about serialization in ChronicleMap manual for more information. + * + * @param valueMarshaller the marshaller used to serialize values + * @return this builder back + * @see #valueMarshallers(BytesWriter, BytesReader) + * @see OffHeapUpdatableChronicleMapBuilder + * @see #objectSerializer(ObjectSerializer) + */ public ChronicleMapBuilder valueMarshaller( @NotNull BytesMarshaller valueMarshaller) { - valueBuilder.marshaller(valueMarshaller, null); + valueBuilder.marshaller(valueMarshaller); return this; } + /** + * Configures the marshallers, used to serialize/deserialize values to/from off-heap + * memory in maps, created by this builder. See + * the section about + * serialization in ChronicleMap manual for more information. + * + *

Configuring marshalling this way results to a little bit more compact in-memory layout + * of the map, comparing to a single interface configuration: + * {@link #valueMarshaller(BytesMarshaller)}. + * + *

Passing {@link BytesInterop} instead of plain {@link BytesWriter} is, of cause, possible, + * but currently pointless for values. + * + * @param valueWriter the new value object → {@link Bytes} writer (interop) strategy + * @param valueReader the new {@link Bytes} → value object reader strategy + * @return this builder back + * @see #valueMarshaller(BytesMarshaller) + */ public ChronicleMapBuilder valueMarshallers( - BytesWriter valueWriter, BytesReader valueReader) { + @NotNull BytesWriter valueWriter, @NotNull BytesReader valueReader) { valueBuilder.writer(valueWriter); - valueBuilder.reader(valueReader, valueBuilder.factory()); + valueBuilder.reader(valueReader); + return this; + } + + /** + * Configures the marshaller used to serialize actual value sizes to off-heap memory in maps, + * created by this builder. + * + *

Default value size marshaller is so-called {@linkplain SizeMarshallers#stopBit() + * stop bit encoding marshalling}. If {@linkplain #constantValueSizeBySample(Object) constant + * value size} is configured, or defaulted if the value type is always constant and {@code + * ChronicleHashBuilder} implementation knows about it, this configuration takes no effect, + * because a special {@link SizeMarshaller} implementation, which doesn't actually do any + * marshalling, and just returns the known constant size on {@link SizeMarshaller#readSize( + * Bytes)} calls, is used instead of any {@code SizeMarshaller} configured using this method. + * + * @param valueSizeMarshaller the new marshaller, used to serialize actual value sizes to + * off-heap memory + * @return this builder back + */ + public ChronicleMapBuilder valueSizeMarshaller( + @NotNull SizeMarshaller valueSizeMarshaller) { + valueBuilder.sizeMarshaller(valueSizeMarshaller); return this; } } diff --git a/src/main/java/net/openhft/chronicle/hash/exceptions/IORuntimeException.java b/src/main/java/net/openhft/chronicle/map/IORuntimeException.java similarity index 88% rename from src/main/java/net/openhft/chronicle/hash/exceptions/IORuntimeException.java rename to src/main/java/net/openhft/chronicle/map/IORuntimeException.java index 48fafd020..e58603bd8 100644 --- a/src/main/java/net/openhft/chronicle/hash/exceptions/IORuntimeException.java +++ b/src/main/java/net/openhft/chronicle/map/IORuntimeException.java @@ -16,14 +16,14 @@ * limitations under the License. */ -package net.openhft.chronicle.hash.exceptions; +package net.openhft.chronicle.map; import java.io.IOException; /** * @author Rob Austin. */ -public class IORuntimeException extends RuntimeException{ +class IORuntimeException extends RuntimeException{ public IORuntimeException(IOException e) { super(e); 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/MapEventListener.java b/src/main/java/net/openhft/chronicle/map/MapEventListener.java index a4dbe53f9..9c0cc1fae 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEventListener.java +++ b/src/main/java/net/openhft/chronicle/map/MapEventListener.java @@ -20,30 +20,28 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHashBuilder; import net.openhft.lang.io.Bytes; +import org.jetbrains.annotations.Nullable; import java.io.Serializable; /** - * Contains methods which are called when {@link ChronicleMap} key events occur. Typical use cases: - *

    - *
  • Map data backup / replication.
    • - *
  • Logging, monitoring, debugging.
    • - *
+ * Contains methods which are called when {@link ChronicleMap} key events occur. Typical use cases:
    + *
  • Map data backup / replication.
  • Logging, monitoring, debugging.
* - *

This is an adapter class - all methods have default implementations as no-ops. Extend this - * class and override only methods corresponding the events you are interested in. + *

This is an adapter class - all methods have default implementations as no-ops. Extend this class and + * override only methods corresponding the events you are interested in. * - *

To configure {@code MapEventListener} for {@code ChronicleMap}, use - * {@link AbstractChronicleMapBuilder#eventListener(MapEventListener)} method. + *

To configure {@code MapEventListener} for {@code ChronicleMap}, use {@link + * AbstractChronicleMapBuilder#eventListener(MapEventListener)} method. * - *

{@link MapEventListeners} uninstantiable class contains several logging implementations and - * the default {@linkplain MapEventListeners#nop() no-op implementation}. + *

{@link MapEventListeners} uninstantiable class contains several logging implementations and the default + * {@linkplain MapEventListeners#nop() no-op implementation}. * - *

All these calls are synchronous while a {@code ChronicleMap} lock is held so make them - * as quick as possible. + *

All these calls are synchronous while a {@code ChronicleMap} lock is held so make them as quick as + * possible. * - *

The {@code entry} passed to the methods is {@code Bytes} instance positioned at meta data - * area. See {@link ChronicleHashBuilder#metaDataBytes(int)} for more information. + *

The {@code entry} passed to the methods is {@code Bytes} instance positioned at meta data area. See + * {@link ChronicleHashBuilder#metaDataBytes(int)} for more information. * * @param key type of the maps, trackable by this event listener * @param value type of the maps, trackable by this event listener @@ -56,14 +54,12 @@ public abstract class MapEventListener> private static final long serialVersionUID = 0L; /** - * This method is called if the key is found in the map during {@link ChronicleMap#get get}, - * {@link ChronicleMap#getUsing getUsing} or {@link ChronicleMap#acquireUsing acquireUsing} - * method call. + * This method is called if the key is found in the map during {@link ChronicleMap#get get}, {@link + * ChronicleMap#getUsing getUsing} or {@link ChronicleMap#acquireUsing acquireUsing} method call. * * @param map the accessed map * @param entry bytes of the entry - * @param metaDataBytes length of {@link ChronicleHashBuilder#metaDataBytes(int) meta data} - * for this map + * @param metaDataBytes length of {@link ChronicleHashBuilder#metaDataBytes(int) meta data} for this map * @param key the key looked up * @param value the value found for the key */ @@ -71,32 +67,32 @@ public abstract class MapEventListener> // do nothing } - void onPut(M map, Bytes entry, int metaDataBytes, boolean added, K key, V value, + void onPut(M map, Bytes entry, int metaDataBytes, boolean added, K key, V replacedValue, V value, long pos, SharedSegment segment) { - onPut(map, entry, metaDataBytes, added, key, value); + onPut(map, entry, metaDataBytes, added, key, value, replacedValue); } /** - * This method is called whenever a new value is put for the key in the map during calls of such - * methods as {@link ChronicleMap#put put}, {@link ChronicleMap#putIfAbsent putIfAbsent}, - * {@link ChronicleMap#replace(Object, Object, Object) replace}, etc. When a new value is - * {@linkplain AbstractChronicleMapBuilder#defaultValue(Object) default} for the map or obtained - * during {@link ChronicleMap#acquireUsing acquireUsing} call is put for the key, this method - * is called as well. + * This method is called whenever a new value is put for the key in the map during calls of such methods + * as {@link ChronicleMap#put put}, {@link ChronicleMap#putIfAbsent putIfAbsent}, {@link + * ChronicleMap#replace(Object, Object, Object) replace}, etc. When a new value is {@linkplain + * AbstractChronicleMapBuilder#defaultValue(Object) default} for the map or obtained during {@link + * ChronicleMap#acquireUsing acquireUsing} call is put for the key, this method is called as well. * *

This method is called when put is already happened. * * @param map the accessed map * @param entry bytes of the entry (with the value already written to) - * @param metaDataBytes length of {@link ChronicleHashBuilder#metaDataBytes(int) meta data} - * for this map - * @param added {@code true} is the key was absent in the map before current - * {@code ChronicleMap} method call, led to putting a value and inherently - * calling this listener method + * @param metaDataBytes length of {@link net.openhft.chronicle.hash.ChronicleHashBuilder#metaDataBytes(int) + * meta data} for this map + * @param added {@code true} is the key was absent in the map before current {@code ChronicleMap} + * method call, led to putting a value and inherently calling this listener method * @param key the key the given value is put for * @param value the value which is now associated with the given key + * @param replacedValue the value which was replaced by {@code value}, is NULL if {@code added} is true */ - public void onPut(M map, Bytes entry, int metaDataBytes, boolean added, K key, V value) { + public void onPut(M map, Bytes entry, int metaDataBytes, boolean added, K key, V value, + @Nullable V replacedValue) { // do nothing } @@ -106,13 +102,12 @@ public abstract class MapEventListener> } /** - * This is called when an entry is removed. Misses, i. e. when {@code map.remove(key)} - * is called, but key is already absent in the map, are not notified. + * This is called when an entry is removed. Misses, i. e. when {@code map.remove(key)} is called, but key + * is already absent in the map, are not notified. * * @param map accessed * @param entry bytes of the entry just removed, but any data is not yet erased - * @param metaDataBytes length of {@link ChronicleHashBuilder#metaDataBytes(int) meta data} - * for this map + * @param metaDataBytes length of {@link ChronicleHashBuilder#metaDataBytes(int) meta data} for this map * @param key the key removed from the map * @param value the value which was associated with the given key */ diff --git a/src/main/java/net/openhft/chronicle/map/MapEventListeners.java b/src/main/java/net/openhft/chronicle/map/MapEventListeners.java index 4ddb4d251..754c54f9c 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEventListeners.java +++ b/src/main/java/net/openhft/chronicle/map/MapEventListeners.java @@ -68,7 +68,7 @@ public final class MapEventListeners { @Override public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, - Object key, Object value, long pos, SharedSegment segment) { + Object key, Object replacedValue, Object value, long pos, SharedSegment segment) { logOperation(map, entry, metaDataBytes, added ? " +put " : " put "); } @@ -90,7 +90,7 @@ public final class MapEventListeners { @Override public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, - Object key, Object value, long pos, SharedSegment segment) { + Object key, Object replacedValue, Object value, long pos, SharedSegment segment) { LOGGER.info("{} put {} => {}", map.file(), key, value); } diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index 97eca2983..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,8 +221,8 @@ public class NodeDiscovery { // add our identifier and host:port to the list of known identifiers knownNodes.add(ourAddressAndPort, identifier); - final TcpConfig tcpConfig = TcpConfig - .unknownTopology(ourAddressAndPort.getPort(), + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .forUnknownTopology(ourAddressAndPort.getPort(), toInetSocketCollection(knownHostPorts)) .heartBeatInterval(1, SECONDS) .nonUniqueIdentifierListener(identifierListener); @@ -259,8 +259,8 @@ public class NodeDiscovery { /** - * bitwise OR's the two bit sets or put another way, merges the source bitset into the - * destination bitset and returns the destination + * bitwise OR's the two bit sets or put another way, merges the source bitset into the destination bitset + * and returns the destination * * @param source to or * @param destination to or @@ -326,8 +326,8 @@ public class NodeDiscovery { } /** - * Broadcast the nodes host ports and identifiers over UDP, to make it easy to join a grid of remote - * nodes just by name, this functionality requires UDP + * Broadcast the nodes host ports and identifiers over UDP, to make it easy to join a grid of remote nodes + * just by name, this functionality requires UDP * * @author Rob Austin. */ @@ -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 { @@ -738,8 +738,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { } /** - * this is used to tell nodes that are connecting to us which host and ports are in our grid, - * along with all the identifiers. + * this is used to tell nodes that are connecting to us which host and ports are in our grid, along with + * all the identifiers. */ @Override public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) { @@ -819,7 +819,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { if (bootstrap.addressAndPort().equals(this.ourAddressAndPort)) return; - LOG.info("Received Bootstrap from " + bootstrap); + if (LOG.isDebugEnabled()) + LOG.debug("Received Bootstrap from " + bootstrap); proposedIdentifiersWithHost.put(bootstrap.addressAndPort, bootstrap); @@ -839,7 +840,7 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { if (sourceAddressAndPort.equals(ourAddressAndPort)) return; - LOG.info("Received Proposal"); + LOG.debug("Received Proposal"); this.remoteNode.readMarshallable(in); this.proposedIdentifiersWithHost.readMarshallable(in); @@ -929,8 +930,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { /** - * sends a bootstrap message to the other nodes in the grid, the bootstrap message contains the - * host:port and perhaps even proposed identifier of the node that sent it. + * sends a bootstrap message to the other nodes in the grid, the bootstrap message contains the host:port + * and perhaps even proposed identifier of the node that sent it. * * @param proposedNodes */ diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 46c7586cc..63953105b 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -199,13 +199,20 @@ final class ReplicatedChronicleMap, * as this will trigger the modification iterators to send out another replication event. So instead with * call this method iterators * - * @return + * @return the event listener added by the user */ - MapEventListener> eventListenerWithoutModIterators() { + MapEventListener> attachedEventListener() { return eventListener; } + /** + * @return true if the user has added an event listerner + */ + boolean hasAttachedEventListeners() { + return eventListener != null; + } + @Override public V put(K key, V value) { return put0(key, value, true, localIdentifier, timeProvider.currentTimeMillis()); @@ -433,7 +440,8 @@ final class ReplicatedChronicleMap, // we store the isDeleted flag in the identifier // ( when the identifier is negative it is deleted ) - destination.writeByte(isDeleted ? -identifier : identifier); + destination.writeByte(identifier); + destination.writeBoolean(isDeleted); // write the key entry.position(keyPosition); @@ -481,14 +489,12 @@ final class ReplicatedChronicleMap, final long valueSize = valueSizeMarshaller.readSize(source); final long timeStamp = source.readStopBit(); final byte id = source.readByte(); - final byte remoteIdentifier; - final boolean isDeleted; + final boolean isDeleted = source.readBoolean(); - if (id < 0) { - isDeleted = true; - remoteIdentifier = (byte) -id; - } else if (id != 0) { - isDeleted = false; + final byte remoteIdentifier; + + if (id != 0) { + // isDeleted = false; remoteIdentifier = id; } else { throw new IllegalStateException("identifier can't be 0"); @@ -513,8 +519,7 @@ final class ReplicatedChronicleMap, if (isDeleted) { if (debugEnabled) { - LOG.debug( - "READING FROM SOURCE - into local-id={}, remote={}, remove(key={})", + LOG.debug("READING FROM SOURCE - into local-id={}, remote={}, remove(key={})", localIdentifier, remoteIdentifier, source.toString().trim() ); } @@ -635,7 +640,7 @@ final class ReplicatedChronicleMap, throw new IllegalStateException("ENTRY WRITE_BUFFER_SIZE TOO LARGE : Replicated " + "ChronicleMap's" + " are restricted to an " + - "entry size of " + Integer.MAX_VALUE + ", " + + "entry size of " + Integer.MAX_VALUE + ", " + "your entry size=" + result); return result; @@ -678,11 +683,14 @@ final class ReplicatedChronicleMap, bytes.skip(ADDITIONAL_ENTRY_BYTES); } + /** * called from a remote node as part of replication */ private void remoteRemove(Bytes keyBytes, long hash2, final long timestamp, final byte identifier) { + + writeLock(); try { long keySize = keyBytes.remaining(); @@ -694,6 +702,8 @@ final class ReplicatedChronicleMap, continue; // key is found + + long keyPosition = entry.position(); entry.skip(keySize); long timeStampPos = entry.position(); @@ -702,7 +712,7 @@ final class ReplicatedChronicleMap, return; } - // skip the is deleted flag + // read the is deleted flag boolean wasDeleted = entry.readBoolean(); if (!wasDeleted) { @@ -714,9 +724,35 @@ final class ReplicatedChronicleMap, entry.writeLong(timestamp); assert identifier > 0; entry.writeByte(identifier); + // was deleted entry.writeBoolean(true); + + long valuePosition = entry.position(); + + if (hasAttachedEventListeners() && !wasDeleted) { + entry.position(keyPosition); + K key = null; + ThreadLocalCopies keyCopies = null; + + if (hasAttachedEventListeners()) { + keyCopies = keyInteropProvider.getCopies(null); + BytesReader keyReader = keyReaderProvider.get(keyCopies, originalKeyReader); + key = keyReader.read(entry, keySize); + } + + entry.position(valuePosition); + ThreadLocalCopies valueCopies = valueReaderProvider.getCopies(keyCopies); + BytesReader valueReader = valueReaderProvider.get(valueCopies, + originalValueReader); + V value = valueReader.read(entry, readValueSize(entry)); + + notifyRemoteRemove(offset, key, value, pos); + } + + return; } + // key is not found if (LOG.isDebugEnabled()) LOG.debug("Segment.remoteRemove() : key=" + keyBytes.toString().trim() + @@ -727,8 +763,20 @@ final class ReplicatedChronicleMap, } + void notifyRemoteRemove(long offset, K key, V value, final long pos) { + if (eventListener() != MapEventListeners.NOP) { + MultiStoreBytes b = acquireTmpBytes(); + b.storePositionAndSize(bytes, offset, entrySize); + attachedEventListener().onRemove(ReplicatedChronicleMap.this, b, metaDataBytes, + key, value, pos, this); + } + } + + void notifyRemotePut(long offset, boolean added, final long pos, long keySize, final Bytes inBytes, - final long valueSize, final long valuePosition, final long keyPosition) { + final long valueSize, final long valuePosition, + final long keyPosition, final V replacedValue, BytesReader valueReader, + BytesReader keyReader) { if (eventListener() != MapEventListeners.NOP) { final MultiStoreBytes b = acquireTmpBytes(); b.storePositionAndSize(bytes, offset, entrySize); @@ -736,19 +784,16 @@ final class ReplicatedChronicleMap, inBytes.limit(inBytes.capacity()); // deserialize the key - final ThreadLocalCopies copies = keyReaderProvider.getCopies(null); inBytes.position(keyPosition); - final K key = keyReaderProvider.get(copies, originalKeyReader).read(inBytes, keySize); + final K key = keyReader.read(inBytes, keySize); // deserialize the value - final ThreadLocalCopies copiesV = valueReaderProvider.getCopies(copies); - inBytes.limit(inBytes.capacity()); inBytes.position(valuePosition); - final V value = valueReaderProvider.get(copiesV, originalValueReader).read(inBytes, valueSize); + final V value = valueReader.read(inBytes, valueSize); - eventListener().onPut(ReplicatedChronicleMap.this, b, metaDataBytes, - added, key, value, pos, this); + attachedEventListener().onPut(ReplicatedChronicleMap.this, b, metaDataBytes, + added, key, replacedValue, value, pos, this); } } @@ -759,10 +804,25 @@ final class ReplicatedChronicleMap, private void remotePut(@NotNull final Bytes inBytes, long hash2, final byte identifier, final long timestamp, long valuePos, long valueLimit) { + + + BytesReader valueReader = null; + BytesReader keyReader = null; + + if (hasAttachedEventListeners()) { + ThreadLocalCopies keyCopies = keyInteropProvider.getCopies(null); + keyReader = keyReaderProvider.get(keyCopies, originalKeyReader); + + ThreadLocalCopies valueCopies = valueReaderProvider.getCopies(keyCopies); + valueReader = valueReaderProvider.get(valueCopies, originalValueReader); + } + writeLock(); try { // inBytes position and limit correspond to the key final long keySize = inBytes.remaining(); + V replacedValue = null; + hashLookupLiveAndDeleted.startSearch(hash2); for (long pos; (pos = hashLookupLiveAndDeleted.nextPos()) >= 0L; ) { @@ -774,7 +834,6 @@ final class ReplicatedChronicleMap, continue; // key is found - entry.skip(keySize); final long timeStampPos = entry.positionAddr(); @@ -793,13 +852,17 @@ final class ReplicatedChronicleMap, entry.writeBoolean(false); long valueSizePos = entry.position(); - long valueSize = readValueSize(entry); - long entryEndAddr = entry.positionAddr() + valueSize; + long replacedValueSize = readValueSize(entry); + long entryEndAddr = entry.positionAddr() + replacedValueSize; // write the value inBytes.limit(valueLimit); inBytes.position(valuePos); + // if we have event listeners attached we have to read back the old value + if (hasAttachedEventListeners()) + replacedValue = valueReader.read(entry, replacedValueSize, null); + putValue(pos, offset, entry, valueSizePos, entryEndAddr, null, null, inBytes, hashLookupLiveAndDeleted); @@ -809,10 +872,23 @@ final class ReplicatedChronicleMap, incrementSize(); } - // this is just like notifyPut, but it won't trigger the modification iterators - notifyRemotePut(offset, false, posFromOffset(offset), keySize, inBytes, valueLimit - -valuePos, - valuePos, keyPosition); + if (hasAttachedEventListeners()) { + + // this is just like notifyPut, but it won't trigger the modification iterators + notifyRemotePut(offset, + false, + posFromOffset(offset), + keySize, + inBytes, + valueLimit - valuePos, + valuePos, + keyPosition, + replacedValue, + valueReader, + keyReader + ); + } + return; } @@ -848,7 +924,8 @@ final class ReplicatedChronicleMap, incrementSize(); // this is just like notifyPut, but it won't trigger the modification iterators - notifyRemotePut(offset, true, posFromOffset(offset), keySize, inBytes, valueSize, valuePos, kPosition); + notifyRemotePut(offset, true, posFromOffset(offset), keySize, inBytes, valueSize, valuePos, + kPosition, null, valueReader, keyReader); } finally { writeUnlock(); @@ -913,13 +990,14 @@ final class ReplicatedChronicleMap, long offset = putEntry(copies, metaKeyWriter, keyWriter, key, keySize, hash2, value, identifier, timestamp, hashLookup); incrementSize(); - notifyPut(offset, true, key, value, posFromOffset(offset)); + notifyPut(offset, true, key, value, null, posFromOffset(offset)); return null; } finally { writeUnlock(); } } + /** * Used only with replication, its sometimes possible to receive an old ( or stale update ) from a * remote map. This method is used to determine if we should ignore such updates.

We can reject @@ -1309,12 +1387,12 @@ final class ReplicatedChronicleMap, @Override public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, - boolean added, K key, V value, long pos, SharedSegment segment) { + boolean added, K key, V replacedValue, V value, long pos, SharedSegment segment) { assert ReplicatedChronicleMap.this == map : "ModificationIterator.onPut() is called from outside of the parent map"; try { - nextListener.onPut(map, entry, metaDataBytes, added, key, value, pos, segment); + nextListener.onPut(map, entry, metaDataBytes, added, key, replacedValue, value, pos, segment); } catch (Exception e) { LOG.error("", e); } @@ -1323,7 +1401,7 @@ final class ReplicatedChronicleMap, try { final ModificationIterator modificationIterator = modificationIterators.get((int) next); modificationIterator.onPut(map, entry, metaDataBytes, - added, key, value, pos, segment); + added, key, replacedValue, value, pos, segment); } catch (Exception e) { LOG.error("", e); } @@ -1445,7 +1523,7 @@ final class ReplicatedChronicleMap, @Override public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, - boolean added, K key, V value, long pos, SharedSegment segment) { + boolean added, K key, V replacedValue, V value, long pos, SharedSegment segment) { assert ReplicatedChronicleMap.this == map : "ModificationIterator.onPut() is called from outside of the parent map"; 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/SerializationBuilder.java b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java index 3c6b46510..99adb577c 100644 --- a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java @@ -60,7 +60,7 @@ final class SerializationBuilder implements Cloneable { private enum CopyingInterop {FROM_MARSHALLER, FROM_WRITER} - private final Serializable bufferIdentity; + private final Role role; final Class eClass; private boolean instancesAreMutable; private SizeMarshaller sizeMarshaller = stopBit(); @@ -70,11 +70,10 @@ final class SerializationBuilder implements Cloneable { private MetaBytesInterop metaInterop; private MetaProvider metaInteropProvider; private long maxSize; - private ObjectFactory factory; @SuppressWarnings("unchecked") SerializationBuilder(Class eClass, Role role) { - this.bufferIdentity = role; + this.role = role; this.eClass = eClass; instancesAreMutable = instancesAreMutable(eClass); @@ -83,39 +82,39 @@ final class SerializationBuilder implements Cloneable { NullObjectFactory.of(); if (concreteClass(eClass) && Byteable.class.isAssignableFrom(eClass)) { - ByteableMarshaller agileBytesMarshaller = ByteableMarshaller.of((Class) eClass); - sizeMarshaller(agileBytesMarshaller) - .reader(agileBytesMarshaller, factory) - .interop(agileBytesMarshaller); + ByteableMarshaller byteableMarshaller = ByteableMarshaller.of((Class) eClass); + sizeMarshaller(byteableMarshaller) + .reader(byteableMarshaller) + .interop(byteableMarshaller); } else if (eClass == CharSequence.class || eClass == String.class) { - reader((BytesReader) CharSequenceReader.of(), factory); + reader((BytesReader) CharSequenceReader.of()); writer((BytesWriter) CharSequenceWriter.instance()); } else if (eClass == Void.class) { sizeMarshaller(VoidMarshaller.INSTANCE) - .reader((BytesReader) VoidMarshaller.INSTANCE, factory) + .reader((BytesReader) VoidMarshaller.INSTANCE) .interop((BytesInterop) VoidMarshaller.INSTANCE); } else if (eClass == Long.class) { sizeMarshaller(LongMarshaller.INSTANCE) - .reader((BytesReader) LongMarshaller.INSTANCE, factory) + .reader((BytesReader) LongMarshaller.INSTANCE) .interop((BytesInterop) LongMarshaller.INSTANCE); } else if (eClass == Double.class) { sizeMarshaller(DoubleMarshaller.INSTANCE) - .reader((BytesReader) DoubleMarshaller.INSTANCE, factory) + .reader((BytesReader) DoubleMarshaller.INSTANCE) .interop((BytesInterop) DoubleMarshaller.INSTANCE); } else if (eClass == Integer.class) { sizeMarshaller(IntegerMarshaller.INSTANCE) - .reader((BytesReader) IntegerMarshaller.INSTANCE, factory) + .reader((BytesReader) IntegerMarshaller.INSTANCE) .interop((BytesInterop) IntegerMarshaller.INSTANCE); } else if (eClass == byte[].class) { - reader((BytesReader) ByteArrayMarshaller.INSTANCE, factory); + reader((BytesReader) ByteArrayMarshaller.INSTANCE); interop((BytesInterop) ByteArrayMarshaller.INSTANCE); } else if (eClass == char[].class) { - reader((BytesReader) CharArrayMarshaller.INSTANCE, factory); + reader((BytesReader) CharArrayMarshaller.INSTANCE); interop((BytesInterop) CharArrayMarshaller.INSTANCE); } else if (concreteClass(eClass)) { BytesMarshaller marshaller = chooseMarshaller(eClass, eClass); if (marshaller != null) - marshaller(marshaller, factory); + marshaller(marshaller); } if (concreteClass(eClass) && marshallerUseFactory(eClass)) { factory(factory); @@ -150,17 +149,15 @@ final class SerializationBuilder implements Cloneable { return copyingInterop(CopyingInterop.FROM_WRITER) .setInterop(writer) .metaInterop(CopyingMetaBytesInterop - .>forBytesWriter(bufferIdentity)); + .>forBytesWriter(role)); } - public SerializationBuilder marshaller(BytesMarshaller marshaller, - ObjectFactory factory) { - if (factory == null) factory = this.factory; + public SerializationBuilder marshaller(BytesMarshaller marshaller) { return copyingInterop(CopyingInterop.FROM_MARSHALLER) - .reader(BytesReaders.fromBytesMarshaller(marshaller), factory) + .reader(BytesReaders.fromBytesMarshaller(marshaller)) .setInterop(marshaller) .metaInterop(CopyingMetaBytesInterop - .>forBytesMarshaller(bufferIdentity)); + .>forBytesMarshaller(role)); } public SerializationBuilder maxSize(long maxSize) { @@ -218,10 +215,7 @@ final class SerializationBuilder implements Cloneable { public SerializationBuilder objectSerializer(ObjectSerializer serializer) { if (reader == null || interop == null) { - ObjectFactory factory = this.factory; - if (factory == null) - factory = NullObjectFactory.INSTANCE; - marshaller(new SerializableMarshaller(serializer), factory); + marshaller(new SerializableMarshaller(serializer)); } return this; } @@ -244,10 +238,8 @@ final class SerializationBuilder implements Cloneable { return reader; } - public SerializationBuilder reader(BytesReader reader, ObjectFactory factory) { + public SerializationBuilder reader(BytesReader reader) { this.reader = reader; - if (factory != null) - this.factory = factory; return this; } @@ -281,10 +273,6 @@ final class SerializationBuilder implements Cloneable { return this; } - public ObjectFactory factory() { - return factory; - } - @SuppressWarnings("unchecked") public SerializationBuilder factory(ObjectFactory factory) { if (!marshallerUseFactory(eClass)) { @@ -292,19 +280,17 @@ final class SerializationBuilder implements Cloneable { " value don't use object factory"); } else if (interop instanceof ByteableMarshaller) { if (factory instanceof AllocateInstanceObjectFactory) { - ByteableMarshaller agileBytesMarshaller = + ByteableMarshaller byteableMarshaller = ByteableMarshaller.of((Class) eClass); - sizeMarshaller(agileBytesMarshaller) - .reader(agileBytesMarshaller, factory) - .interop(agileBytesMarshaller); + sizeMarshaller(byteableMarshaller) + .reader(byteableMarshaller) + .interop(byteableMarshaller); } else { - ByteableMarshaller agileBytesMarshaller = + ByteableMarshaller byteableMarshaller = ByteableMarshaller.of((Class) eClass, (ObjectFactory) factory); - ObjectFactory factory1 = factory; - if (factory1 == null) factory1 = this.factory; - sizeMarshaller(agileBytesMarshaller) - .reader(agileBytesMarshaller, factory1) - .interop(agileBytesMarshaller); + sizeMarshaller(byteableMarshaller) + .reader(byteableMarshaller) + .interop(byteableMarshaller); } } else if (interop instanceof BytesMarshallableMarshaller) { if (factory instanceof AllocateInstanceObjectFactory) { @@ -329,11 +315,12 @@ final class SerializationBuilder implements Cloneable { } else { // interop is custom, it is user's responsibility to use the same factory inside // marshaller and standalone + String role = this.role.toString().toLowerCase(); throw new IllegalStateException( - "Change the value factory simultaneously with marshaller " + - "using *AndFactory() method"); + role + "DeserializationFactory() should be called " + + "only if the Map " + role + " type is Byteable, BytesMarshallable " + + "or Externalizable subtype and no custom marshallers were configured."); } - this.factory = factory; return this; } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 77784449a..38eee0456 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -19,8 +19,7 @@ package net.openhft.chronicle.map; import com.sun.jdi.connect.spi.ClosedConnectionException; -import net.openhft.chronicle.hash.exceptions.IORuntimeException; -import net.openhft.chronicle.hash.exceptions.TimeoutRuntimeException; +import net.openhft.chronicle.hash.RemoteCallTimeoutException; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; import net.openhft.lang.threadlocal.ThreadLocalCopies; @@ -942,7 +941,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private void checkTimeout(long timeoutTime) { if (timeoutTime < System.currentTimeMillis()) - throw new TimeoutRuntimeException(); + throw new RemoteCallTimeoutException(); } private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId) { 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/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 98a46941c..135db017f 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -19,7 +19,9 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHashErrorListener; -import net.openhft.chronicle.hash.serialization.*; +import net.openhft.chronicle.hash.serialization.BytesReader; +import net.openhft.chronicle.hash.serialization.Hasher; +import net.openhft.chronicle.hash.serialization.SizeMarshaller; import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter; import net.openhft.chronicle.hash.serialization.internal.MetaProvider; @@ -32,6 +34,7 @@ import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory; import net.openhft.lang.threadlocal.Provider; import net.openhft.lang.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -58,8 +61,7 @@ class VanillaChronicleMap, private static final Logger LOG = LoggerFactory.getLogger(VanillaChronicleMap.class); /** - * Because DirectBitSet implementations couldn't find more than 64 continuous clear or set - * bits. + * Because DirectBitSet implementations couldn't find more than 64 continuous clear or set bits. */ static final int MAX_ENTRY_OVERSIZE_FACTOR = 64; static final ThreadLocal tmpBytes = new ThreadLocal<>(); @@ -76,7 +78,6 @@ class VanillaChronicleMap, final VW originalValueWriter; final MVW originalMetaValueWriter; final MetaProvider metaValueWriterProvider; - final ObjectFactory valueFactory; final DefaultValueProvider defaultValueProvider; final PrepareValueBytesAsWriter prepareValueBytesAsWriter; final int metaDataBytes; @@ -126,7 +127,6 @@ class VanillaChronicleMap, originalValueWriter = (VW) valueBuilder.interop(); originalMetaValueWriter = (MVW) valueBuilder.metaInterop(); metaValueWriterProvider = (MetaProvider) valueBuilder.metaInteropProvider(); - valueFactory = valueBuilder.factory(); defaultValueProvider = builder.defaultValueProvider(); prepareValueBytesAsWriter = builder.prepareValueBytesAsWriter(); @@ -484,9 +484,9 @@ class VanillaChronicleMap, /** - * removes ( if there exists ) an entry from the map, if the {@param key} and {@param - * expectedValue} match that of a maps.entry. If the {@param expectedValue} equals null then ( - * if there exists ) an entry whose key equals {@param key} this is removed. + * removes ( if there exists ) an entry from the map, if the {@param key} and {@param expectedValue} match + * that of a maps.entry. If the {@param expectedValue} equals null then ( if there exists ) an entry whose + * key equals {@param key} this is removed. * * @param k the key of the entry to remove * @param expectedValue null if not required @@ -546,8 +546,8 @@ class VanillaChronicleMap, * replace the value in a map, only if the existing entry equals {@param existingValue} * * @param key the key into the map - * @param existingValue the expected existing value in the map ( could be null when we don't - * wish to do this check ) + * @param existingValue the expected existing value in the map ( could be null when we don't wish to do + * this check ) * @param newValue the new value you wish to store in the map * @return the value that was replaced */ @@ -680,7 +680,7 @@ class VanillaChronicleMap, /** * close for native values does not call put on close */ - final MutableLockedEntry writeUnlock = isNativeValueClass + final WriteLocked writeUnlock = isNativeValueClass ? new NativeWriteLocked() : new HeapWriteLocked(); @@ -766,7 +766,7 @@ class VanillaChronicleMap, } - public MutableLockedEntry readLock() throws IllegalStateException { + public ReadLocked readLock() throws IllegalStateException { while (true) { final boolean success = segmentHeader.tryRWReadLock(LOCK_OFFSET, lockTimeOutNS); if (success) { @@ -784,7 +784,7 @@ class VanillaChronicleMap, } } - public MutableLockedEntry writeLock() throws IllegalStateException { + public WriteLocked writeLock() throws IllegalStateException { while (true) { final boolean success = segmentHeader.tryRWWriteLock(LOCK_OFFSET, lockTimeOutNS); if (success) return writeUnlock; @@ -926,7 +926,7 @@ class VanillaChronicleMap, entry.skip(keySizeMarshaller.sizeEncodingSize(keySize) + keySize); skipReplicationBytes(entry); V v = readValue(copies, entry, usingValue); - notifyPut(offset, true, key, v, posFromOffset(offset)); + notifyPut(offset, true, key, v, null, posFromOffset(offset)); notifyGet(offset, key, v); return v; } @@ -965,7 +965,7 @@ class VanillaChronicleMap, // key is not found long offset = putEntry(copies, metaKeyInterop, keyInterop, key, keySize, value); incrementSize(); - notifyPut(offset, true, key, value, posFromOffset(offset)); + notifyPut(offset, true, key, value, null, posFromOffset(offset)); return null; } @@ -981,7 +981,7 @@ class VanillaChronicleMap, // putValue may relocate entry and change offset offset = putValue(pos, offset, entry, valueSizePos, entryEndAddr, copies, value, null, searchedHashLookup); - notifyPut(offset, false, key, value, posFromOffset(offset)); + notifyPut(offset, false, key, value, prevValue, posFromOffset(offset)); return prevValue; } @@ -1082,7 +1082,7 @@ class VanillaChronicleMap, return readValue(copies, entry, value, readValueSize(entry)); } - V readValue(ThreadLocalCopies copies, MultiStoreBytes entry, V value, long valueSize) { + V readValue(ThreadLocalCopies copies, NativeBytes entry, V value, long valueSize) { copies = valueReaderProvider.getCopies(copies); BytesReader valueReader = valueReaderProvider.get(copies, originalValueReader); return valueReader.read(entry, valueSize, value); @@ -1097,12 +1097,12 @@ class VanillaChronicleMap, /** * Removes a key (or key-value pair) from the Segment. * - *

The entry will only be removed if {@code expectedValue} equals to {@code null} or the - * value previously corresponding to the specified key. + *

The entry will only be removed if {@code expectedValue} equals to {@code null} or the value + * previously corresponding to the specified key. * * @param hash2 a hash code related to the {@code keyBytes} - * @return the value of the entry that was removed if the entry corresponding to the {@code - * keyBytes} exists and {@link #removeReturnsNull} is {@code false}, {@code null} otherwise + * @return the value of the entry that was removed if the entry corresponding to the {@code keyBytes} + * exists and {@link #removeReturnsNull} is {@code false}, {@code null} otherwise */ V remove(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V expectedValue, long hash2) { @@ -1162,9 +1162,9 @@ class VanillaChronicleMap, } /** - * Replaces the specified value for the key with the given value. {@code newValue} is set - * only if the existing value corresponding to the specified key is equal to {@code - * expectedValue} or {@code expectedValue == null}. + * Replaces the specified value for the key with the given value. {@code newValue} is set only if the + * existing value corresponding to the specified key is equal to {@code expectedValue} or {@code + * expectedValue == null}. * * @param hash2 a hash code related to the {@code keyBytes} * @return the replaced value or {@code null} if the value was not replaced @@ -1205,19 +1205,20 @@ class VanillaChronicleMap, // putValue may relocate entry and change offset offset = putValue(pos, offset, entry, valueSizePos, entryEndAddr, copies, newValue, null, searchedHashLookup); - notifyPut(offset, false, key, newValue, posFromOffset(offset)); + notifyPut(offset, false, key, newValue, valueRead, posFromOffset(offset)); return valueRead; } return null; } - void notifyPut(long offset, boolean added, K key, V value, final long pos) { + void notifyPut(long offset, boolean added, K key, V value, @Nullable final V replacedValue, + final long pos) { if (eventListener() != MapEventListeners.NOP) { MultiStoreBytes b = acquireTmpBytes(); b.storePositionAndSize(bytes, offset, entrySize); eventListener().onPut(VanillaChronicleMap.this, b, metaDataBytes, - added, key, value, pos, this); + added, key, replacedValue, value, pos, this); } } @@ -1240,17 +1241,17 @@ class VanillaChronicleMap, } /** - * Replaces value in existing entry. May cause entry relocation, because there may be not - * enough space for new value in location already allocated for this entry. + * Replaces value in existing entry. May cause entry relocation, because there may be not enough space + * for new value in location already allocated for this entry. * * @param pos index of the first block occupied by the entry - * @param offset relative offset of the entry in Segment bytes (before, i. e. - * including metaData) + * @param offset relative offset of the entry in Segment bytes (before, i. e. including + * metaData) * @param entry relative pointer in Segment bytes * @param valueSizePos relative position of value size in entry * @param entryEndAddr absolute address of the entry end - * @return relative offset of the entry in Segment bytes after putting value (that may cause - * entry relocation) + * @return relative offset of the entry in Segment bytes after putting value (that may cause entry + * relocation) */ long putValue(long pos, long offset, NativeBytes entry, long valueSizePos, long entryEndAddr, ThreadLocalCopies copies, V value, Bytes valueBytes, @@ -1661,7 +1662,7 @@ class VanillaChronicleMap, } - private static abstract class ReadLocked extends MutableLockedEntry extends MutableLockedEntry implements ReadContext { boolean present; @@ -1680,7 +1681,7 @@ class VanillaChronicleMap, } } - private static abstract class WriteLocked extends MutableLockedEntry extends MutableLockedEntry implements WriteContext { } } diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index fd607424c..e91126a96 100644 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -19,9 +19,12 @@ 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; +import net.openhft.chronicle.hash.serialization.SizeMarshaller; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; import net.openhft.lang.io.serialization.BytesMarshaller; @@ -53,7 +56,8 @@ public class ChronicleSetBuilder ChronicleSetBuilder(Class keyClass) { chronicleMapBuilder = ChronicleMapBuilder.of(keyClass, DummyValue.class) - .valueMarshallers(DummyValueMarshaller.INSTANCE, DummyValueMarshaller.INSTANCE); + .valueMarshallers(DummyValueMarshaller.INSTANCE, DummyValueMarshaller.INSTANCE) + .valueSizeMarshaller(DummyValueMarshaller.INSTANCE); } public static ChronicleSetBuilder of(Class keyClass) { @@ -225,6 +229,19 @@ public class ChronicleSetBuilder return this; } + @Override + public ChronicleSetBuilder keyMarshallers(@NotNull BytesWriter keyWriter, + @NotNull BytesReader keyReader) { + chronicleMapBuilder.keyMarshallers(keyWriter, keyReader); + return this; + } + + @Override + public ChronicleSetBuilder keySizeMarshaller(@NotNull SizeMarshaller keySizeMarshaller) { + chronicleMapBuilder.keySizeMarshaller(keySizeMarshaller); + return this; + } + /** * {@inheritDoc} * @@ -249,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 41db8cced..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.unknownTopology(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/CHMMetaDataTest.java b/src/test/java/net/openhft/chronicle/map/CHMMetaDataTest.java index 21204db14..65f699a45 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMMetaDataTest.java +++ b/src/test/java/net/openhft/chronicle/map/CHMMetaDataTest.java @@ -81,7 +81,7 @@ public class CHMMetaDataTest { @Override public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, - boolean added, String key, String value, long pos, + boolean added, String key, String replacedValue, String value, long pos, SharedSegment segment) { assertEquals(8, metaDataBytes); if (added) diff --git a/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java index 452f53657..3feaf6dfb 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,13 +53,13 @@ public class ChannelReplicationTest { @Before public void setup() throws IOException { { - TcpConfig tcpConfig = TcpConfig + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forSendingNode(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); hubA = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) - .create((byte) 1); + .createWithId((byte) 1); map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) @@ -67,11 +67,11 @@ 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); + .createWithId((byte) 2); map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) diff --git a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java index f7782a1d0..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,20 +30,49 @@ 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 * * @author Rob Austin. */ - public class EventListenerTestWithTCPSocketReplication { ChronicleMap map1; ChronicleMap map2; + final AtomicReference keyRef = new AtomicReference(); + + final AtomicReference replacedValueRef = new AtomicReference(); + final AtomicReference valueRef = new AtomicReference(); + final AtomicBoolean putWasCalled = new AtomicBoolean(false); + + final AtomicBoolean wasRemoved = new AtomicBoolean(false); + final AtomicReference valueRemoved = new AtomicReference(); + + final MapEventListener> eventListener = new + MapEventListener>() { + private static final long serialVersionUID = 1L; + + @Override + public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, Integer key, + CharSequence value, CharSequence replacedValue) { + putWasCalled.getAndSet(true); + keyRef.set(key); + valueRef.set(value); + replacedValueRef.set(replacedValue); + } + + @Override + public void onRemove(ChronicleMap map, Bytes entry, int metaDataBytes, Integer key, CharSequence value) { + wasRemoved.set(true); + valueRemoved.set(value); + } + }; + + @After public void tearDown() throws InterruptedException { @@ -54,8 +83,12 @@ public class EventListenerTestWithTCPSocketReplication { e.printStackTrace(); } } - - // Thread.sleep(1600); + wasRemoved.set(false); + valueRemoved.set(null); + putWasCalled.set(false); + valueRef.set(null); + replacedValueRef.set(null); + keyRef.set(null); } @@ -63,60 +96,26 @@ public class EventListenerTestWithTCPSocketReplication { public void testAdded() throws IOException, InterruptedException { - final AtomicBoolean putWasCalled = new AtomicBoolean(false); - - final AtomicReference keyRef = new AtomicReference(); - - - final AtomicReference valueRef = new AtomicReference(); - - MapEventListener> eventListener = new MapEventListener>() { - private static final long serialVersionUID = 1L; - - @Override - public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, Integer key, - CharSequence value) { - putWasCalled.getAndSet(true); - keyRef.set(key); - valueRef.set(value); - } - }; - // ---------- SERVER1 1 ---------- { // we connect the maps via a TCP socket connection on port 8077 - TcpConfig tcpConfig = forSendingNode(8076, new InetSocketAddress("localhost", 8077)); - ChronicleMapBuilder map1Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 2, tcpConfig); + ChronicleMapBuilder map1Builder = serverBuilder(); map1 = map1Builder.create(); } // ---------- SERVER2 2 on the same server as ---------- { - TcpConfig tcpConfig = TcpConfig.forReceivingOnlyNode(8077); - ChronicleMapBuilder map2Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 1, tcpConfig) - .eventListener(eventListener); + ChronicleMapBuilder map2Builder = clientBuilder(); map2 = map2Builder.create(); } // we will stores some data into one map here map1.put(5, "EXAMPLE"); - int t = 0; - for (; t < 5000; t++) { - if (map1.equals(map2)) { - break; - } - Thread.sleep(1); - } + waitTillReplicated(); Assert.assertEquals("Maps should be equal", map1, map2); Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); @@ -126,141 +125,130 @@ public class EventListenerTestWithTCPSocketReplication { Assert.assertEquals(Integer.valueOf(5), keyRef.get()); Assert.assertEquals("EXAMPLE", valueRef.get()); - keyRef.set(null); - valueRef.set(null); } @Test - public void testAddedOriginalValueToBeOverWritten() throws IOException, InterruptedException { - - - final AtomicBoolean putWasCalled = new AtomicBoolean(false); - - final AtomicReference keyRef = new AtomicReference(); - - - final AtomicReference valueRef = new AtomicReference(); - - MapEventListener> eventListener = new MapEventListener>() { - private static final long serialVersionUID = 1L; - - @Override - public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, Integer key, - CharSequence value) { - putWasCalled.getAndSet(true); - keyRef.set(key); - valueRef.set(value); - } - }; + public void testRemoteNotifyPutLargerValue() throws IOException, InterruptedException { // ---------- SERVER1 1 ---------- { // we connect the maps via a TCP socket connection on port 8077 - TcpConfig tcpConfig = forSendingNode(8076, new InetSocketAddress("localhost", 8077)); - ChronicleMapBuilder map1Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 2, tcpConfig); + ChronicleMapBuilder map1Builder = serverBuilder(); map1 = map1Builder.create(); + } // ---------- SERVER2 2 on the same server as ---------- { - TcpConfig tcpConfig = TcpConfig.forReceivingOnlyNode(8077); - ChronicleMapBuilder map2Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 1, tcpConfig) - .eventListener(eventListener); - map2 = map2Builder.create(); + map2 = clientBuilder().create(); map2.put(5, "WILL_GET_OVER-WRITTEN"); } + // we will stores some data into one map here map1.put(5, "EXAMPLE"); - int t = 0; - for (; t < 5000; t++) { - if (map1.equals(map2)) { - break; - } - Thread.sleep(1); - } + waitTillReplicated(); - Assert.assertEquals("Maps should be equal", map1, map2); - Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); - Assert.assertTrue("Map 2 should not be empty", !map2.isEmpty()); Assert.assertTrue("The eventListener.onPut should have been called", putWasCalled.get()); Assert.assertEquals(Integer.valueOf(5), keyRef.get()); Assert.assertEquals("EXAMPLE", valueRef.get()); - - keyRef.set(null); - valueRef.set(null); + Assert.assertEquals("WILL_GET_OVER-WRITTEN", replacedValueRef.get()); } @Test - public void testAlreadyAddedWithSmallInitialValue() throws IOException, InterruptedException { + public void testRemoteNotifyPutSmallerValue() throws IOException, InterruptedException { - final AtomicBoolean putWasCalled = new AtomicBoolean(false); - - final AtomicReference keyRef = new AtomicReference(); - - - final AtomicReference valueRef = new AtomicReference(); - - MapEventListener> eventListener = new MapEventListener>() { - private static final long serialVersionUID = 1L; - - @Override - public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, Integer key, - CharSequence value) { - putWasCalled.getAndSet(true); - keyRef.set(key); - valueRef.set(value); - } - }; - // ---------- SERVER1 1 ---------- { // we connect the maps via a TCP socket connection on port 8077 - TcpConfig tcpConfig = forSendingNode(8076, new InetSocketAddress("localhost", 8077)); - ChronicleMapBuilder map1Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 2, tcpConfig); + map1 = serverBuilder().create(); - map1 = map1Builder.create(); - map1.put(5, "WILL_GET_OVERWRITEN"); } // ---------- SERVER2 2 on the same server as ---------- { - TcpConfig tcpConfig = TcpConfig.forReceivingOnlyNode(8077); - ChronicleMapBuilder map2Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 1, tcpConfig) - .eventListener(eventListener); - map2 = map2Builder.create(); + map2 = clientBuilder().create(); map2.put(5, "small"); } // we will stores some data into one map here map1.put(5, "EXAMPLE"); + + waitTillReplicated(); + + + Assert.assertTrue("The eventListener.onPut should have been called", putWasCalled.get()); + Assert.assertEquals(Integer.valueOf(5), keyRef.get()); + Assert.assertEquals("small", replacedValueRef.get()); + Assert.assertEquals("EXAMPLE", valueRef.get()); + + } + + + @Test + public void testRemoteNotifyRemove() throws IOException, InterruptedException { + + // ---------- SERVER1 1 ---------- + { + + // we connect the maps via a TCP socket connection on port 8077 + + ChronicleMapBuilder map1Builder = serverBuilder(); + + map1 = map1Builder.create(); + + + } + + // ---------- SERVER2 2 on the same server as ---------- + { + map2 = clientBuilder().create(); + map2.put(5, "WILL_GET_REMOVED"); + } + + // its we have to wait here for this entry to be removed - as removing an entry that you dont have + // on map 1 will have no effect + waitTillReplicated(); + + // no more puts after this point + putWasCalled.set(false); + + // we will stores some data into one map here + map1.remove(5); + + waitTillReplicated(); + + Assert.assertTrue(!putWasCalled.get()); + Assert.assertTrue(wasRemoved.get()); + Assert.assertEquals("WILL_GET_REMOVED", valueRemoved.get()); + Assert.assertEquals(null, map2.get(5)); + } + + private ChronicleMapBuilder clientBuilder() { + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.forReceivingOnlyNode(8077); + return ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replication((byte) 1, tcpConfig) + .eventListener(eventListener); + } + + + private void waitTillReplicated() throws InterruptedException { int t = 0; for (; t < 5000; t++) { if (map1.equals(map2)) { @@ -268,17 +256,13 @@ public class EventListenerTestWithTCPSocketReplication { } Thread.sleep(1); } + } - Assert.assertEquals("Maps should be equal", map1, map2); - Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); - Assert.assertTrue("Map 2 should not be empty", !map2.isEmpty()); - - Assert.assertTrue("The eventListener.onPut should have been called", putWasCalled.get()); - Assert.assertEquals(Integer.valueOf(5), keyRef.get()); - Assert.assertEquals("EXAMPLE", valueRef.get()); - - keyRef.set(null); - valueRef.set(null); + private ChronicleMapBuilder serverBuilder() { + TcpTransportAndNetworkConfig tcpConfig = forSendingNode(8076, new InetSocketAddress("localhost", 8077)); + return ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replication((byte) 2, tcpConfig); } } \ No newline at end of file diff --git a/src/test/java/net/openhft/chronicle/map/Issue3.java b/src/test/java/net/openhft/chronicle/map/Issue3Test.java similarity index 98% rename from src/test/java/net/openhft/chronicle/map/Issue3.java rename to src/test/java/net/openhft/chronicle/map/Issue3Test.java index e622a4afc..c10802ccd 100644 --- a/src/test/java/net/openhft/chronicle/map/Issue3.java +++ b/src/test/java/net/openhft/chronicle/map/Issue3Test.java @@ -26,7 +26,7 @@ import java.io.IOException; import java.util.Random; import java.util.Set; -public class Issue3 { +public class Issue3Test { @Test public void test() throws IOException { diff --git a/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java b/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java index f555addca..b86bf8ce2 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,12 +62,12 @@ 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); hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig) - .create((byte) 1); + .createWithId((byte) 1); // this is how you add maps after the custer is created map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) @@ -81,12 +81,12 @@ public class PostChannelCreationTest { { - final TcpConfig tcpConfig = TcpConfig + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forSendingNode(8087, new InetSocketAddress("localhost", 8086)) .heartBeatInterval(1, SECONDS); hubB = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig) - .create((byte) 2); + .createWithId((byte) 2); // this is how you add maps after the custer is created map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) diff --git a/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java b/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java index 9d1af470e..e58e04b19 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; @@ -68,7 +68,7 @@ public class SingleMapChannelTest { hubA = ReplicationHub.builder() .maxEntrySize(maxEntrySize) .tcpTransportAndNetwork(tcpConfig) - .create(identifier); + .createWithId(identifier); // this is how you add maps after the custer is created map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) @@ -77,11 +77,11 @@ public class SingleMapChannelTest { { - final TcpConfig tcpConfig = TcpConfig + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forSendingNode(8087, new InetSocketAddress("localhost", 8086)) .heartBeatInterval(1, SECONDS); - hubB = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig).create((byte) 2); + hubB = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig).createWithId((byte) 2); // this is how you add maps after the custer is created map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) 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/TCPSocketReplicationBootStrapTests.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java index 963d0ed13..bad095b27 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java @@ -68,8 +68,7 @@ public class TCPSocketReplicationBootStrapTests { { // restart map 2 but don't doConnect it to map one final ChronicleMap map2b = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .createPersistedTo(map2File); + map2aBuilder.createPersistedTo(map2File); // add data into it map2b.put(11, "ADDED WHEN DISCONNECTED TO MAP1"); map2b.close(); @@ -112,8 +111,7 @@ public class TCPSocketReplicationBootStrapTests { { // restart map 2 but don't doConnect it to map one final ChronicleMap map2b = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .createPersistedTo(map2File); + map2aBuilder.createPersistedTo(map2File); // add data into it map2b.put(11, "ADDED WHEN DISCONNECTED TO MAP1"); map2b.close(); diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java index 98636a26b..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.unknownTopology(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..cfc7988c9 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,14 +69,14 @@ public class ChannelExample { byte identifier = (byte) 1; - TcpConfig tcpConfig = TcpConfig + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forSendingNode(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); hubOnServer1 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) - .create(identifier); + .createWithId(identifier); // this demotes favoriteColour short channel1 = (short) 1; @@ -104,12 +104,12 @@ 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) - .create(identifier); + .createWithId(identifier); // this demotes favoriteColour diff --git a/src/test/java/net/openhft/chronicle/map/jrs166/map/MapCheck.java b/src/test/java/net/openhft/chronicle/map/jrs166/map/MapCheck.java index b97be30c2..e97ce82af 100644 --- a/src/test/java/net/openhft/chronicle/map/jrs166/map/MapCheck.java +++ b/src/test/java/net/openhft/chronicle/map/jrs166/map/MapCheck.java @@ -33,7 +33,7 @@ import java.util.*; /** * @test * @synopsis Times and checks basic map operations - *

+ *

* When run with "s" second arg, this requires file "testwords", which * is best used with real words. We can't check in this file, but you * can create one from a real dictionary (1 line per word) and then run 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 89325d86e..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; @@ -26,7 +26,6 @@ import org.junit.Test; import java.io.Closeable; import java.io.IOException; import java.net.InetSocketAddress; -import java.util.Arrays; import static java.util.Arrays.asList; import static java.util.concurrent.TimeUnit.SECONDS; @@ -53,7 +52,7 @@ public class SetForTCPSocketReplicationTest3way { final int serverPort, final InetSocketAddress... endpoints) throws IOException { return (T) ChronicleSetBuilder.of(Integer.class) - .replication(identifier, TcpConfig.unknownTopology(serverPort, asList(endpoints)) + .replication(identifier, TcpTransportAndNetworkConfig.forUnknownTopology(serverPort, asList(endpoints)) .heartBeatInterval(1L, SECONDS)) .createPersistedTo(getPersistenceFile()); }