MulticastEventStreamClient implements EventStreamClient { @Override public Observable<String> readServerSideEvents() { return serverSideEvents; } MulticastEventStreamClient(EventStreamClient targetClient, Scheduler scheduler); @Override Observable<String> readServerSideEvents();  }
@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!"); }
@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(); }
@Test @Ignore public void Should_wait_2_sec_before_trying_to_reconnect_when_there_is_a_connection_error() { TestScheduler scheduler = Schedulers.test(); AtomicBoolean firstRetry = new AtomicBoolean(true); Observable<String> source = Observable.create(subscriber -> { if (firstRetry.getAndSet(false)) { subscriber.onError(new Exception("shit happens")); } subscriber.onNext("Hello!"); }); EventStreamClient targetClient = mock(EventStreamClient.class); when(targetClient.readServerSideEvents()).thenReturn(source); MulticastEventStreamClient multicastEventStreamClient = new MulticastEventStreamClient(targetClient, scheduler); Observable<String> events = multicastEventStreamClient.readServerSideEvents(); TestSubscriber<String> subscriber = new TestSubscriber<>(); events.subscribe(subscriber); scheduler.advanceTimeBy(2, TimeUnit.SECONDS); assertThat(subscriber.getOnNextEvents()).hasSize(1).contains("Hello!"); }