StreamStatusMonitor implements AgentStatusMonitor { @Override public AgentStatus getStatus(String agentInstanceId) { AgentStatus agentStatus = instanceStatuses.get(agentInstanceId); if (agentStatus == null) { if (failOnMissingData) { throw AgentManagementException.agentNotFound(agentInstanceId); } AgentInstance agentInstance = agentManagementService.getAgentInstance(agentInstanceId); return AgentStatus.healthy(source, agentInstance, "No data recorded yet; assuming healthy", scheduler.now()); } return agentStatus; } StreamStatusMonitor(String source,
                               boolean failOnMissingData,
                               AgentManagementService agentManagementService,
                               Observable<AgentStatus> agentStatusObservable,
                               Registry registry,
                               Scheduler scheduler); void shutdown(); @Override AgentStatus getStatus(String agentInstanceId); @Override boolean isHealthy(String agentInstanceId); @Override Observable<AgentStatus> monitor();  }
@Test public void testStatusUpdatePropagation() { statusUpdateSubject.onNext(AgentStatus.healthy(MY_SOURCE, instance, "OK", testScheduler.now())); assertThat(testSubscriber.takeNext().getStatusCode()).isEqualTo(AgentStatusCode.Healthy); assertThat(monitor.getStatus(instance.getId()).getStatusCode()).isEqualTo(AgentStatusCode.Healthy); }
@Test public void testRemovedInstanceCleanup() { statusUpdateSubject.onNext(AgentStatus.healthy(MY_SOURCE, instance, "OK", testScheduler.now())); statusUpdateSubject.onNext(AgentStatus.terminated(MY_SOURCE, instance, "Terminated", testScheduler.now())); AgentStatus status = monitor.getStatus(instance.getId()); assertThat(status.getStatusCode()).isEqualTo(AgentStatusCode.Healthy); assertThat(status.getDescription()).contains("No data recorded yet"); }