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 4935d72fd..6565f9fc3 --- 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; public boolean bootstapOnlyLocalEntries = false; AbstractChronicleMapBuilder(Class keyClass, Class valueClass) { @@ -184,6 +186,12 @@ abstract class AbstractChronicleMapBuilder pushTo(InetSocketAddress... addresses) { + this.pushToAddresses = addresses; + return this; + } + @Override public B clone() { try { @@ -803,6 +811,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); @@ -847,6 +856,7 @@ abstract class AbstractChronicleMapBuilder createReplicated(byte identifier) throws IOException { + pushingToMapEventListener(); preMapConstruction(true); VanillaChronicleMap map = new ReplicatedChronicleMap, @@ -859,6 +869,7 @@ abstract class AbstractChronicleMapBuilder createWithoutFile( SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) { + pushingToMapEventListener(); try { VanillaChronicleMap map = newMap(singleHashReplication, channel); BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE, @@ -871,8 +882,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()).error("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 1226d21f5..d50a13cef --- 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 bb765910e..cdf5df893 --- 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);