Merge pull request #2 from jano7/v2.0

V2.0

Author: jano7

README.md

@@ -1,7 +1,7 @@
 # Key Sequential Executor
 This small library provides an optimized solution to a problem where tasks for a particular key need to be processed
 sequentially as they arrive. This kind of problem can be solved by a [`SingleThreadExecutor`](
-https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor--), however it is
+https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor--); however, it is
 not efficient. The issue is that the tasks for unrelated keys are not being processed in parallel, instead they are put
 into a queue common to all keys and wait for the single thread to execute them. This library allows them to be executed
 concurrently. Moreover this library works well in a situation where all the possible keys and their number is **not**
@@ -10,7 +10,7 @@ known upfront.
 A typical scenario in order management or booking systems is that messages for a particular trade **A** must be
 processed sequentially in the same order as they are received (otherwise the state of the trade will be incorrect). The
 same is true for any other trade - for example messages for the trade **B** must be processed sequentially as well.
-However it is desirable that a message for the trade **A** does not block processing of a message for the trade **B**
+However, it is desirable that a message for the trade **A** does not block processing of a message for the trade **B**
 (and vice versa) if they happen to arrive at the same time.
 ```java
 ExecutorService underlyingExecutor = Executors.newFixedThreadPool(10);
@@ -33,7 +33,7 @@ runner.run(tradeIdB, task); // execution is not blocked by the task for tradeIdA
 
 runner.run(tradeIdA, task); // execution starts when the previous task for tradeIdA completes
 ```
-In the example above the key is a Trade ID. Tasks for a particular Trade ID are executed sequentially but they do not
+In the example above the key is a Trade ID. Tasks for a particular Trade ID are executed sequentially, but they do not
 block tasks for other Trade IDs (unless the tasks are blocked by the underlying executor).
 
 Please note the Key needs to correctly implement `hashCode` and `equals` methods as the implementation stores the tasks
@@ -47,32 +47,42 @@ key.
 ```java
 Executor executor = new KeySequentialExecutor(underlyingExecutor);
 
-Runnable runnable = new KeyRunnable<>(tradeIdA, task); // helper class delegating hashCode and equals to the key
+Runnable runnable =
+        new KeyRunnable<>(tradeIdA, task); // helper class delegating 'hashCode' and 'equals' to the key
 
 executor.execute(runnable);
-```
 
+underlyingExecutor.shutdown();
+
+// at this point, tasks for new keys will be rejected; however, tasks for keys being currently executed may
+// still be accepted (and executed)
+
+underlyingExecutor.awaitTermination(timeout, TimeUnit.SECONDS);
+
+// if the executor terminates before a timeout, then it is guaranteed all accepted tasks have been executed
+```
 The `KeySequentialExecutor` and `KeySequentialRunner` do not support back-pressure. It means that `execute` and `run`
 methods never block, instead the submitted tasks are put into a queue where they wait until executed by the underlying
-executor. In many cases this is not a problem, however in some situations it may cause an application to run out of
+executor. In many cases this is not a problem, but in some situations it may cause an application to run out of
 memory as the number of waiting tasks grows. If you want to restrict the number of queued tasks, consider use of a
-[`KeySequentialBoundedExecutor`](src/main/java/com/jano7/executor/KeySequentialBoundedExecutor.java) which blocks the
-task submission when the number of tasks, which haven't been executed yet, hits the limit.
+[`KeySequentialBoundedExecutor`](src/main/java/com/jano7/executor/KeySequentialBoundedExecutor.java) which can be
+configured to block the task submission when the number of tasks, which haven't been executed yet, reaches the limit.
 ```java
 ExecutorService underlyingExecutor = Executors.newCachedThreadPool();
 int maxTasks = 10;
-KeySequentialBoundedExecutor boundedExecutor = new KeySequentialBoundedExecutor(maxTasks, underlyingExecutor);
+KeySequentialBoundedExecutor boundedExecutor =
+        new KeySequentialBoundedExecutor(maxTasks, BoundedStrategy.BLOCK, underlyingExecutor);
 
-KeyRunnable<String> aTask = new KeyRunnable<>("my key", () -> {
+KeyRunnable<String> task = new KeyRunnable<>("my key", () -> {
     // do something
 });
 
-boundedExecutor.execute(aTask);
+boundedExecutor.execute(task);
 
 // execute more tasks ... at most 10 will be scheduled
 
 // before shutting down you can call a 'drain' method which blocks until all submitted task have been executed
-boundedExecutor.drain(aTimeout, TimeUnit.SECONDS); // returns true if drained; false if the timeout elapses
+boundedExecutor.drain(timeout, TimeUnit.SECONDS); // returns true if drained; false if the timeout elapses
 
 // newly submitted tasks will be rejected after calling 'drain'
 
@@ -81,12 +91,12 @@ underlyingExecutor.shutdownNow(); // safe to call 'shutdownNow' if drained as th
 The source code of the examples can be found [here](src/test/java/com/jano7/executor/Examples.java).
 
 A note on **thread-safety**: The library is thread-safe; i.e. methods `run`, `execute` or `drain` can be safely invoked
-from multiple threads without synchronization. 
+from multiple threads without synchronization.
 ## Maven Dependency
 ```xml
 <dependency>
   <groupId>com.jano7</groupId>
   <artifactId>executor</artifactId>
-  <version>1.0.6</version>
+  <version>2.0.0</version>
 </dependency>
 ```

pom.xml

@@ -3,7 +3,7 @@
 
     <groupId>com.jano7</groupId>
     <artifactId>executor</artifactId>
-    <version>1.0.7-SNAPSHOT</version>
+    <version>2.0.0-SNAPSHOT</version>
 
     <name>Java Key Sequential Executor</name>
     <description>This small library provides an optimized solution to a problem where tasks for a particular key need to
@@ -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-1.0.6</tag>
+        <tag>executor-2.0.0</tag>
     </scm>
 
 </project>

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

@@ -0,0 +1,97 @@
+/*
+MIT License
+
+Copyright (c) 2020 Jan Gaspar
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+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 BoundedExecutor implements DrainableExecutor {
+
+    private final int maxTasks;
+
+    private final Semaphore semaphore;
+
+    private final Executor underlyingExecutor;
+
+    private final Runnable acquire;
+
+    private boolean drained = false;
+
+    public BoundedExecutor(int maxTasks, BoundedStrategy onTasksExceeded, Executor underlyingExecutor) {
+        this.maxTasks = maxTasks;
+        this.semaphore = new Semaphore(maxTasks);
+        this.underlyingExecutor = 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) {
+        checkNotNull(task);
+        synchronized (this) {
+            if (drained) {
+                throw new RejectedExecutionException("executor drained");
+            } else {
+                acquire.run();
+            }
+        }
+        try {
+            underlyingExecutor.execute(new KeyRunnable<>(
+                    task,
+                    () -> {
+                        try {
+                            task.run();
+                        } finally {
+                            semaphore.release();
+                        }
+                    })
+            );
+        } catch (RejectedExecutionException e) {
+            semaphore.release();
+            throw e;
+        }
+    }
+
+    @Override
+    public synchronized boolean drain(long timeout, TimeUnit unit) throws InterruptedException {
+        if (!drained && semaphore.tryAcquire(maxTasks, timeout, unit)) {
+            drained = true;
+        }
+        return drained;
+    }
+}

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

@@ -0,0 +1,28 @@
+/*
+MIT License
+
+Copyright (c) 2020 Jan Gaspar
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+package com.jano7.executor;
+
+public enum BoundedStrategy {
+    REJECT, BLOCK
+}

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

@@ -0,0 +1,32 @@
+/*
+MIT License
+
+Copyright (c) 2020 Jan Gaspar
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+package com.jano7.executor;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+
+public interface DrainableExecutor extends Executor {
+
+    boolean drain(long timeout, TimeUnit unit) throws InterruptedException;
+}

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

@@ -38,8 +38,12 @@ public KeyRunnable(Key key, Runnable runnable) {
 
     @Override
     public boolean equals(Object o) {
-        if (this == o) return true;
-        if (o == null || getClass() != o.getClass()) return false;
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
         KeyRunnable<?> that = (KeyRunnable<?>) o;
         return Objects.equals(key, that.key);
     }
@@ -49,6 +53,11 @@ public int hashCode() {
         return Objects.hashCode(key);
     }
 
+    @Override
+    public String toString() {
+        return Objects.toString(key);
+    }
+
     @Override
     public void run() {
         runnable.run();

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

@@ -24,61 +24,23 @@ 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;
 
-public final class KeySequentialBoundedExecutor implements Executor {
+public final class KeySequentialBoundedExecutor implements DrainableExecutor {
 
-    private final int maxTasks;
+    private final BoundedExecutor boundedExecutor;
 
-    private final Semaphore semaphore;
-
-    private final KeySequentialExecutor executor;
-
-    private boolean drained = false;
-
-    public KeySequentialBoundedExecutor(int maxTasks, KeySequentialExecutor executor) {
-        this.maxTasks = maxTasks;
-        this.semaphore = new Semaphore(maxTasks);
-        this.executor = executor;
-    }
-
-    public KeySequentialBoundedExecutor(int maxTasks, Executor underlyingExecutor) {
-        this(maxTasks, new KeySequentialExecutor(underlyingExecutor));
+    public KeySequentialBoundedExecutor(int maxTasks, BoundedStrategy onTasksExceeded, Executor underlyingExecutor) {
+        boundedExecutor = new BoundedExecutor(maxTasks, onTasksExceeded, new KeySequentialExecutor(underlyingExecutor));
     }
 
     @Override
     public void execute(Runnable task) {
-        if (task == null) throw new NullPointerException("Task is null");
-        synchronized (this) {
-            if (drained) {
-                throw new RejectedExecutionException(getClass().getSimpleName() + " drained");
-            } else {
-                semaphore.acquireUninterruptibly();
-            }
-        }
-        try {
-            executor.execute(new KeyRunnable<>(
-                    task,
-                    () -> {
-                        try {
-                            task.run();
-                        } finally {
-                            semaphore.release();
-                        }
-                    })
-            );
-        } catch (RejectedExecutionException e) {
-            semaphore.release();
-            throw e;
-        }
+        boundedExecutor.execute(task);
     }
 
-    public synchronized boolean drain(long timeout, TimeUnit unit) throws InterruptedException {
-        if (!drained && semaphore.tryAcquire(maxTasks, timeout, unit)) {
-            drained = true;
-        }
-        return drained;
+    @Override
+    public boolean drain(long timeout, TimeUnit unit) throws InterruptedException {
+        return boundedExecutor.drain(timeout, unit);
     }
 }

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

@@ -30,16 +30,15 @@ public final class KeySequentialExecutor implements Executor {
     private final KeySequentialRunner<Runnable> runner;
 
     public KeySequentialExecutor(Executor underlyingExecutor) {
-        this.runner = new KeySequentialRunner<>(underlyingExecutor);
+        runner = new KeySequentialRunner<>(underlyingExecutor);
     }
 
-    public KeySequentialExecutor(Executor underlyingExecutor, TaskExceptionHandler exceptionHandler) {
-        this.runner = new KeySequentialRunner<>(underlyingExecutor, exceptionHandler);
+    public KeySequentialExecutor(Executor underlyingExecutor, TaskExceptionHandler<Runnable> exceptionHandler) {
+        runner = new KeySequentialRunner<>(underlyingExecutor, exceptionHandler);
     }
 
     @Override
     public void execute(Runnable task) {
-        if (task == null) throw new NullPointerException("Task is null");
         runner.run(task, task);
     }
 }