@Test public void testEventStream() { ExtTestSubscriber<SupervisorEvent> eventSubscriber = new ExtTestSubscriber<>(); supervisorOperations.events().subscribe(eventSubscriber); List<MasterInstance> initialInstances = new MutableDataGenerator<>( masterInstances(MasterState.Starting, "id1", "id2", "id3") ).getValues(3); for (int i = 0; i < 3; i++) { masterMonitorSubject.onNext(initialInstances.subList(0, i + 1)); expectMasterInstanceUpdateEvent(eventSubscriber, initialInstances.get(i)); } List<MasterInstance> firstUpdated = asList( MasterInstanceGenerator.moveTo(initialInstances.get(0), MasterState.NonLeader), initialInstances.get(1), initialInstances.get(2) ); masterMonitorSubject.onNext(firstUpdated); expectMasterInstanceUpdateEvent(eventSubscriber, firstUpdated.get(0)); masterMonitorSubject.onNext(firstUpdated.subList(0, 2)); expectMasterInstanceRemovedEvent(eventSubscriber, firstUpdated.get(2)); }
@Override public Observable<SupervisorEvent> events() { return masterMonitor.observeMasters() .compose(ObservableExt.mapWithState(Collections.emptyMap(), this::buildEventList)) .flatMap(Observable::from); }
DefaultSupervisorOperations implements SupervisorOperations { @Override public Observable<SupervisorEvent> events() { return masterMonitor.observeMasters() .compose(ObservableExt.mapWithState(Collections.emptyMap(), this::buildEventList)) .flatMap(Observable::from); } }
DefaultSupervisorOperations implements SupervisorOperations { @Override public Observable<SupervisorEvent> events() { return masterMonitor.observeMasters() .compose(ObservableExt.mapWithState(Collections.emptyMap(), this::buildEventList)) .flatMap(Observable::from); } @Inject  DefaultSupervisorOperations(LifecycleManager lifecycleManager,
                                       MasterMonitor masterMonitor,
                                       LeaderActivator leaderActivator,
                                       TitusRuntime titusRuntime); }
DefaultSupervisorOperations implements SupervisorOperations { @Override public Observable<SupervisorEvent> events() { return masterMonitor.observeMasters() .compose(ObservableExt.mapWithState(Collections.emptyMap(), this::buildEventList)) .flatMap(Observable::from); } @Inject  DefaultSupervisorOperations(LifecycleManager lifecycleManager,
                                       MasterMonitor masterMonitor,
                                       LeaderActivator leaderActivator,
                                       TitusRuntime titusRuntime); @Override List<MasterInstance> getMasterInstances(); @Override Optional<MasterInstance> findMasterInstance(String instanceId); @Override MasterInstance getMasterInstance(String instanceId); @Override Optional<MasterInstance> findLeader(); @Override Observable<SupervisorEvent> events(); @Override void stopBeingLeader(CallMetadata callMetadata); }
DefaultSupervisorOperations implements SupervisorOperations { @Override public Observable<SupervisorEvent> events() { return masterMonitor.observeMasters() .compose(ObservableExt.mapWithState(Collections.emptyMap(), this::buildEventList)) .flatMap(Observable::from); } @Inject  DefaultSupervisorOperations(LifecycleManager lifecycleManager,
                                       MasterMonitor masterMonitor,
                                       LeaderActivator leaderActivator,
                                       TitusRuntime titusRuntime); @Override List<MasterInstance> getMasterInstances(); @Override Optional<MasterInstance> findMasterInstance(String instanceId); @Override MasterInstance getMasterInstance(String instanceId); @Override Optional<MasterInstance> findLeader(); @Override Observable<SupervisorEvent> events(); @Override void stopBeingLeader(CallMetadata callMetadata);  }