@Test @Ignore public void Should_generate_only_one_subscription_side_effect_with_multiple_subscribers() { Observable<String> source = Observable.create(subscriber -> { subscriber.onNext("open"); }); TestScheduler scheduler = Schedulers.test(); EventStreamClient targetClient = mock(EventStreamClient.class); when(targetClient.readServerSideEvents()).thenReturn(source); MulticastEventStreamClient multicastEventStreamClient = new MulticastEventStreamClient(targetClient, scheduler); Observable<String> events = multicastEventStreamClient.readServerSideEvents(); TestSubscriber<String> subscriber1 = new TestSubscriber<>(); TestSubscriber<String> subscriber2 = new TestSubscriber<>(); events.subscribe(subscriber1); events.subscribe(subscriber2); assertThat(subscriber1.getOnNextEvents()).hasSize(1); assertThat(subscriber2.getOnNextEvents()).isEmpty(); }
@Override public Observable<String> readServerSideEvents() { return serverSideEvents; }
MulticastEventStreamClient implements EventStreamClient { @Override public Observable<String> readServerSideEvents() { return serverSideEvents; } }
MulticastEventStreamClient implements EventStreamClient { @Override public Observable<String> readServerSideEvents() { return serverSideEvents; } MulticastEventStreamClient(EventStreamClient targetClient, Scheduler scheduler); }
MulticastEventStreamClient implements EventStreamClient { @Override public Observable<String> readServerSideEvents() { return serverSideEvents; } MulticastEventStreamClient(EventStreamClient targetClient, Scheduler scheduler); @Override Observable<String> readServerSideEvents(); }
MulticastEventStreamClient implements EventStreamClient { @Override public Observable<String> readServerSideEvents() { return serverSideEvents; } MulticastEventStreamClient(EventStreamClient targetClient, Scheduler scheduler); @Override Observable<String> readServerSideEvents();  }