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();  }
@Test public void operations_are_passed_to_emitter() throws Exception { MessageStatus status = new MessageStatus(15134544543L, 100); given(adminClient.updateMessageStatus(testRoad)).willReturn(singletonList(add("/messageStatus", status))); messageStatusUpdator.updateMessageStatusInModel(); then(emitter).should().emit(new PatchSet("test_road", singletonList(add("/messageStatus", status)))); }
@Test public void no_operations_means_no_PatchSet() throws Exception { given(adminClient.updateMessageStatus(testRoad)).willReturn(emptyList()); messageStatusUpdator.updateMessageStatusInModel(); then(emitter).shouldHaveZeroInteractions(); }