OneThreadTaskExecutor implements Destroyable { public void executeCommand(Command<?> command) { logger.debug("[executeCommand][offer it in pool]{}", command); boolean offer = false; synchronized (this) { offer = tasks.offer(command); } if (!offer) { throw new IllegalStateException("pool full:" + tasks.size()); } doExecute(); } OneThreadTaskExecutor(Executor executors);  OneThreadTaskExecutor(RetryCommandFactory<?> retryCommandFactory, Executor executors); void executeCommand(Command<?> command); @Override void destroy();  }
@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 AbstractCommand<Object>() { @Override protected void doExecute() throws Exception { 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 testStartMtimes() { int times = 50; CommandFuture<Void> future = new DefaultCommandFuture<>(); when(command.execute()).thenReturn(future); future.setSuccess(); for (int i = 0; i < times; i++) { oneThreadTaskExecutor.executeCommand(command); sleep(30); verify(command, times(i + 1)).execute(); } }
@Test public void testRetryTemplate() throws TimeoutException { int retryTimes = 10; CommandFuture<Void> future = new DefaultCommandFuture<>(); when(command.execute()).thenReturn(future); future.setFailure(new Exception()); RetryCommandFactory retryCommandFactory = DefaultRetryCommandFactory.retryNTimes(scheduled, retryTimes, 10); OneThreadTaskExecutor oneThreadTaskExecutor = new OneThreadTaskExecutor(retryCommandFactory, executors); oneThreadTaskExecutor.executeCommand(command); waitConditionUntilTimeOut(() -> { try { verify(command, times(retryTimes + 1)).execute(); return true; } catch (Throwable e) { } return false; }, 2000); }