DefaultHeartBeatService extends AbstractHeartBeatService<DateTime> { @Override protected boolean isAlive(HeartBeatPayLoad<DateTime> payLoad) { return payLoad != null && (DateTime.now().getMillis() - payLoad.getHeartBeatPayLoad().getMillis()) < MAX_HEARTBEAT_DELAY; } DefaultHeartBeatService(Collection<HeartBeatStreamingService> heartBeatStreamingServices);   }
@Test public void heartBeatTest() throws Exception { HeartBeatStreamingService heartBeatStreamingService = mock(HeartBeatStreamingService.class); when(heartBeatStreamingService.getHeartBeatSourceId()).thenReturn(TESTSTREAM); DefaultHeartBeatService service = new DefaultHeartBeatService(Lists.newArrayList(heartBeatStreamingService)); service.warmUpTime = 1L; service.init(); EventBus eventBus = new EventBus(); eventBus.register(service); HeartBeatCallback<DateTime> heartBeatCallBack = new HeartBeatCallbackImpl<DateTime>(eventBus); DateTime now = DateTime.now(); HeartBeatPayLoad<DateTime> payload = new HeartBeatPayLoad<DateTime>(now.minusMinutes(2), TESTSTREAM); heartBeatCallBack.onHeartBeat(payload); verify(heartBeatStreamingService, times(1)).startHeartBeatStreaming(); heartBeatCallBack.onHeartBeat(new HeartBeatPayLoad<DateTime>(DateTime.now(), TESTSTREAM)); service.serviceUp = false; do { Thread.sleep(2L); } while (service.heartBeatsObserverThread.isAlive()); }