move logic from BoundedExecutor to KeySequentialBoundedExecutor

README.md

@@ -101,6 +101,6 @@ from multiple threads without synchronization.
 <dependency>
   <groupId>com.jano7</groupId>
   <artifactId>executor</artifactId>
-  <version>2.0.1</version>
+  <version>2.0.2</version>
 </dependency>
 ```

pom.xml

@@ -152,7 +152,7 @@
         <connection>scm:git:git://github.com/jano7/executor.git</connection>
         <developerConnection>scm:git:git@github.com:jano7/executor.git</developerConnection>
         <url>https://github.com/jano7/executor</url>
-        <tag>executor-2.0.1</tag>
+        <tag>executor-2.0.2</tag>
     </scm>
 
 </project>

src/main/java/com/jano7/executor/KeySequentialBoundedExecutor.java

@@ -24,23 +24,74 @@ of this software and associated documentation files (the "Software"), to deal
 package com.jano7.executor;
 
 import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
+import static com.jano7.executor.BoundedStrategy.BLOCK;
+import static com.jano7.executor.Util.checkNotNull;
+
 public final class KeySequentialBoundedExecutor implements DrainableExecutor {
 
-    private final BoundedExecutor boundedExecutor;
+    private final int maxTasks;
+
+    private final Semaphore semaphore;
+
+    private final KeySequentialExecutor keySequentialExecutor;
+
+    private final Runnable acquire;
+
+    private boolean drained = false;
 
     public KeySequentialBoundedExecutor(int maxTasks, BoundedStrategy onTasksExceeded, Executor underlyingExecutor) {
-        boundedExecutor = new BoundedExecutor(maxTasks, onTasksExceeded, new KeySequentialExecutor(underlyingExecutor));
+        this.maxTasks = maxTasks;
+        this.semaphore = new Semaphore(maxTasks);
+        this.keySequentialExecutor = new KeySequentialExecutor(underlyingExecutor);
+        this.acquire = onTasksExceeded == BLOCK ? this::blockOnTasksExceeded : this::rejectOnTasksExceeded;
+    }
+
+    private void blockOnTasksExceeded() {
+        semaphore.acquireUninterruptibly();
+    }
+
+    private void rejectOnTasksExceeded() {
+        if (!semaphore.tryAcquire()) {
+            throw new RejectedExecutionException("task limit of " + maxTasks + " exceeded");
+        }
     }
 
     @Override
     public void execute(Runnable task) {
-        boundedExecutor.execute(task);
+        checkNotNull(task);
+        synchronized (this) {
+            if (drained) {
+                throw new RejectedExecutionException("executor drained");
+            } else {
+                acquire.run();
+            }
+        }
+        try {
+            keySequentialExecutor.execute(new KeyRunnable<>(
+                    task,
+                    () -> {
+                        try {
+                            task.run();
+                        } finally {
+                            semaphore.release();
+                        }
+                    })
+            );
+        } catch (RejectedExecutionException e) {
+            semaphore.release();
+            throw e;
+        }
     }
 
     @Override
-    public boolean drain(long timeout, TimeUnit unit) throws InterruptedException {
-        return boundedExecutor.drain(timeout, unit);
+    public synchronized boolean drain(long timeout, TimeUnit unit) throws InterruptedException {
+        if (!drained && semaphore.tryAcquire(maxTasks, timeout, unit)) {
+            drained = true;
+        }
+        return drained;
     }
 }