@Test public void no_operations_means_no_PatchSet() throws Exception { given(adminClient.updateMessageStatus(testRoad)).willReturn(emptyList()); messageStatusUpdator.updateMessageStatusInModel(); then(emitter).shouldHaveZeroInteractions(); }
@Scheduled(initialDelayString = "${messageStatus.initialdelay:PT60m}", fixedRateString = "${messageStatus.fixedrate:PT60m}") public void updateMessageStatusInModel() { Map<String, KafkaRoad> store = new HashMap<>(this.store); store.forEach((key, model) -> { try { List<PatchOperation> operations = adminClient.updateMessageStatus(model); if (!operations.isEmpty()) { emitter.emit(new PatchSet(key, operations)); } } catch (Exception e) { log.warn("Problem updating MessageStatus model \"{}\"", key, e); } }); }
UpdateModel { @Scheduled(initialDelayString = "${messageStatus.initialdelay:PT60m}", fixedRateString = "${messageStatus.fixedrate:PT60m}") public void updateMessageStatusInModel() { Map<String, KafkaRoad> store = new HashMap<>(this.store); store.forEach((key, model) -> { try { List<PatchOperation> operations = adminClient.updateMessageStatus(model); if (!operations.isEmpty()) { emitter.emit(new PatchSet(key, operations)); } } catch (Exception e) { log.warn("Problem updating MessageStatus model \"{}\"", key, e); } }); } }
UpdateModel { @Scheduled(initialDelayString = "${messageStatus.initialdelay:PT60m}", fixedRateString = "${messageStatus.fixedrate:PT60m}") public void updateMessageStatusInModel() { Map<String, KafkaRoad> store = new HashMap<>(this.store); store.forEach((key, model) -> { try { List<PatchOperation> operations = adminClient.updateMessageStatus(model); if (!operations.isEmpty()) { emitter.emit(new PatchSet(key, operations)); } } catch (Exception e) { log.warn("Problem updating MessageStatus model \"{}\"", key, e); } }); } @Autowired  UpdateModel(@Value("#{store}") Map<String, KafkaRoad> store, PatchSetEmitter emitter, KafkaAdminClient adminClient); }
UpdateModel { @Scheduled(initialDelayString = "${messageStatus.initialdelay:PT60m}", fixedRateString = "${messageStatus.fixedrate:PT60m}") public void updateMessageStatusInModel() { Map<String, KafkaRoad> store = new HashMap<>(this.store); store.forEach((key, model) -> { try { List<PatchOperation> operations = adminClient.updateMessageStatus(model); if (!operations.isEmpty()) { emitter.emit(new PatchSet(key, operations)); } } catch (Exception e) { log.warn("Problem updating MessageStatus model \"{}\"", key, e); } }); } @Autowired  UpdateModel(@Value("#{store}") Map<String, KafkaRoad> store, PatchSetEmitter emitter, KafkaAdminClient adminClient); @Scheduled(initialDelayString = "${messageStatus.initialdelay:PT60m}", fixedRateString = "${messageStatus.fixedrate:PT60m}") void updateMessageStatusInModel(); }
UpdateModel { @Scheduled(initialDelayString = "${messageStatus.initialdelay:PT60m}", fixedRateString = "${messageStatus.fixedrate:PT60m}") public void updateMessageStatusInModel() { Map<String, KafkaRoad> store = new HashMap<>(this.store); store.forEach((key, model) -> { try { List<PatchOperation> operations = adminClient.updateMessageStatus(model); if (!operations.isEmpty()) { emitter.emit(new PatchSet(key, operations)); } } catch (Exception e) { log.warn("Problem updating MessageStatus model \"{}\"", key, e); } }); } @Autowired  UpdateModel(@Value("#{store}") Map<String, KafkaRoad> store, PatchSetEmitter emitter, KafkaAdminClient adminClient); @Scheduled(initialDelayString = "${messageStatus.initialdelay:PT60m}", fixedRateString = "${messageStatus.fixedrate:PT60m}") void updateMessageStatusInModel();  }