Cleanup dependencies (#4)

* Switch to upstream disruptor and update to 4.0.0 * Cleanup dependencies - Use official log4j instead of wso2 fork - Remove org.eclipse.osgi.services dependency and SiddhiManagerComponent class - Remove unused webjars dependency - Remove wso2 Maven repositories * Remove unused OSGi extension support
graylog-labs · Jan 25, 2024 · ac193e8 · ac193e8
1 parent b9d5574
commit ac193e8
Show file tree

Hide file tree

Showing 14 changed files with 52 additions and 334 deletions.
diff --git a/modules/siddhi-core/pom.xml b/modules/siddhi-core/pom.xml
@@ -44,11 +44,11 @@
             <artifactId>siddhi-annotations</artifactId>
         </dependency>
         <dependency>
-            <groupId>org.apache.log4j.wso2</groupId>
+            <groupId>log4j</groupId>
             <artifactId>log4j</artifactId>
         </dependency>
         <dependency>
-            <groupId>org.wso2.orbit.com.lmax</groupId>
+            <groupId>com.lmax</groupId>
             <artifactId>disruptor</artifactId>
         </dependency>
         <dependency>
@@ -77,14 +77,6 @@
             <groupId>com.google.code.gson</groupId>
             <artifactId>gson</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.eclipse.osgi</groupId>
-            <artifactId>org.eclipse.osgi.services</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.osgi</groupId>
-            <artifactId>org.osgi.core</artifactId>
-        </dependency>
         <dependency>
             <groupId>org.atteo.classindex</groupId>
             <artifactId>classindex</artifactId>

diff --git a/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/config/SiddhiAppContext.java b/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/config/SiddhiAppContext.java
@@ -38,6 +38,7 @@
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
 
 /**
  * Holder object for context information of {@link org.wso2.siddhi.query.api.SiddhiApp}.
@@ -50,6 +51,7 @@ public class SiddhiAppContext {
     private boolean enforceOrder;
     private StatisticsManager statisticsManager = null;
 
+    private ThreadFactory executorThreadFactory;
     private ExecutorService executorService;
     private ScheduledExecutorService scheduledExecutorService;
     private List<EternalReferencedHolder> eternalReferencedHolders;
@@ -139,6 +141,14 @@ public void setThreadBarrier(ThreadBarrier threadBarrier) {
         this.threadBarrier = threadBarrier;
     }
 
+    public ThreadFactory getExecutorThreadFactory() {
+        return this.executorThreadFactory;
+    }
+
+    public void setExecutorThreadFactory(ThreadFactory threadFactory) {
+        this.executorThreadFactory = threadFactory;
+    }
+
     public ExecutorService getExecutorService() {
         return executorService;
     }

diff --git a/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/partition/PartitionRuntime.java b/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/partition/PartitionRuntime.java
@@ -135,7 +135,7 @@ public QueryRuntime addQuery(QueryRuntime metaQueryRuntime) {
 
                 if (outputStreamJunction == null) {
                     outputStreamJunction = new StreamJunction(streamDefinition,
-                            siddhiAppContext.getExecutorService(),
+                            siddhiAppContext.getExecutorThreadFactory(),
                             siddhiAppContext.getBufferSize(),
                             null, siddhiAppContext);
                     localStreamJunctionMap.putIfAbsent(id, outputStreamJunction);
@@ -148,7 +148,7 @@ public QueryRuntime addQuery(QueryRuntime metaQueryRuntime) {
 
                 if (outputStreamJunction == null) {
                     outputStreamJunction = new StreamJunction(streamDefinition,
-                            siddhiAppContext.getExecutorService(),
+                            siddhiAppContext.getExecutorThreadFactory(),
                             siddhiAppContext.getBufferSize(),
                             null, siddhiAppContext);
                     streamJunctionMap.putIfAbsent(id, outputStreamJunction);
@@ -166,7 +166,7 @@ public QueryRuntime addQuery(QueryRuntime metaQueryRuntime) {
 
             if (outputStreamJunction == null) {
                 outputStreamJunction = new StreamJunction(streamDefinition,
-                        siddhiAppContext.getExecutorService(),
+                        siddhiAppContext.getExecutorThreadFactory(),
                         siddhiAppContext.getBufferSize(),
                         null, siddhiAppContext);
                 streamJunctionMap.putIfAbsent(id, outputStreamJunction);
@@ -291,7 +291,7 @@ private synchronized void clonePartition(String key) {
                         StreamJunction streamJunction = localStreamJunctionMap.get(streamId + key);
                         if (streamJunction == null) {
                             streamJunction = new StreamJunction(streamDefinition, siddhiAppContext
-                                    .getExecutorService(),
+                                    .getExecutorThreadFactory(),
                                     siddhiAppContext.getBufferSize(),
                                     null, siddhiAppContext);
                             localStreamJunctionMap.put(streamId + key, streamJunction);

diff --git a/...les/siddhi-core/src/main/java/org/wso2/siddhi/core/partition/PartitionStreamReceiver.java b/...les/siddhi-core/src/main/java/org/wso2/siddhi/core/partition/PartitionStreamReceiver.java
@@ -310,7 +310,7 @@ public void addStreamJunction(String key, List<QueryRuntime> queryRuntimeList) {
     }
 
     private StreamJunction createStreamJunction() {
-        return new StreamJunction(streamDefinition, siddhiAppContext.getExecutorService(),
+        return new StreamJunction(streamDefinition, siddhiAppContext.getExecutorThreadFactory(),
                 siddhiAppContext.getBufferSize(), null, siddhiAppContext);
     }
 

diff --git a/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/stream/StreamJunction.java b/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/stream/StreamJunction.java
@@ -52,7 +52,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadFactory;
 
 /**
  * Stream Junction is the place where streams are collected and distributed. There will be an Stream Junction per
@@ -69,7 +69,7 @@ public class StreamJunction implements EventBufferHolder {
     private int bufferSize;
     private List<Receiver> receivers = new CopyOnWriteArrayList<Receiver>();
     private List<Publisher> publishers = Collections.synchronizedList(new LinkedList<>());
-    private ExecutorService executorService;
+    private ThreadFactory threadFactory;
     private boolean async = false;
     private Disruptor<EventExchangeHolder> disruptor;
     private RingBuffer<EventExchangeHolder> ringBuffer;
@@ -80,12 +80,12 @@ public class StreamJunction implements EventBufferHolder {
     private OnErrorAction onErrorAction = OnErrorAction.LOG;
     private ExceptionListener exceptionListener;
 
-    public StreamJunction(StreamDefinition streamDefinition, ExecutorService executorService, int bufferSize,
+    public StreamJunction(StreamDefinition streamDefinition, ThreadFactory threadFactory, int bufferSize,
                           StreamJunction faultStreamJunction, SiddhiAppContext siddhiAppContext) {
         this.streamDefinition = streamDefinition;
         this.bufferSize = bufferSize;
         this.batchSize = bufferSize;
-        this.executorService = executorService;
+        this.threadFactory = threadFactory;
         this.siddhiAppContext = siddhiAppContext;
         if (siddhiAppContext.getStatisticsManager() != null) {
             this.throughputTracker = QueryParserHelper.createThroughputTracker(siddhiAppContext,
@@ -285,7 +285,7 @@ public synchronized void startProcessing() {
                     ProducerType producerType = ProducerType.MULTI;
                     disruptor = new Disruptor<EventExchangeHolder>(
                             new EventExchangeHolderFactory(streamDefinition.getAttributeList().size()),
-                            bufferSize, executorService, producerType,
+                            bufferSize, threadFactory, producerType,
                             new BlockingWaitStrategy());
                     disruptor.handleExceptionsWith(siddhiAppContext.getDisruptorExceptionHandler());
                     break;
@@ -294,7 +294,7 @@ public synchronized void startProcessing() {
             if (disruptor == null) {
                 disruptor = new Disruptor<EventExchangeHolder>(
                         new EventExchangeHolderFactory(streamDefinition.getAttributeList().size()),
-                        bufferSize, executorService);
+                        bufferSize, threadFactory);
                 disruptor.handleExceptionsWith(siddhiAppContext.getDisruptorExceptionHandler());
             }
             if (workers > 0) {

diff --git a/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/util/ReferenceHolder.java b/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/util/ReferenceHolder.java
diff --git a/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/util/SiddhiAppRuntimeBuilder.java b/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/util/SiddhiAppRuntimeBuilder.java
@@ -201,7 +201,7 @@ public String addQuery(QueryRuntime queryRuntime) {
 
             if (outputStreamJunction == null) {
                 outputStreamJunction = new StreamJunction(streamDefinition,
-                        siddhiAppContext.getExecutorService(),
+                        siddhiAppContext.getExecutorThreadFactory(),
                         siddhiAppContext.getBufferSize(), null, siddhiAppContext);
                 streamJunctionMap.putIfAbsent(streamDefinition.getId(), outputStreamJunction);
             }
@@ -217,7 +217,7 @@ public String addQuery(QueryRuntime queryRuntime) {
 
             if (outputStreamJunction == null) {
                 outputStreamJunction = new StreamJunction(streamDefinition,
-                        siddhiAppContext.getExecutorService(),
+                        siddhiAppContext.getExecutorThreadFactory(),
                         siddhiAppContext.getBufferSize(), null, siddhiAppContext);
                 streamJunctionMap.putIfAbsent(streamDefinition.getId(), outputStreamJunction);
             }

diff --git a/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/util/SiddhiExtensionLoader.java b/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/util/SiddhiExtensionLoader.java
@@ -20,19 +20,13 @@
 
 import org.apache.log4j.Logger;
 import org.atteo.classindex.ClassIndex;
-import org.osgi.framework.Bundle;
-import org.osgi.framework.BundleContext;
-import org.osgi.framework.BundleEvent;
-import org.osgi.framework.BundleListener;
-import org.osgi.framework.wiring.BundleWiring;
 import org.wso2.siddhi.annotation.Extension;
 import org.wso2.siddhi.core.executor.incremental.IncrementalAggregateBaseTimeFunctionExecutor;
 import org.wso2.siddhi.core.executor.incremental.IncrementalShouldUpdateFunctionExecutor;
 import org.wso2.siddhi.core.executor.incremental.IncrementalStartTimeEndTimeFunctionExecutor;
 import org.wso2.siddhi.core.executor.incremental.IncrementalTimeGetTimeZone;
 import org.wso2.siddhi.core.executor.incremental.IncrementalUnixTimeFunctionExecutor;
 
-import java.util.HashMap;
 import java.util.Map;
 
 /**
@@ -49,22 +43,6 @@ public class SiddhiExtensionLoader {
      */
     public static void loadSiddhiExtensions(Map<String, Class> siddhiExtensionsMap) {
         loadLocalExtensions(siddhiExtensionsMap);
-        BundleContext bundleContext = ReferenceHolder.getInstance().getBundleContext();
-        if (bundleContext != null) {
-            loadExtensionOSGI(bundleContext, siddhiExtensionsMap);
-        }
-    }
-
-    /**
-     * Load Extensions in OSGi environment.
-     *
-     * @param bundleContext       OSGi bundleContext
-     * @param siddhiExtensionsMap reference map for the Siddhi extension
-     */
-    private static void loadExtensionOSGI(BundleContext bundleContext, Map<String, Class> siddhiExtensionsMap) {
-        ExtensionBundleListener extensionBundleListener = new ExtensionBundleListener(siddhiExtensionsMap);
-        bundleContext.addBundleListener(extensionBundleListener);
-        extensionBundleListener.loadAllExtensions(bundleContext);
     }
 
     /**
@@ -151,52 +129,4 @@ private static void addExtensionToMap(String fqExtensionName, Class extensionCla
                     "loaded with the same namespace and name '" + fqExtensionName + "'");
         }
     }
-
-    /**
-     * Class to listen to Bundle changes to update available extensions.
-     */
-    private static class ExtensionBundleListener implements BundleListener {
-
-        private Map<Class, Integer> bundleExtensions = new HashMap<Class, Integer>();
-        private Map<String, Class> siddhiExtensionsMap;
-
-        ExtensionBundleListener(Map<String, Class> siddhiExtensionsMap) {
-            this.siddhiExtensionsMap = siddhiExtensionsMap;
-        }
-
-        @Override
-        public void bundleChanged(BundleEvent bundleEvent) {
-            if (bundleEvent.getType() == BundleEvent.STARTED) {
-                addExtensions(bundleEvent.getBundle());
-            } else {
-                removeExtensions(bundleEvent.getBundle());
-            }
-        }
-
-        private void addExtensions(Bundle bundle) {
-            ClassLoader classLoader = bundle.adapt(BundleWiring.class).getClassLoader();
-            Iterable<Class<?>> extensions = ClassIndex.getAnnotated(Extension.class, classLoader);
-            for (Class extension : extensions) {
-                addExtensionToMap(extension, siddhiExtensionsMap);
-                bundleExtensions.put(extension, (int) bundle.getBundleId());
-            }
-        }
-
-        private void removeExtensions(Bundle bundle) {
-            bundleExtensions.entrySet().stream().filter(entry -> entry.getValue() ==
-                    bundle.getBundleId()).forEachOrdered(entry -> {
-                siddhiExtensionsMap.remove(entry.getKey());
-            });
-            bundleExtensions.entrySet().removeIf(entry -> entry.getValue() ==
-                    bundle.getBundleId());
-        }
-
-        void loadAllExtensions(BundleContext bundleContext) {
-            for (Bundle b : bundleContext.getBundles()) {
-                if (b.getState() == Bundle.ACTIVE) {
-                    addExtensions(b);
-                }
-            }
-        }
-    }
 }
diff --git a/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/util/SiddhiManagerComponent.java b/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/util/SiddhiManagerComponent.java
diff --git a/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/util/parser/OutputParser.java b/modules/siddhi-core/src/main/java/org/wso2/siddhi/core/util/parser/OutputParser.java
@@ -250,7 +250,7 @@ public static OutputCallback constructOutputCallback(OutputStream outStream, Str
             StreamJunction outputStreamJunction = streamJunctionMap.get(id + key);
             if (outputStreamJunction == null) {
                 outputStreamJunction = new StreamJunction(outputStreamDefinition,
-                        siddhiAppContext.getExecutorService(),
+                        siddhiAppContext.getExecutorThreadFactory(),
                         siddhiAppContext.getBufferSize(), null, siddhiAppContext);
                 streamJunctionMap.putIfAbsent(id + key, outputStreamJunction);
             }