From 07b75d8b8ac815fdd120dd5639f011de5195ec70 Mon Sep 17 00:00:00 2001 From: leventov Date: Thu, 22 Oct 2015 00:19:59 +0300 Subject: [PATCH] Allow ReplicatedChronicleMap to be accessed without configuring replicators --- .../chronicle/map/ChronicleMapBuilder.java | 8 +- .../chronicle/map/ReplicatedChronicleMap.java | 7 +- ...dMapWithoutConfiguringReplicatorsTest.java | 102 ++++++++++++++++++ 3 files changed, 114 insertions(+), 3 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/DeserializeReplicatedMapWithoutConfiguringReplicatorsTest.java diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index eabbe5d98..110fbbc2f 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -1633,7 +1633,6 @@ public final class ChronicleMapBuilder implements persisted = false; try { - // pushingToMapEventListener(); VanillaChronicleMap map = newMap(singleHashReplication, channel); // TODO this method had been moved // if(OS.warnOnWindows(map.sizeInBytesWithoutTiers())){ @@ -1737,12 +1736,17 @@ public final class ChronicleMapBuilder implements replicators.add(Replicators.tcp(singleHashReplication)); if (singleHashReplication.udpTransport() != null) replicators.add(Replicators.udp(singleHashReplication.udpTransport())); - } else { + } else if (channel != null) { ReplicationHub hub = channel.hub(); ChannelProvider provider = ChannelProvider.getProvider(hub); ChannelProvider.ChronicleChannel ch = provider.createChannel(channel.channelId()); replicators.add(ch); + } else { + assert persisted && !((ReplicatedChronicleMap) map).createdOrInMemory : + "No Replicators for replicated ChronicleMap could be only on " + + "deserialization/access of existing replicated ChronicleMap, " + + "when replication in not needed in this JVM/run"; } for (Replicator replicator : replicators) { Closeable token = replicator.applyTo(this, result, result, diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 7d7d7cccc..7efe0f8b3 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -128,7 +128,6 @@ public class ReplicatedChronicleMap) builder.remoteOperations; this.localIdentifier = replication.identifier(); this.bootstrapOnlyLocalEntries = replication.bootstrapOnlyLocalEntries(); @@ -177,6 +176,12 @@ public class ReplicatedChronicleMap builder) { + super.initTransientsFromBuilder(builder); + this.remoteOperations = (MapRemoteOperations) builder.remoteOperations; + } + private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { in.defaultReadObject(); initOwnTransients(); diff --git a/src/test/java/net/openhft/chronicle/map/DeserializeReplicatedMapWithoutConfiguringReplicatorsTest.java b/src/test/java/net/openhft/chronicle/map/DeserializeReplicatedMapWithoutConfiguringReplicatorsTest.java new file mode 100644 index 000000000..1c40423a0 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/DeserializeReplicatedMapWithoutConfiguringReplicatorsTest.java @@ -0,0 +1,102 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.map; + +import com.google.common.collect.ImmutableMap; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.Set; + +import static org.junit.Assert.assertEquals; + +public class DeserializeReplicatedMapWithoutConfiguringReplicatorsTest { + + @Test + public void deserializeReplicatedMapWithoutConfiguringReplicatorsTest() + throws IOException, InterruptedException { + + File file1 = File.createTempFile("test", "cron"); + file1.deleteOnExit(); + try (ChronicleMap map1 = ChronicleMap.of(Integer.class, Integer.class) + .entries(100) + .replication((byte) 1) + .createPersistedTo(file1)) { + map1.put(1, 1); + } + + try (ChronicleMap map1 = ChronicleMap.of(Integer.class, Integer.class) + .entries(100) +// .replication((byte) 1) + .createPersistedTo(file1)) { + map1.put(2, 2); + } + + File file2 = File.createTempFile("test", "cron"); + file2.deleteOnExit(); + try (ChronicleMap map2 = ChronicleMap.of(Integer.class, Integer.class) + .entries(100) + .replication((byte) 2) + .createPersistedTo(file2)) { + map2.put(3, 3); + } + + // Let the subsequent remove from map2 be later than put in map1 + Thread.sleep(1); + + try (ChronicleMap map2 = ChronicleMap.of(Integer.class, Integer.class) + .entries(100) +// .replication((byte) 2) + .createPersistedTo(file2)) { + map2.put(2, 0); + map2.remove(2); + } + + try (ChronicleMap map1 = ChronicleMap.of(Integer.class, Integer.class) + .entries(100) + .replication((byte) 1, TcpTransportAndNetworkConfig + .of(10000, new InetSocketAddress("localhost", 10001))) + .createPersistedTo(file1); + ChronicleMap map2 = ChronicleMap.of(Integer.class, Integer.class) + .entries(100) + .replication((byte) 2, TcpTransportAndNetworkConfig + .of(10001, new InetSocketAddress("localhost", 10000))) + .createPersistedTo(file2)) { + Builder.waitTillEqual(map1, map2, 3000); + assertEquals(map1, map2); + assertEquals(map1, ImmutableMap.of(1, 1, 3, 3)); + assertEquals(map2, ImmutableMap.of(1, 1, 3, 3)); + } + } + + Set threads; + + @Before + public void sampleThreads() { + threads = Thread.getAllStackTraces().keySet(); + } + + @After + public void checkThreadsShutdown() { + ChannelReplicationTest.checkThreadsShutdown(threads); + } +}