LifecycleAgentStatusMonitor implements AgentStatusMonitor { @Override public Observable<AgentStatus> monitor() { return agentManagementService.events(false).flatMap(event -> { if (event instanceof AgentInstanceUpdateEvent) { AgentInstance agentInstance = ((AgentInstanceUpdateEvent) event).getAgentInstance(); knownInstances.put(agentInstance.getId(), agentInstance); return Observable.just(statusOf(agentInstance)); } if (event instanceof AgentInstanceRemovedEvent) { AgentInstance agentInstance = knownInstances.remove(((AgentInstanceRemovedEvent) event).getAgentInstanceId()); if (agentInstance != null) { return Observable.just(toTerminatedEvent(agentInstance)); } } return Observable.empty(); }); } @Inject  LifecycleAgentStatusMonitor(AgentManagementService agentManagementService, TitusRuntime titusRuntime); @Override AgentStatus getStatus(String agentInstanceId); @Override boolean isHealthy(String agentInstanceId); @Override Observable<AgentStatus> monitor();  }
@Test public void testEventStream() { PublishSubject<AgentEvent> eventSubject = PublishSubject.create(); when(agentManagementService.events(false)).thenReturn(eventSubject); ExtTestSubscriber<AgentStatus> testSubscriber = new ExtTestSubscriber<>(); monitor.monitor().subscribe(testSubscriber); eventSubject.onNext(new AgentInstanceUpdateEvent(STARTED_INSTANCE)); assertThat(testSubscriber.takeNext().getStatusCode()).isEqualTo(AgentStatusCode.Healthy); eventSubject.onNext(new AgentInstanceRemovedEvent(STARTED_INSTANCE.getId())); AgentStatus terminatedStatus = testSubscriber.takeNext(); assertThat(terminatedStatus.getStatusCode()).isEqualTo(AgentStatusCode.Terminated); assertThat(terminatedStatus.getAgentInstance().getId()).isEqualTo(STARTED_INSTANCE.getId()); assertThat(terminatedStatus.getAgentInstance().getLifecycleStatus().getState()).isEqualTo(InstanceLifecycleState.Stopped); }