A little progress towards Issue #50. This combines SimpleSchedulerLim…

…iter and SubmitterSchedulerLimiter into one (simpler) class called "SchedulerLimiter".

This reduced a lot of code, as well as provided a MUCH simpler simplementation compared to what SubmitterSchedulerLimiter used to be.  This change also exposed a defect in PriorityScheduledExecutor with removal of callables which are wrapped in runnables.  This commit fixes that in order to fix the unit tests.

We still need to look at what we are going to do for the PrioritySchedulerLimiter, but I may wait on that for now, since those changes will be more dramatic (and possibly more risky).  It was important for me to do these before 1.0.0 release because these changes break the API/interface.

src/main/java/org/threadly/concurrent/PriorityScheduledExecutor.java

@@ -661,6 +661,10 @@ protected static boolean isContained(Runnable startRunnable,
           return false;
         }
       }
+    } else if (startRunnable instanceof RunnableContainerInterface) {
+      RunnableContainerInterface rci = (RunnableContainerInterface)startRunnable;
+
+      return isContained(rci.getContainedRunnable(), compareTo);
     } else {
       return false;
     }

src/main/java/org/threadly/concurrent/limiter/SimpleSchedulerLimiter.java → src/main/java/org/threadly/concurrent/limiter/SchedulerLimiter.java

@@ -1,8 +1,13 @@
 package org.threadly.concurrent.limiter;
 
+import java.util.concurrent.Callable;
+
 import org.threadly.concurrent.RunnableContainerInterface;
 import org.threadly.concurrent.SimpleSchedulerInterface;
+import org.threadly.concurrent.SubmitterSchedulerInterface;
 import org.threadly.concurrent.VirtualRunnable;
+import org.threadly.concurrent.future.ListenableFuture;
+import org.threadly.concurrent.future.ListenableFutureTask;
 
 /**
  * <p>This class is designed to limit how much parallel execution happens 
@@ -21,8 +26,8 @@
  * 
  * @author jent - Mike Jensen
  */
-public class SimpleSchedulerLimiter extends ExecutorLimiter 
-                                    implements SimpleSchedulerInterface {
+public class SchedulerLimiter extends ExecutorLimiter 
+                              implements SubmitterSchedulerInterface {
   protected final SimpleSchedulerInterface scheduler;
 
   /**
@@ -31,8 +36,8 @@ public class SimpleSchedulerLimiter extends ExecutorLimiter
    * @param scheduler {@link SimpleSchedulerInterface} implementation to submit task executions to.
    * @param maxConcurrency maximum qty of runnables to run in parallel
    */
-  public SimpleSchedulerLimiter(SimpleSchedulerInterface scheduler, 
-                                int maxConcurrency) {
+  public SchedulerLimiter(SimpleSchedulerInterface scheduler, 
+                          int maxConcurrency) {
     this(scheduler, maxConcurrency, null);
   }
 
@@ -43,8 +48,8 @@ public SimpleSchedulerLimiter(SimpleSchedulerInterface scheduler,
    * @param maxConcurrency maximum qty of runnables to run in parallel
    * @param subPoolName name to describe threads while tasks running in pool (null to not change thread names)
    */
-  public SimpleSchedulerLimiter(SimpleSchedulerInterface scheduler, 
-                                int maxConcurrency, String subPoolName) {
+  public SchedulerLimiter(SimpleSchedulerInterface scheduler, 
+                          int maxConcurrency, String subPoolName) {
     super(scheduler, maxConcurrency, subPoolName);
 
     this.scheduler = scheduler;
@@ -55,6 +60,45 @@ public boolean isShutdown() {
     return scheduler.isShutdown();
   }
 
+  @Override
+  public ListenableFuture<?> submitScheduled(Runnable task, long delayInMs) {
+    if (task == null) {
+      throw new IllegalArgumentException("Must provide task");
+    }
+
+    ListenableFutureTask<?> ft = new ListenableFutureTask<Object>(false, task);
+
+    schedule(ft, delayInMs);
+
+    return ft;
+  }
+
+  @Override
+  public <T> ListenableFuture<T> submitScheduled(Runnable task, T result, long delayInMs) {
+    if (task == null) {
+      throw new IllegalArgumentException("Must provide task");
+    }
+
+    ListenableFutureTask<T> ft = new ListenableFutureTask<T>(false, task, result);
+
+    schedule(ft, delayInMs);
+
+    return ft;
+  }
+
+  @Override
+  public <T> ListenableFuture<T> submitScheduled(Callable<T> task, long delayInMs) {
+    if (task == null) {
+      throw new IllegalArgumentException("Must provide task");
+    }
+
+    ListenableFutureTask<T> ft = new ListenableFutureTask<T>(false, task);
+
+    schedule(ft, delayInMs);
+
+    return ft;
+  }
+
   @Override
   public void schedule(Runnable task, long delayInMs) {
     if (task == null) {