SPI Queue/QE

Author: JerryShea

src/main/java/net/openhft/chronicle/queue/impl/single/SingleChronicleQueueBuilder.java

@@ -36,14 +36,17 @@
 import net.openhft.chronicle.queue.impl.table.ReadonlyTableStore;
 import net.openhft.chronicle.queue.impl.table.SingleTableBuilder;
 import net.openhft.chronicle.queue.internal.domestic.QueueOffsetSpec;
-import net.openhft.chronicle.threads.*;
+import net.openhft.chronicle.queue.providers.EnterpriseQueueFactories;
+import net.openhft.chronicle.threads.MediumEventLoop;
+import net.openhft.chronicle.threads.Pauser;
+import net.openhft.chronicle.threads.TimingPauser;
+import net.openhft.chronicle.threads.YieldingPauser;
 import net.openhft.chronicle.wire.*;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
 import javax.crypto.spec.SecretKeySpec;
 import java.io.File;
-import java.lang.reflect.Constructor;
 import java.nio.file.Path;
 import java.time.LocalTime;
 import java.time.ZoneId;
@@ -64,7 +67,6 @@ public class SingleChronicleQueueBuilder extends SelfDescribingMarshallable impl
     public static final long SMALL_BLOCK_SIZE = OS.isWindows() ? OS.SAFE_PAGE_SIZE : OS.pageSize(); // the smallest safe block size on Windows 8+
 
     public static final long DEFAULT_SPARSE_CAPACITY = 512L << 30;
-    private static final Constructor<?> ENTERPRISE_QUEUE_CONSTRUCTOR;
     private static final WireStoreFactory storeFactory = SingleChronicleQueueBuilder::createStore;
     private static final Supplier<TimingPauser> TIMING_PAUSER_SUPPLIER = DefaultPauserSupplier.INSTANCE;
 
@@ -74,15 +76,6 @@ public class SingleChronicleQueueBuilder extends SelfDescribingMarshallable impl
         CLASS_ALIASES.addAlias(SCQRoll.class, "SCQSRoll");
         CLASS_ALIASES.addAlias(SCQIndexing.class, "SCQSIndexing");
         CLASS_ALIASES.addAlias(SingleChronicleQueueStore.class, "SCQStore");
-
-        Constructor<?> co;
-        try {
-            co = ((Class<?>) Class.forName("software.chronicle.enterprise.queue.EnterpriseSingleChronicleQueue")).getDeclaredConstructors()[0];
-            Jvm.setAccessible(co);
-        } catch (Exception e) {
-            co = null;
-        }
-        ENTERPRISE_QUEUE_CONSTRUCTOR = co;
     }
 
     private BufferMode writeBufferMode = BufferMode.None;
@@ -224,7 +217,7 @@ static SingleChronicleQueueStore createStore(@NotNull RollingChronicleQueue queu
     }
 
     public static boolean areEnterpriseFeaturesAvailable() {
-        return ENTERPRISE_QUEUE_CONSTRUCTOR != null;
+        return EnterpriseQueueFactories.get().areEnterpriseFeaturesAvailable();
     }
 
     private static RollCycle loadDefaultRollCycle() {
@@ -284,10 +277,10 @@ public SingleChronicleQueue build() {
 
         // It is important to check enterprise features after preBuild()
         // Enterprise-only config options can be loaded from the metadata
-        if (checkEnterpriseFeaturesRequested())
-            chronicleQueue = buildEnterprise();
-        else
-            chronicleQueue = new SingleChronicleQueue(this);
+        if (checkEnterpriseFeaturesRequested() && !areEnterpriseFeaturesAvailable()) {
+            throw new IllegalStateException("Enterprise features requested but Chronicle Queue Enterprise is not in the class path!");
+        }
+        chronicleQueue = EnterpriseQueueFactories.get().newInstance(this);
 
         postBuild(chronicleQueue);
 
@@ -324,23 +317,11 @@ private boolean checkEnterpriseFeaturesRequested() {
     }
 
     public static boolean onlyAvailableInEnterprise(final String feature) {
-        if (ENTERPRISE_QUEUE_CONSTRUCTOR == null)
+        if (!areEnterpriseFeaturesAvailable())
             Jvm.warn().on(SingleChronicleQueueBuilder.class, feature + " is only supported in Chronicle Queue Enterprise. If you would like to use this feature, please contact sales@chronicle.software for more information.");
         return true;
     }
 
-    @NotNull
-    private SingleChronicleQueue buildEnterprise() {
-        if (ENTERPRISE_QUEUE_CONSTRUCTOR == null)
-            throw new IllegalStateException("Enterprise features requested but Chronicle Queue Enterprise is not in the class path!");
-
-        try {
-            return (SingleChronicleQueue) ENTERPRISE_QUEUE_CONSTRUCTOR.newInstance(this);
-        } catch (Exception e) {
-            throw new IllegalStateException("Couldn't create an instance of Enterprise queue", e);
-        }
-    }
-
     public SingleChronicleQueueBuilder aesEncryption(@Nullable byte[] keyBytes) {
         if (keyBytes == null) {
             codingSuppliers(null, null);

src/main/java/net/openhft/chronicle/queue/impl/single/SingleChronicleQueueFactory.java

@@ -0,0 +1,16 @@
+package net.openhft.chronicle.queue.impl.single;
+
+import net.openhft.chronicle.queue.providers.QueueFactory;
+
+public class SingleChronicleQueueFactory implements QueueFactory {
+
+    @Override
+    public boolean areEnterpriseFeaturesAvailable() {
+        return false;
+    }
+
+    @Override
+    public SingleChronicleQueue newInstance(SingleChronicleQueueBuilder queueBuilder) {
+        return new SingleChronicleQueue(queueBuilder);
+    }
+}

src/main/java/net/openhft/chronicle/queue/providers/EnterpriseQueueFactories.java

@@ -0,0 +1,39 @@
+package net.openhft.chronicle.queue.providers;
+
+import net.openhft.chronicle.core.Jvm;
+import net.openhft.chronicle.queue.impl.single.SingleChronicleQueueFactory;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.ServiceLoader;
+
+public class EnterpriseQueueFactories {
+
+    private static final String PREFERRED_FACTORY_CLASS_NAME = Jvm.getProperty("net.openhft.chronicle.queue.providers.EnterpriseQueueWrapper",
+            "software.chronicle.enterprise.queue.EnterpriseQueueFactory");
+
+    private static QueueFactory queueFactory;
+
+    /**
+     * Get the {@link QueueFactory}
+     *
+     * @return the active queue wrapper
+     */
+    @NotNull
+    public static QueueFactory get() {
+        if (queueFactory == null) {
+            final ServiceLoader<QueueFactory> load = ServiceLoader.load(QueueFactory.class);
+            for (QueueFactory factory : load) {
+                // last one in wins, unless we encounter the "preferred" one
+                queueFactory = factory;
+                if (PREFERRED_FACTORY_CLASS_NAME.equals(factory.getClass().getName())) {
+                    break;
+                }
+            }
+            if (queueFactory == null) {
+                Jvm.error().on(EnterpriseQueueFactories.class, "There's no queue wrapper factory configured, this shouldn't happen. ");
+                queueFactory = new SingleChronicleQueueFactory();
+            }
+        }
+        return queueFactory;
+    }
+}

src/main/java/net/openhft/chronicle/queue/providers/QueueFactory.java

@@ -0,0 +1,12 @@
+package net.openhft.chronicle.queue.providers;
+
+import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
+import net.openhft.chronicle.queue.impl.single.SingleChronicleQueueBuilder;
+
+public interface QueueFactory {
+
+    // SPI for SingleChronicleQueueBuilder
+    boolean areEnterpriseFeaturesAvailable();
+
+    SingleChronicleQueue newInstance(SingleChronicleQueueBuilder queueBuilder);
+}

src/main/resources/META-INF/services/net.openhft.chronicle.queue.providers.QueueFactory

@@ -0,0 +1 @@
+net.openhft.chronicle.queue.impl.single.SingleChronicleQueueFactory