AlertEntityImmediateSubscriber extends AbstractAlertEntitySubscriber { @Override protected void doProcessAlert(AlertEntity alert) { if(hasBeenSentOut(alert)) { logger.debug("[doProcessAlert]Alert has been sent out once: {}", alert); return; } logger.debug("[sendTaskBegin] {}", sendTaskBegin.get()); sendingAlerts.add(alert); if(sendTaskBegin.compareAndSet(false, true)) { logger.debug("send alert @{}, alert: {}", DateTimeUtils.currentTimeAsString(), alert); scheduleSendTask(); } }  @PostConstruct void initCleaner(); @VisibleForTesting Set<AlertEntity> getExistingAlerts(); @VisibleForTesting Set<AlertEntity> getSendingAlerts();  }
@Test public void doProcessAlert() { subscriber.doProcessAlert(alert); sleep(50); subscriber.doProcessAlert(newAlert(alert)); sleep(50); subscriber.doProcessAlert(newAlert(alert)); sleep(50); subscriber.doProcessAlert(newAlert(alert)); Set<AlertEntity> alertEntities = subscriber.getExistingAlerts(); Assert.assertEquals(1, alertEntities.size()); for(AlertEntity alertEntity : alertEntities) { logger.info("[AlertEntity] {}", alertEntity); Assert.assertTrue(System.currentTimeMillis() - alertEntity.getDate().getTime() < 50); } }
@Test public void doProcessAlert2() throws InterruptedException { policyManager.markCheckInterval(ALERT_TYPE.XREDIS_VERSION_NOT_VALID, ()->10); subscriber.doProcessAlert(alert); Thread.sleep(1500); logger.info("Sleep time up"); subscriber.doProcessAlert(alert); subscriber.doProcessAlert(alert); }
@Test public void doProcessAlert3() throws InterruptedException { policyManager.markCheckInterval(ALERT_TYPE.XREDIS_VERSION_NOT_VALID, ()->1000); subscriber.doProcessAlert(alert); Thread.sleep(1500); logger.info("Sleep time up"); subscriber.doProcessAlert(alert); subscriber.doProcessAlert(alert); }