MutexableOneThreadTaskExecutor extends OneThreadTaskExecutor { @Override public void executeCommand(Command<?> command) { if (command instanceof RequestResponseCommand) { super.executeCommand(command); } else { throw new IllegalArgumentException("Only timeout enabled command is allowed"); } } MutexableOneThreadTaskExecutor(Executor executors, ScheduledExecutorService scheduled); @Override void executeCommand(Command<?> command); @SuppressWarnings("unchecked") void clearAndExecuteCommand(RequestResponseCommand<?> command);  }
@Test public void testStart() { when(command.future()).thenReturn(new DefaultCommandFuture<>()); oneThreadTaskExecutor.executeCommand(command); sleep(50); verify(command).execute(); }
@Test public void testSequence() throws InterruptedException { int count = 100; Queue<Integer> list = new ConcurrentLinkedQueue<>(); CountDownLatch latch = new CountDownLatch(count); for (int i = 0; i < count; i++) { int finalI = i; oneThreadTaskExecutor.executeCommand(new AbstractRequestResponseCommand<Object>() { @Override public int getCommandTimeoutMilli() { return 10; } @Override protected void doExecute() { try{ logger.debug("{}", this); list.offer(finalI); future().setSuccess(); }finally { latch.countDown(); } } @Override protected void doReset() { } @Override public String getName() { return getTestName() + ":" + finalI; } }); } Assert.assertTrue(latch.await(5, TimeUnit.SECONDS)); Assert.assertEquals(count, list.size()); int previous = -1; while (true){ Integer current = list.poll(); if(current == null){ break; } Assert.assertTrue(current > previous); previous = current; } }
@Test public void testClose() throws Exception { CommandFuture<Void> future = new DefaultCommandFuture<>(); when(command.execute()).thenReturn(future); future.setFailure(new Exception()); RetryCommandFactory<Void> retryCommandFactory = DefaultRetryCommandFactory.retryForever(scheduled, 0); OneThreadTaskExecutor oneThreadTaskExecutor = new OneThreadTaskExecutor(retryCommandFactory, executors); oneThreadTaskExecutor.executeCommand(command); sleep(100); logger.info("[testClose][destroy]"); oneThreadTaskExecutor.destroy(); retryCommandFactory.destroy(); sleep(50); verify(command, new AtLeast(1)).execute(); verify(command, new AtLeast(1)).reset(); logger.info("[testClose][sleep verify no more interactions]"); sleep(100); verifyNoMoreInteractions(command); }