FlowPurgerMBean implements InitializingBean, DisposableBean { @ManagedOperation(description = "Executes a purge job once for current application") public void execute() { executeJob(flowManager.getApplicationConfig(application)); } FlowPurgerMBean(); @ManagedAttribute(description="Application name") String getApplication(); @ManagedAttribute(description="Application name") void setApplication(String application); @ManagedAttribute(description="Purge flows older than given duration " + "(e.g. \"30d\" will purge flows older than 30 days)") String getPurgeFlowsOlderThan(); @ManagedAttribute(description="Set to true to exclude ERROR flows from being purged. " + "Set to false to purge CLEAN and ERROR flows") boolean isDoNotPurgeErrorFlows(); @ManagedAttribute(description="Cron expression for purge schedule") String getPurgeSchedule(); @ManagedAttribute(description="Purge job status for current application") boolean isPurgeScheduled(); @ManagedAttribute(description="Purge flows older than given duration " + "(e.g. \"30d\" will purge flows older than 30 days)") void setPurgeFlowsOlderThan(String purgeFlowsOlderThan); @ManagedAttribute(description="Set to true to exclude ERROR flows from being purged. " + "Set to false to purge CLEAN and ERROR flows") void setDoNotPurgeErrorFlows(boolean doNotPurgeErrorFlows); @ManagedAttribute(description="Cron expression for purge schedule") void setPurgeSchedule(String purgeSchedule); @ManagedOperation(description = "Executes a purge job once for current application") void execute(); @ManagedOperation(description = "Schedules or reschedules a purge job for current application and schedule") void schedule(); @ManagedOperation(description = "Unschedules a purge job for current application and schedule") void unschedule(); @Override void afterPropertiesSet(); @Override void destroy(); Scheduler getScheduler(); void setScheduler(Scheduler scheduler); @ManagedAttribute(description="Schedulers metadata") Map<String, String> getSchedulerMetaData();  }
@Test public void testExecute() throws Exception { persistFlowHistory(); assertEquals(6, flowRepository.findFlows(allFlows).size()); flowPurgerMBean.getScheduler().getListenerManager().addJobListener(new PurgeJobListener(latch)); flowPurgerMBean.setDoNotPurgeErrorFlows(false); flowPurgerMBean.setPurgeFlowsOlderThan("9d"); flowPurgerMBean.execute(); latch.await(WAIT_TIMEOUT, TimeUnit.SECONDS); assertEquals(1, flowRepository.findFlows(allFlows).size()); flowPurgerMBean.unschedule(); }