diff --git a/src/main/java/net/openhft/chronicle/hash/FindByName.java b/src/main/java/net/openhft/chronicle/hash/FindByName.java index e8f31fa2e..dcfc27564 100644 --- a/src/main/java/net/openhft/chronicle/hash/FindByName.java +++ b/src/main/java/net/openhft/chronicle/hash/FindByName.java @@ -1,6 +1,7 @@ package net.openhft.chronicle.hash; import java.io.IOException; +import java.util.concurrent.TimeoutException; /** * @author Rob Austin. @@ -14,6 +15,6 @@ public interface FindByName { * @throws IllegalArgumentException if a map with this name can not be found * @throws IOException if it not possible to create the map or set */ - T create(String name) throws IllegalArgumentException, - IOException; + T from(String name) throws IllegalArgumentException, + IOException, TimeoutException, InterruptedException; } 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 7a1e590f9..fa39cc671 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java @@ -119,5 +119,14 @@ public abstract class AbstractReplication implements Serializable{ throw new IllegalArgumentException("Identifier must be positive, " + identifier + " given"); } + + @Override + public String toString() { + return + "tcpConfig=" + tcpConfig + + ", udpConfig=" + udpConfig + + ", remoteNodeValidator=" + remoteNodeValidator ; + + } } } 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 fb2a1c5db..866d7fa17 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java @@ -23,6 +23,8 @@ import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.set.ChronicleSet; import org.jetbrains.annotations.NotNull; +import java.util.Arrays; + /** * This class holds all configurations of * multicontainer replication, which is usable, when you want to replicate several {@linkplain @@ -68,6 +70,16 @@ public final class ReplicationHub extends AbstractReplication { channels = new ReplicationChannel[builder.maxNumberOfChannels]; } + @Override + public String toString() { + return "ReplicationHub{" + + "maxEntrySize=" + maxEntrySize + + ", channels=" + Arrays.toString(channels) + + super.toString() + + '}'; + } + + /** * Returns the maximum {@linkplain ChronicleHashBuilder#entrySize(int) entry size} of {@link * ChronicleHash}es, replicated by this {@code ReplicationHub}, could have. {@code RuntimeException} is @@ -173,5 +185,8 @@ public final class ReplicationHub extends AbstractReplication { check(identifier); return new ReplicationHub(identifier, this); } + + } + } diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java index dffbcec6c..4dddab982 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java @@ -296,9 +296,6 @@ interface ChronicleMapBuilderI extends Serializable { ChronicleMapBuilderI name(String name); - - - String name(); diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index c7dbdf23e..89d7c7c13 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -238,12 +238,16 @@ public class NodeDiscovery { } }; + // remove self + final HashSet knownHostPorts0 = new HashSet<>(knownHostPorts); + knownHostPorts0.remove(ourAddressAndPort); + // add our identifier and host:port to the list of known identifiers knownNodes.add(ourAddressAndPort, identifier); final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(ourAddressAndPort.getPort(), - toInetSocketCollection(knownHostPorts)) + toInetSocketCollection(knownHostPorts0)) .heartBeatInterval(1, SECONDS); diff --git a/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java b/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java index 7179b4983..6d2abcab2 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java @@ -19,11 +19,16 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.chronicle.hash.ChronicleHashInstanceConfig; import net.openhft.chronicle.hash.FindByName; +import net.openhft.chronicle.hash.replication.ReplicationChannel; import net.openhft.chronicle.hash.replication.ReplicationHub; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; +import java.io.Serializable; import java.util.Map; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import static net.openhft.chronicle.map.ChronicleMapBuilder.of; @@ -32,29 +37,66 @@ import static net.openhft.chronicle.map.ChronicleMapBuilder.of; * @author Rob Austin. */ class ReplicationHubFindByName implements FindByName { + public static final Logger LOG = LoggerFactory.getLogger(ReplicationHubFindByName.class.getName()); public static final int MAP_BY_NAME_CHANNEL = 1; - private final ReplicationHub replicationHub; - int maxEntrySize = 10 * 1024; private final AtomicInteger nextFreeChannel = new AtomicInteger(1); - private final Map map; + private final Map map; + private final ReplicationHub replicationHub; + + + public static class ChronicleMapBuilderWithChannelId implements Serializable { + ChronicleMapBuilder chronicleMapBuilder; + int channelId; + } /** * @throws IOException */ public ReplicationHubFindByName(ReplicationHub replicationHub) throws IOException { + + + LOG.info("connecting to replicationHub=" + replicationHub); + this.replicationHub = replicationHub; - this.map = (Map) of(CharSequence.class, ChronicleMapBuilder.class) + ReplicationChannel channel = replicationHub.createChannel((short) MAP_BY_NAME_CHANNEL); + + this.map = (Map) of(CharSequence.class, ChronicleMapBuilderWithChannelId.class) .entrySize(2500) .entries(128) .instance() - .replicatedViaChannel(replicationHub.createChannel((short) MAP_BY_NAME_CHANNEL)).create(); + .replicatedViaChannel(channel) + .create(); } - ChronicleMapBuilder get(String name) throws IllegalArgumentException { - ChronicleMapBuilder chronicleMapBuilder = map.get(name); + public T create(ChronicleMapBuilder builder) throws IllegalArgumentException, + IOException, TimeoutException, InterruptedException { + + int withChannelId = nextFreeChannel.incrementAndGet(); + T result = (T) toReplicatedViaChannel(builder, withChannelId).create(); + ChronicleMapBuilderWithChannelId builderWithChannelId = new + ChronicleMapBuilderWithChannelId(); + + builderWithChannelId.channelId = withChannelId; + builderWithChannelId.chronicleMapBuilder = builder; + map.put(builder.name(), builderWithChannelId); + return result; + } + + + public T from(String name) throws IllegalArgumentException, + IOException, TimeoutException, InterruptedException { + return (T) replicatedViaChannel(name).create(); + } + + + ChronicleMapBuilderWithChannelId get(String name) throws IllegalArgumentException, + TimeoutException, + InterruptedException { + + ChronicleMapBuilderWithChannelId chronicleMapBuilder = waitTillEntryReceived(1000, name); if (chronicleMapBuilder == null) throw new IllegalArgumentException("A map name=" + name + " can not be found."); @@ -62,13 +104,29 @@ class ReplicationHubFindByName implements FindByName { } - public T create(String name) throws IllegalArgumentException, - IOException { - return (T) replicatedViaChannel(name).create(); + /** + * * waits until map1 and map2 show the same value + * + * @param timeOutMs timeout in milliseconds + * @throws InterruptedException + */ + private ChronicleMapBuilderWithChannelId waitTillEntryReceived(final int timeOutMs, String name) throws TimeoutException, InterruptedException { + int t = 0; + for (; t < timeOutMs; t++) { + ChronicleMapBuilderWithChannelId builder = map.get(name); + + if (builder != null) + return builder; + Thread.sleep(1); + } + throw new TimeoutException("timed out wait for map name=" + name); + } - private ChronicleHashInstanceConfig replicatedViaChannel(String name) { - return toReplicatedViaChannel(get(name)); + + private ChronicleHashInstanceConfig replicatedViaChannel(String name) throws TimeoutException, InterruptedException { + ChronicleMapBuilderWithChannelId builder = get(name); + return toReplicatedViaChannel(builder.chronicleMapBuilder, builder.channelId); } /** @@ -83,7 +141,7 @@ class ReplicationHubFindByName implements FindByName { * @see ChronicleMapBuilder#createPersistedTo(File file) */ public T createPersistedTo(String name, File file) throws IllegalArgumentException, - IOException { + IOException, TimeoutException, InterruptedException { return (T) replicatedViaChannel(name).persistedTo(file).create(); } @@ -95,7 +153,11 @@ class ReplicationHubFindByName implements FindByName { return builder.instance().replicatedViaChannel(replicationHub.createChannel((short) channelId)); } - public void add(ChronicleMapBuilder builder) { - this.map.put(builder.name(), builder); + private ChronicleHashInstanceConfig toReplicatedViaChannel(ChronicleMapBuilder builder, int + withChannelId) { + + return builder.instance().replicatedViaChannel(replicationHub.createChannel((short) withChannelId)); } + + } diff --git a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java index 43fd7efdc..763b4069c 100644 --- a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java +++ b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java @@ -24,6 +24,7 @@ import org.junit.Ignore; import org.junit.Test; import java.io.IOException; +import java.util.concurrent.TimeoutException; /** * @author Rob Austin. @@ -44,7 +45,8 @@ public class MapByNameTest { } @Test - public void testSerializingBuilder() { + @Ignore + public void testSerializingBuilder() throws TimeoutException, InterruptedException, IOException { { final ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) .minSegments(2) @@ -52,25 +54,25 @@ public class MapByNameTest { .replication((byte) 1, TcpTransportAndNetworkConfig.of(8244)) .removeReturnsNull(true); - findMapByName.add(builder); + // findMapByName.add(builder); } - final ChronicleMap map = findMapByName.get("map1").create(); + final ChronicleMap map = findMapByName.from("map1"); map.put("hello", "world"); Assert.assertEquals(map.get("hello"), "world"); } - @Test(expected = IllegalArgumentException.class) - public void testSerializingBuilderUnknownMap() { - findMapByName.get("hello"); - } + // @Test(expected = IllegalArgumentException.class) + // public void testSerializingBuilderUnknownMap() throws TimeoutException, InterruptedException { + // findMapByName.get("hello"); + // } // currently work in progress @Test @Ignore - public void testConnectByName() throws IOException, InterruptedException { + public void testConnectByName() throws IOException, InterruptedException, TimeoutException { NodeDiscovery nodeDiscovery = new NodeDiscovery(); @@ -81,30 +83,36 @@ public class MapByNameTest { .removeReturnsNull(true); ReplicationHubFindByName mapByName = nodeDiscovery.mapByName(); - mapByName.add(builder); + // mapByName.add(builder); - ChronicleMap myMap2 = mapByName.create("myMap"); + ChronicleMap myMap2 = mapByName.from("myMap"); } - public static void main(String... args) throws IOException, InterruptedException { + public static void main(String... args) throws IOException, InterruptedException, TimeoutException { NodeDiscovery nodeDiscovery = new NodeDiscovery(); -/* + + + final ReplicationHubFindByName mapByName = nodeDiscovery.mapByName(); + ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence .class, CharSequence.class) .minSegments(2) - .name("myMap") + .name("myMap4") .removeReturnsNull(true); -*/ + ChronicleMap chronicleHash = mapByName.create(builder); + chronicleHash.put("hello", "world"); - ReplicationHubFindByName mapByName = nodeDiscovery.mapByName(); + final ChronicleMap myMap = mapByName.from("myMap3"); + + // myMap.put("hello", "world"); - ChronicleMapBuilder myMap = mapByName.get("myMap"); - + System.out.print(myMap); + Thread.sleep(2000); }