@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);
    // Initial set
    for (int i = 0; i < 3; i++) {
        masterMonitorSubject.onNext(initialInstances.subList(0, i + 1));
        expectMasterInstanceUpdateEvent(eventSubscriber, initialInstances.get(i));
    }
    // Change state of first instance
    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));
    // Remove last instance
    masterMonitorSubject.onNext(firstUpdated.subList(0, 2));
    expectMasterInstanceRemovedEvent(eventSubscriber, firstUpdated.get(2));
}