From a3b149d8fb35174ea084bba82166ad6726b3ad4a Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Tue, 18 Nov 2014 20:57:55 +0000 Subject: [PATCH 1/2] CHENT-8 Add support for pushing changes to remote maps. --- .../map/AbstractChronicleMapBuilder.java | 35 +++++++++++++++++++ .../chronicle/map/ChronicleMapBuilder.java | 5 +++ .../chronicle/map/ChronicleMapBuilderI.java | 2 ++ 3 files changed, 42 insertions(+) mode change 100644 => 100755 src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java mode change 100644 => 100755 src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java mode change 100644 => 100755 src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java old mode 100644 new mode 100755 index 1eec457db..5bdf343ed --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java @@ -45,6 +45,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.*; +import java.lang.reflect.Constructor; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.util.ArrayList; @@ -142,6 +143,7 @@ abstract class AbstractChronicleMapBuilder prepareValueBytes = null; private SingleChronicleHashReplication singleHashReplication = null; + private InetSocketAddress[] pushToAddresses; AbstractChronicleMapBuilder(Class keyClass, Class valueClass) { keyBuilder = new SerializationBuilder(keyClass, SerializationBuilder.Role.KEY); @@ -183,6 +185,12 @@ abstract class AbstractChronicleMapBuilder pushTo(InetSocketAddress... addresses) { + this.pushToAddresses = addresses; + return this; + } + @Override public B clone() { try { @@ -802,6 +810,7 @@ abstract class AbstractChronicleMapBuilder createWithFile(File file, SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) throws IOException { + pushingToMapEventListener(); for (int i = 0; i < 10; i++) { if (file.exists() && file.length() > 0) { try (FileInputStream fis = new FileInputStream(file); @@ -846,6 +855,7 @@ abstract class AbstractChronicleMapBuilder createReplicated(byte identifier) throws IOException { + pushingToMapEventListener(); preMapConstruction(true); VanillaChronicleMap map = new ReplicatedChronicleMap, @@ -858,6 +868,7 @@ abstract class AbstractChronicleMapBuilder createWithoutFile( SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) { + pushingToMapEventListener(); try { VanillaChronicleMap map = newMap(singleHashReplication, channel); BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE, @@ -870,8 +881,32 @@ abstract class AbstractChronicleMapBuilder pmel = Class.forName("com.higherfrequencytrading.chronicle.engine.map.PushingMapEventListener"); + Constructor constructor = pmel.getConstructor(ChronicleMap[].class); + // create a stateless client for each address + ChronicleMap[] statelessClients = new ChronicleMap[pushToAddresses.length]; + ChronicleMapBuilderI cmb = clone(); + cmb.pushTo((InetSocketAddress[]) null); + for (int i = 0; i < pushToAddresses.length; i++) { + cmb.statelessClient(pushToAddresses[i]); + statelessClients[i] = cmb.create(); + } + eventListener = (MapEventListener>) constructor.newInstance(statelessClients); + } catch (ClassNotFoundException e) { + LoggerFactory.getLogger(getClass().getName()).warn("Chronicle Enterprise not found in the class path"); + } catch (Exception e) { + LoggerFactory.getLogger(getClass().getName()).warn("PushingMapEventListener failed to load", e); + } + } + ChronicleMap createStatelessMap(StatelessMapConfig statelessBuilder) throws IOException { + pushingToMapEventListener(); preMapConstruction(false); return new StatelessChronicleMap( statelessBuilder, diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java old mode 100644 new mode 100755 index fe2e7d240..a448e808e --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -356,7 +356,12 @@ public final class ChronicleMapBuilder implements ChronicleMapBuilderI replication(SingleChronicleHashReplication withId) { delegate.replication(withId); return this; + } + @Override + public ChronicleMapBuilderI pushTo(InetSocketAddress... addresses) { + delegate.pushTo(addresses); + return this; } @Override diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java old mode 100644 new mode 100755 index 4dddab982..316e74064 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java @@ -248,6 +248,8 @@ interface ChronicleMapBuilderI extends Serializable { ChronicleMapBuilderI replication(SingleChronicleHashReplication withId); + ChronicleMapBuilderI pushTo(InetSocketAddress... addresses); + ChronicleMap create(); ChronicleMapBuilderI actualSegments(int actualSegments); From 2a5551ad68b69e8296308cb24532f18ae417b40b Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Tue, 18 Nov 2014 21:01:03 +0000 Subject: [PATCH 2/2] CHENT-8 Add support for pushing changes to remote maps. --- .../net/openhft/chronicle/map/AbstractChronicleMapBuilder.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java index 25a9c7998..6565f9fc3 100755 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java @@ -901,7 +901,7 @@ abstract class AbstractChronicleMapBuilder