### Question:
RxJavaEventloopScheduler extends Scheduler { @Override public Worker createWorker() { final EventLoop eventLoop = eventLoopGroup.next(); return new EventloopWorker(eventLoop); } RxJavaEventloopScheduler(EventLoopGroup eventLoopGroup); @Override Worker createWorker();  }### Answer:
@Test(timeout = 60000) public void testRemoveNonDelayedTasks() throws Exception { RxJavaEventloopScheduler scheduler = new RxJavaEventloopScheduler(new NioEventLoopGroup()); final EventloopWorker worker = (EventloopWorker) scheduler.createWorker(); final EventloopWorker worker2 = (EventloopWorker) scheduler.createWorker(); assertThat("New worker already has subscriptions.", worker.hasScheduledSubscriptions(), is(false)); final AtomicBoolean isScheduledBeforeExecute = new AtomicBoolean(); final CountDownLatch executed = new CountDownLatch(1); Subscription subscription = worker.schedule(new Action0() { @Override public void call() { isScheduledBeforeExecute.set(worker.hasScheduledSubscriptions()); worker2.schedule(new Action0() { @Override public void call() { executed.countDown(); } }); } }); executed.await(); assertThat("No scheduled subscriptions on executing the action.", isScheduledBeforeExecute.get(), is(true)); assertThat("Action not unsubscribed.", subscription.isUnsubscribed(), is(true)); assertThat("Subscription not removed post execution.", worker.hasScheduledSubscriptions(), is(false)); }
@Test(timeout = 60000) public void testRemoveDelayedTasks() throws Exception { RxJavaEventloopScheduler scheduler = new RxJavaEventloopScheduler(new NioEventLoopGroup()); final EventloopWorker worker = (EventloopWorker) scheduler.createWorker(); final EventloopWorker worker2 = (EventloopWorker) scheduler.createWorker(); assertThat("New worker already has subscriptions.", worker.hasDelayScheduledSubscriptions(), is(false)); final AtomicBoolean isScheduledBeforeExecute = new AtomicBoolean(); final CountDownLatch executed = new CountDownLatch(1); Subscription subscription = worker.schedule(new Action0() { @Override public void call() { isScheduledBeforeExecute.set(worker.hasDelayScheduledSubscriptions()); worker2.schedule(new Action0() { @Override public void call() { executed.countDown(); } }); } }, 1, TimeUnit.MILLISECONDS); executed.await(); assertThat("No scheduled subscriptions on executing the action.", isScheduledBeforeExecute.get(), is(true)); assertThat("Action not unsubscribed.", subscription.isUnsubscribed(), is(true)); assertThat("Subscription not removed post execution.", worker.hasDelayScheduledSubscriptions(), is(false)); }
@Test(timeout = 60000) public void testUnsubscribeDelayedTasks() throws Exception { RxJavaEventloopScheduler scheduler = new RxJavaEventloopScheduler(new NioEventLoopGroup()); final EventloopWorker worker = (EventloopWorker) scheduler.createWorker(); assertThat("New worker already has subscriptions.", worker.hasDelayScheduledSubscriptions(), is(false)); Subscription subscription = worker.schedule(new Action0() { @Override public void call() { } }, 100, TimeUnit.DAYS); assertThat("No subscriptions post schedule.", worker.hasDelayScheduledSubscriptions(), is(true)); subscription.unsubscribe(); assertThat("Subscription not removed post cancellation.", worker.hasDelayScheduledSubscriptions(), is(false)); }