SingletonTask { public void reschedule(long delay, TimeUnit unit) { boolean needQueue = true; SingletonTaskWorker stw = null; synchronized (context) { if (context.taskRunning || context.taskShouldRun) { if (context.taskRunning) { if (delay > 0) { long now = System.nanoTime(); long then = now + TimeUnit.NANOSECONDS.convert(delay, unit); context.waitingTask.nextschedule = then; logger.debug("rescheduled task " + this + " for " + TimeUnit.SECONDS.convert(then, TimeUnit.NANOSECONDS) + "s. A bunch of these messages -may- indicate you have a blocked task."); } else { context.waitingTask.nextschedule = 0; } needQueue = false; } else { context.waitingTask.canceled = true; context.waitingTask = null; } } context.taskShouldRun = true; if (needQueue) { stw = context.waitingTask = new SingletonTaskWorker(this); } } if (needQueue) { if (delay <= 0) { ses.execute(stw); } else { ses.schedule(stw, delay, unit); } } } SingletonTask(ScheduledExecutorService ses,
            Runnable task); void reschedule(long delay, TimeUnit unit);  }
@Test public void testBasic() throws InterruptedException { ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor(); SingletonTask st1 = new SingletonTask(ses, new Runnable() { @Override public void run() { ran += 1; } }); st1.reschedule(0, null); ses.shutdown(); ses.awaitTermination(5, TimeUnit.SECONDS); assertEquals("Check that task ran", 1, ran); }
@Test public void testDelay() throws InterruptedException { ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor(); SingletonTask st1 = new SingletonTask(ses, new Runnable() { @Override public void run() { ran += 1; time = System.nanoTime(); } }); st1.reschedule(10, TimeUnit.MILLISECONDS); assertFalse("Check that task hasn't run yet", ran > 0); ses.shutdown(); ses.awaitTermination(5, TimeUnit.SECONDS); assertEquals("Check that task ran", 1, ran); }
@Test public void testReschedule() throws InterruptedException { ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor(); final Object tc = this; SingletonTask st1 = new SingletonTask(ses, new Runnable() { @Override public void run() { synchronized (tc) { ran += 1; } time = System.nanoTime(); } }); st1.reschedule(20, TimeUnit.MILLISECONDS); Thread.sleep(5); assertFalse("Check that task hasn't run yet", ran > 0); st1.reschedule(20, TimeUnit.MILLISECONDS); Thread.sleep(5); assertFalse("Check that task hasn't run yet", ran > 0); st1.reschedule(20, TimeUnit.MILLISECONDS); Thread.sleep(5); assertFalse("Check that task hasn't run yet", ran > 0); st1.reschedule(20, TimeUnit.MILLISECONDS); Thread.sleep(5); assertFalse("Check that task hasn't run yet", ran > 0); st1.reschedule(20, TimeUnit.MILLISECONDS); Thread.sleep(5); assertFalse("Check that task hasn't run yet", ran > 0); st1.reschedule(20, TimeUnit.MILLISECONDS); Thread.sleep(5); assertFalse("Check that task hasn't run yet", ran > 0); st1.reschedule(20, TimeUnit.MILLISECONDS); Thread.sleep(5); assertFalse("Check that task hasn't run yet", ran > 0); st1.reschedule(20, TimeUnit.MILLISECONDS); Thread.sleep(5); assertFalse("Check that task hasn't run yet", ran > 0); ses.shutdown(); ses.awaitTermination(5, TimeUnit.SECONDS); assertEquals("Check that task ran only once", 1, ran); }
@Test public void testConcurrentAddDelay() throws InterruptedException { ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor(); final Object tc = this; SingletonTask st1 = new SingletonTask(ses, new Runnable() { @Override public void run() { synchronized (tc) { ran += 1; } try { Thread.sleep(50); } catch (InterruptedException e) { e.printStackTrace(); } synchronized (tc) { finished += 1; time = System.nanoTime(); } } }); st1.reschedule(5, TimeUnit.MILLISECONDS); Thread.sleep(20); assertEquals("Check that task started", 1, ran); assertEquals("Check that task not finished", 0, finished); st1.reschedule(75, TimeUnit.MILLISECONDS); assertTrue("Check task running state true", st1.context.taskRunning); assertTrue("Check task should run state true", st1.context.taskShouldRun); assertEquals("Check that task started", 1, ran); assertEquals("Check that task not finished", 0, finished); Thread.sleep(150); assertTrue("Check task running state false", !st1.context.taskRunning); assertTrue("Check task should run state false", !st1.context.taskShouldRun); assertEquals("Check that task ran exactly twice", 2, ran); assertEquals("Check that task finished exactly twice", 2, finished); ses.shutdown(); ses.awaitTermination(15, TimeUnit.SECONDS); }
@Test public void testConcurrentAddDelay2() throws InterruptedException { ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor(); final Object tc = this; SingletonTask st1 = new SingletonTask(ses, new Runnable() { @Override public void run() { synchronized (tc) { ran += 1; } try { Thread.sleep(50); } catch (InterruptedException e) { e.printStackTrace(); } synchronized (tc) { finished += 1; time = System.nanoTime(); } } }); st1.reschedule(5, TimeUnit.MILLISECONDS); Thread.sleep(20); assertEquals("Check that task started", 1, ran); assertEquals("Check that task not finished", 0, finished); st1.reschedule(25, TimeUnit.MILLISECONDS); assertTrue("Check task running state true", st1.context.taskRunning); assertTrue("Check task should run state true", st1.context.taskShouldRun); assertEquals("Check that task started", 1, ran); assertEquals("Check that task not finished", 0, finished); Thread.sleep(150); assertTrue("Check task running state false", !st1.context.taskRunning); assertTrue("Check task should run state false", !st1.context.taskShouldRun); assertEquals("Check that task ran exactly twice", 2, ran); assertEquals("Check that task finished exactly twice", 2, finished); ses.shutdown(); ses.awaitTermination(5, TimeUnit.SECONDS); }
@Test public void testConcurrentAddNoDelay() throws InterruptedException { ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor(); final Object tc = this; SingletonTask st1 = new SingletonTask(ses, new Runnable() { @Override public void run() { synchronized (tc) { ran += 1; } try { Thread.sleep(50); } catch (InterruptedException e) { e.printStackTrace(); } synchronized (tc) { finished += 1; time = System.nanoTime(); } } }); st1.reschedule(0, null); Thread.sleep(20); assertEquals("Check that task started", 1, ran); assertEquals("Check that task not finished", 0, finished); st1.reschedule(0, null); assertTrue("Check task running state true", st1.context.taskRunning); assertTrue("Check task should run state true", st1.context.taskShouldRun); assertEquals("Check that task started", 1, ran); assertEquals("Check that task not finished", 0, finished); Thread.sleep(150); assertTrue("Check task running state false", !st1.context.taskRunning); assertTrue("Check task should run state false", !st1.context.taskShouldRun); assertEquals("Check that task ran exactly twice", 2, ran); assertEquals("Check that task finished exactly twice", 2, finished); ses.shutdown(); ses.awaitTermination(5, TimeUnit.SECONDS); }