+ fixed close for ConseqServiceFactory

Author: q3769

pom.xml

@@ -27,7 +27,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>io.github.q3769</groupId>
     <artifactId>conseq4j</artifactId>
-    <version>20230924.0.0</version>
+    <version>20230924.0.20230927</version>
     <packaging>jar</packaging>
     <name>conseq4j</name>
     <description>A Java concurrent API to sequence related tasks while concurring unrelated ones</description>

src/main/java/conseq4j/summon/ConseqServiceFactory.java

@@ -31,10 +31,7 @@
 import javax.annotation.concurrent.ThreadSafe;
 import java.util.List;
 import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.*;
 
 import static java.lang.Math.floorMod;
 
@@ -67,16 +64,16 @@ private ConseqServiceFactory(int concurrency) {
     /**
      * @return ExecutorService factory with default concurrency
      */
-    public static ConseqServiceFactory instance() {
-        return new ConseqServiceFactory(Runtime.getRuntime().availableProcessors());
+    public static @Nonnull ConseqServiceFactory instance() {
+        return instance(Runtime.getRuntime().availableProcessors());
     }
 
     /**
      * @param concurrency
      *         max number of tasks possible to be executed in parallel
      * @return ExecutorService factory with given concurrency
      */
-    public static ConseqServiceFactory instance(int concurrency) {
+    public static @Nonnull ConseqServiceFactory instance(int concurrency) {
         return new ConseqServiceFactory(concurrency);
     }
 
@@ -86,12 +83,12 @@ public static ConseqServiceFactory instance(int concurrency) {
     @Override
     public ExecutorService getExecutorService(@NonNull Object sequenceKey) {
         return this.sequentialExecutors.computeIfAbsent(bucketOf(sequenceKey),
-                bucket -> new ShutdownDisabledExecutorService(Executors.newFixedThreadPool(1)));
+                bucket -> new ShutdownDisabledExecutorService(Executors.newSingleThreadExecutor()));
     }
 
     @Override
     public void close() {
-        sequentialExecutors.values().forEach(ExecutorService::close);
+        sequentialExecutors.values().forEach(ShutdownDisabledExecutorService::closeDelegate);
     }
 
     private int bucketOf(Object sequenceKey) {
@@ -137,6 +134,32 @@ public void shutdown() {
             throw new UnsupportedOperationException(SHUTDOWN_UNSUPPORTED_MESSAGE);
         }
 
+        @Override
+        public void close() {
+            throw new UnsupportedOperationException(SHUTDOWN_UNSUPPORTED_MESSAGE);
+        }
+
+        void closeDelegate() {
+            boolean terminated = isTerminated();
+            if (!terminated) {
+                shutdownDelegate();
+                boolean interrupted = false;
+                while (!terminated) {
+                    try {
+                        terminated = awaitTermination(1L, TimeUnit.DAYS);
+                    } catch (InterruptedException e) {
+                        if (!interrupted) {
+                            shutdownDelegateNow();
+                            interrupted = true;
+                        }
+                    }
+                }
+                if (interrupted) {
+                    Thread.currentThread().interrupt();
+                }
+            }
+        }
+
         void shutdownDelegate() {
             this.delegate.shutdown();
         }
@@ -153,6 +176,8 @@ private interface ShutdownOperations {
             void shutdown();
 
             List<Runnable> shutdownNow();
+
+            void close();
         }
     }
 }

src/test/java/conseq4j/summon/ConseqServiceFactoryTest.java

@@ -26,20 +26,21 @@
 import com.google.common.collect.Range;
 import conseq4j.SpyingTask;
 import conseq4j.TestUtils;
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.stream.Collectors;
 
 import static conseq4j.TestUtils.createSpyingTasks;
 import static conseq4j.TestUtils.getIfAllCompleteNormal;
 import static java.util.stream.Collectors.toList;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.*;
 
 /**
  * @author Qingtian Wang
@@ -132,4 +133,32 @@ void submitsRunAllTasksOfSameSequenceKeyInSequence() {
         TestUtils.awaitAllComplete(tasks);
         assertSingleThread(tasks);
     }
+
+    @Nested
+    class individualShutdownUnsupported {
+        @Test
+        void whenShutdownsCalled() {
+            ExecutorService sequentialExecutor = ConseqServiceFactory.instance().getExecutorService(UUID.randomUUID());
+
+            assertThrows(UnsupportedOperationException.class, sequentialExecutor::shutdown);
+            assertThrows(UnsupportedOperationException.class, sequentialExecutor::shutdownNow);
+            assertThrows(UnsupportedOperationException.class, sequentialExecutor::close);
+            assertFalse(sequentialExecutor.isShutdown());
+            assertFalse(sequentialExecutor.isTerminated());
+        }
+    }
+
+    @Nested
+    class factoryClose {
+        @Test
+        void whenCloseCalled() {
+            ConseqServiceFactory conseqServiceFactory = ConseqServiceFactory.instance();
+            ExecutorService sequentialExecutor = conseqServiceFactory.getExecutorService(UUID.randomUUID());
+
+            conseqServiceFactory.close();
+
+            assertTrue(sequentialExecutor.isShutdown());
+            assertTrue(sequentialExecutor.isTerminated());
+        }
+    }
 }