@Test public void Should_transmit_events_from_the_target_client() { TestScheduler scheduler = Schedulers.test(); PublishSubject<String> subject = PublishSubject.create(); EventStreamClient targetClient = mock(EventStreamClient.class); when(targetClient.readServerSideEvents()).thenReturn(subject); MulticastEventStreamClient multicastEventStreamClient = new MulticastEventStreamClient(targetClient, scheduler); Observable<String> events = multicastEventStreamClient.readServerSideEvents(); TestSubscriber<String> subscriber = new TestSubscriber<>(); events.subscribe(subscriber); subject.onNext("Hello!"); assertThat(subscriber.getOnNextEvents()).hasSize(1).contains("Hello!"); }
@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();  }