Made ExecutorService configurable in AdvancedRangeCache.Builder

SmartDataAnalytics · Sep 8, 2024 · 1c2e8a6 · 1c2e8a6
1 parent abad2d5
commit 1c2e8a6
Showing 1 changed file with 25 additions and 14 deletions.
diff --git a/...sw-commons-io-buffers/src/main/java/org/aksw/commons/io/cache/AdvancedRangeCacheImpl.java b/...sw-commons-io-buffers/src/main/java/org/aksw/commons/io/cache/AdvancedRangeCacheImpl.java
@@ -46,16 +46,16 @@ public class AdvancedRangeCacheImpl<T>
     // Number of items a worker processes in bulk before signalling available data
     protected int workerBulkSize;
 
-    protected ExecutorService executorService =
-            MoreExecutors.getExitingExecutorService((ThreadPoolExecutor)Executors.newCachedThreadPool());
+    protected ExecutorService executorService;
 
     public AdvancedRangeCacheImpl(
             ReadableChannelSource<T> dataSource,
             Slice<T> slice,
             long requestLimit,
             int workerBulkSize,
             Duration terminationDelay,
-            int maxReadAheadItemCount) {
+            int maxReadAheadItemCount,
+            ExecutorService executorService) {
 
         this.dataSource = dataSource;
 
@@ -64,23 +64,24 @@ public AdvancedRangeCacheImpl(
         this.workerBulkSize = workerBulkSize;
         this.terminationDelay = terminationDelay;
         this.maxReadAheadItemCount = maxReadAheadItemCount;
+        this.executorService = executorService;
     }
 
     @Override
     public ArrayOps<T> getArrayOps() {
         return slice.getArrayOps();
     }
 
-    public static <A> AdvancedRangeCacheImpl<A> create(
-            ReadableChannelSource<A> dataSource,
-            Slice<A> slice,
-            long requestLimit,
-            int workerBulkSize,
-            Duration terminationDelay,
-            int maxReadAheadItemCount) {
-
-        return new AdvancedRangeCacheImpl<>(dataSource, slice, requestLimit, workerBulkSize, terminationDelay, maxReadAheadItemCount);
-    }
+//    public static <A> AdvancedRangeCacheImpl<A> create(
+//            ReadableChannelSource<A> dataSource,
+//            Slice<A> slice,
+//            long requestLimit,
+//            int workerBulkSize,
+//            Duration terminationDelay,
+//            int maxReadAheadItemCount) {
+//
+//        return new AdvancedRangeCacheImpl<>(dataSource, slice, requestLimit, workerBulkSize, terminationDelay, maxReadAheadItemCount);
+//    }
 
 
     public ReadableChannelSource<T> getDataSource() {
@@ -225,6 +226,7 @@ public static class Builder<A> {
         protected Duration terminationDelay;
 
         protected int maxReadAheadItemCount;
+        protected ExecutorService executorService;
 
         public ReadableChannelSource<A> getDataSource() {
             return dataSource;
@@ -280,6 +282,11 @@ public Builder<A> setMaxReadAheadItemCount(int maxReadAheadItemCount) {
             return this;
         }
 
+        public Builder<A> setExecutorService(ExecutorService executorService) {
+            this.executorService = executorService;
+            return this;
+        }
+
 
 //		public Duration getSyncDelay() {
 //			return syncDelay;
@@ -291,7 +298,11 @@ public Builder<A> setMaxReadAheadItemCount(int maxReadAheadItemCount) {
 //		}
 
         public AdvancedRangeCacheImpl<A> build() {
-            return AdvancedRangeCacheImpl.create(dataSource, slice, requestLimit, workerBulkSize, terminationDelay, maxReadAheadItemCount);
+            ExecutorService finalExecutorService = executorService != null
+                ? executorService
+                : MoreExecutors.getExitingExecutorService((ThreadPoolExecutor)Executors.newCachedThreadPool());
+
+            return new AdvancedRangeCacheImpl<>(dataSource, slice, requestLimit, workerBulkSize, terminationDelay, maxReadAheadItemCount, finalExecutorService);
         }
     }