@Test public void testPurgingSystemCollections() throws Exception { RangedWorkNote workNote = RangedWorkNote.createSimpleWorkNote(BATCHJOBID); Exchange ex = Mockito.mock(Exchange.class); Message message = Mockito.mock(Message.class); Mockito.when(ex.getIn()).thenReturn(message); Mockito.when(message.getBody(RangedWorkNote.class)).thenReturn(workNote); NewBatchJob job = new NewBatchJob(); job.setProperty("tenantId", "SLI"); Mockito.when(mockBatchJobDAO.findBatchJobById(BATCHJOBID)).thenReturn(job); Set<String> collectionNames = new HashSet<String>(); collectionNames.add("system.js"); collectionNames.add("system.indexes"); Mockito.when(mongoTemplate.getCollectionNames()).thenReturn(collectionNames); purgeProcessor.process(ex); Mockito.verify(mongoTemplate, Mockito.never()).remove(Mockito.any(Query.class), Mockito.eq("system.js")); }
@Override public void process(Exchange exchange) throws Exception { Stage stage = Stage.createAndStartStage(BATCH_JOB_STAGE, BATCH_JOB_STAGE_DESC); String batchJobId = getBatchJobId(exchange); if (batchJobId != null) { reportStats = new SimpleReportStats(); NewBatchJob newJob = null; try { newJob = batchJobDAO.findBatchJobById(batchJobId); TenantContext.setTenantId(newJob.getTenantId()); String tenantId = newJob.getTenantId(); if (tenantId == null) { handleNoTenantId(batchJobId); } else { purgeForTenant(exchange, newJob, tenantId); } } catch (Exception exception) { handleProcessingExceptions(exchange, batchJobId, exception); } finally { if (newJob != null) { BatchJobUtils.stopStageAndAddToJob(stage, newJob); batchJobDAO.saveBatchJob(newJob); } } } else { missingBatchJobIdError(exchange); } }
PurgeProcessor implements Processor { @Override public void process(Exchange exchange) throws Exception { Stage stage = Stage.createAndStartStage(BATCH_JOB_STAGE, BATCH_JOB_STAGE_DESC); String batchJobId = getBatchJobId(exchange); if (batchJobId != null) { reportStats = new SimpleReportStats(); NewBatchJob newJob = null; try { newJob = batchJobDAO.findBatchJobById(batchJobId); TenantContext.setTenantId(newJob.getTenantId()); String tenantId = newJob.getTenantId(); if (tenantId == null) { handleNoTenantId(batchJobId); } else { purgeForTenant(exchange, newJob, tenantId); } } catch (Exception exception) { handleProcessingExceptions(exchange, batchJobId, exception); } finally { if (newJob != null) { BatchJobUtils.stopStageAndAddToJob(stage, newJob); batchJobDAO.saveBatchJob(newJob); } } } else { missingBatchJobIdError(exchange); } } }
PurgeProcessor implements Processor { @Override public void process(Exchange exchange) throws Exception { Stage stage = Stage.createAndStartStage(BATCH_JOB_STAGE, BATCH_JOB_STAGE_DESC); String batchJobId = getBatchJobId(exchange); if (batchJobId != null) { reportStats = new SimpleReportStats(); NewBatchJob newJob = null; try { newJob = batchJobDAO.findBatchJobById(batchJobId); TenantContext.setTenantId(newJob.getTenantId()); String tenantId = newJob.getTenantId(); if (tenantId == null) { handleNoTenantId(batchJobId); } else { purgeForTenant(exchange, newJob, tenantId); } } catch (Exception exception) { handleProcessingExceptions(exchange, batchJobId, exception); } finally { if (newJob != null) { BatchJobUtils.stopStageAndAddToJob(stage, newJob); batchJobDAO.saveBatchJob(newJob); } } } else { missingBatchJobIdError(exchange); } }  }
PurgeProcessor implements Processor { @Override public void process(Exchange exchange) throws Exception { Stage stage = Stage.createAndStartStage(BATCH_JOB_STAGE, BATCH_JOB_STAGE_DESC); String batchJobId = getBatchJobId(exchange); if (batchJobId != null) { reportStats = new SimpleReportStats(); NewBatchJob newJob = null; try { newJob = batchJobDAO.findBatchJobById(batchJobId); TenantContext.setTenantId(newJob.getTenantId()); String tenantId = newJob.getTenantId(); if (tenantId == null) { handleNoTenantId(batchJobId); } else { purgeForTenant(exchange, newJob, tenantId); } } catch (Exception exception) { handleProcessingExceptions(exchange, batchJobId, exception); } finally { if (newJob != null) { BatchJobUtils.stopStageAndAddToJob(stage, newJob); batchJobDAO.saveBatchJob(newJob); } } } else { missingBatchJobIdError(exchange); } }  @Override void process(Exchange exchange); MongoTemplate getMongoTemplate(); void setMongoTemplate(MongoTemplate mongoTemplate); BatchJobDAO getBatchJobDAO(); AbstractMessageReport getMessageReport(); boolean isSandboxEnabled(); void setBatchJobDAO(BatchJobDAO batchJobDAO); void setMessageReport(AbstractMessageReport messageReport); void setSandboxEnabled(boolean sandboxEnabled); void setPurgeBatchSize(int purgeBatchSize); void setDeltaJournal(DeltaJournal deltaJournal); void setDeltasEnabled(boolean deltasEnabled); }
PurgeProcessor implements Processor { @Override public void process(Exchange exchange) throws Exception { Stage stage = Stage.createAndStartStage(BATCH_JOB_STAGE, BATCH_JOB_STAGE_DESC); String batchJobId = getBatchJobId(exchange); if (batchJobId != null) { reportStats = new SimpleReportStats(); NewBatchJob newJob = null; try { newJob = batchJobDAO.findBatchJobById(batchJobId); TenantContext.setTenantId(newJob.getTenantId()); String tenantId = newJob.getTenantId(); if (tenantId == null) { handleNoTenantId(batchJobId); } else { purgeForTenant(exchange, newJob, tenantId); } } catch (Exception exception) { handleProcessingExceptions(exchange, batchJobId, exception); } finally { if (newJob != null) { BatchJobUtils.stopStageAndAddToJob(stage, newJob); batchJobDAO.saveBatchJob(newJob); } } } else { missingBatchJobIdError(exchange); } }  @Override void process(Exchange exchange); MongoTemplate getMongoTemplate(); void setMongoTemplate(MongoTemplate mongoTemplate); BatchJobDAO getBatchJobDAO(); AbstractMessageReport getMessageReport(); boolean isSandboxEnabled(); void setBatchJobDAO(BatchJobDAO batchJobDAO); void setMessageReport(AbstractMessageReport messageReport); void setSandboxEnabled(boolean sandboxEnabled); void setPurgeBatchSize(int purgeBatchSize); void setDeltaJournal(DeltaJournal deltaJournal); void setDeltasEnabled(boolean deltasEnabled); static final BatchJobStageType BATCH_JOB_STAGE; }