@Test public void testCmdletExecutor() throws InterruptedException { final List<StatusMessage> statusMessages = new ArrayList<>(); StatusReporter reporter = new StatusReporter() { @Override public void report(StatusMessage status) { statusMessages.add(status); } }; CmdletExecutor executor = new CmdletExecutor(new SmartConf()); StatusReportTask statusReportTask = new StatusReportTask(reporter, executor, new SmartConf()); ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(); executorService.scheduleAtFixedRate( statusReportTask, 100, 10, TimeUnit.MILLISECONDS); SmartAction action = new EchoAction(); Map<String, String> args = new HashMap<>(); args.put(EchoAction.PRINT_MESSAGE, "message success"); action.setArguments(args); action.setActionId(101); Cmdlet cmdlet = new Cmdlet(Arrays.asList(action)); executor.execute(cmdlet); Thread.sleep(2000); StatusReport lastReport = (StatusReport) statusMessages.get(statusMessages.size() - 1); ActionStatus status = lastReport.getActionStatuses().get(0); Assert.assertTrue(status.isFinished()); Assert.assertNull(status.getThrowable()); executor.shutdown(); }
public CmdletExecutor(SmartConf smartConf) { this.smartConf = smartConf; this.listenableFutures = new ConcurrentHashMap<>(); this.runningCmdlets = new ConcurrentHashMap<>(); this.idToReportCmdlet = new ConcurrentHashMap<>(); int nThreads = smartConf.getInt( SmartConfKeys.SMART_CMDLET_EXECUTORS_KEY, SmartConfKeys.SMART_CMDLET_EXECUTORS_DEFAULT); this.executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(nThreads)); }
CmdletExecutor { public CmdletExecutor(SmartConf smartConf) { this.smartConf = smartConf; this.listenableFutures = new ConcurrentHashMap<>(); this.runningCmdlets = new ConcurrentHashMap<>(); this.idToReportCmdlet = new ConcurrentHashMap<>(); int nThreads = smartConf.getInt( SmartConfKeys.SMART_CMDLET_EXECUTORS_KEY, SmartConfKeys.SMART_CMDLET_EXECUTORS_DEFAULT); this.executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(nThreads)); } }
CmdletExecutor { public CmdletExecutor(SmartConf smartConf) { this.smartConf = smartConf; this.listenableFutures = new ConcurrentHashMap<>(); this.runningCmdlets = new ConcurrentHashMap<>(); this.idToReportCmdlet = new ConcurrentHashMap<>(); int nThreads = smartConf.getInt( SmartConfKeys.SMART_CMDLET_EXECUTORS_KEY, SmartConfKeys.SMART_CMDLET_EXECUTORS_DEFAULT); this.executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(nThreads)); } CmdletExecutor(SmartConf smartConf); }
CmdletExecutor { public CmdletExecutor(SmartConf smartConf) { this.smartConf = smartConf; this.listenableFutures = new ConcurrentHashMap<>(); this.runningCmdlets = new ConcurrentHashMap<>(); this.idToReportCmdlet = new ConcurrentHashMap<>(); int nThreads = smartConf.getInt( SmartConfKeys.SMART_CMDLET_EXECUTORS_KEY, SmartConfKeys.SMART_CMDLET_EXECUTORS_DEFAULT); this.executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(nThreads)); } CmdletExecutor(SmartConf smartConf); void execute(Cmdlet cmdlet); void stop(Long cmdletId); void shutdown(); StatusReport getStatusReport(); }
CmdletExecutor { public CmdletExecutor(SmartConf smartConf) { this.smartConf = smartConf; this.listenableFutures = new ConcurrentHashMap<>(); this.runningCmdlets = new ConcurrentHashMap<>(); this.idToReportCmdlet = new ConcurrentHashMap<>(); int nThreads = smartConf.getInt( SmartConfKeys.SMART_CMDLET_EXECUTORS_KEY, SmartConfKeys.SMART_CMDLET_EXECUTORS_DEFAULT); this.executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(nThreads)); } CmdletExecutor(SmartConf smartConf); void execute(Cmdlet cmdlet); void stop(Long cmdletId); void shutdown(); StatusReport getStatusReport();  }