TransactionSchedulerImpl implements TransactionScheduler, PriorityScheduler, ProfileListener { public void scheduleTask(KernelRunnable task, Identity owner) { backingQueue.addTask(new ScheduledTaskImpl.Builder( task, owner, defaultPriority).build()); } TransactionSchedulerImpl(Properties properties,
                             TransactionCoordinator transactionCoordinator,
                             ProfileCollectorHandle profileCollectorHandle,
                             AccessCoordinatorHandle accessCoordinator); TaskReservation reserveTask(KernelRunnable task, Identity owner); TaskReservation reserveTask(KernelRunnable task, Identity owner,
                                       long startTime); void scheduleTask(KernelRunnable task, Identity owner); void scheduleTask(KernelRunnable task, Identity owner,
                             long startTime); RecurringTaskHandle scheduleRecurringTask(KernelRunnable task,
                                                     Identity owner,
                                                     long startTime,
                                                     long period); TaskQueue createTaskQueue(); void runTask(KernelRunnable task, Identity owner); TaskReservation reserveTask(KernelRunnable task, Identity owner,
                                       Priority priority); void scheduleTask(KernelRunnable task, Identity owner,
                             Priority priority); void propertyChange(PropertyChangeEvent event); void report(ProfileReport profileReport); void shutdown(); static final String SCHEDULER_QUEUE_PROPERTY; static final String DEFAULT_SCHEDULER_QUEUE; static final String SCHEDULER_RETRY_PROPERTY; static final String DEFAULT_SCHEDULER_RETRY; static final String CONSUMER_THREADS_PROPERTY; static final String DEFAULT_CONSUMER_THREADS; }
@Test public void scheduleSingleTask() throws Exception { RunCountTestRunner runner = new RunCountTestRunner(1); txnScheduler.scheduleTask(runner, taskOwner); Thread.sleep(500L); assertEquals(0, runner.getRunCount()); }
@Test public void scheduleTransactionRetryAfterInterrupt() throws Exception { final AtomicInteger i = new AtomicInteger(0); final KernelRunnable r = new TestAbstractKernelRunnable() { public void run() throws Exception { if (i.getAndIncrement() == 0) throw new InterruptedException("test"); } }; txnScheduler.scheduleTask(r, taskOwner); Thread.sleep(200L); assertEquals(i.get(), 2); }