OperatorMulticast extends ConnectableObservable<R> { @Override public void connect(Action1<? super Subscription> connection) { synchronized (guard) { if (subscription != null) { connection.call(guardedSubscription); return; } else { final Subject<? super T, ? extends R> subject = subjectFactory.call(); subscription = new Subscriber<T>() { @Override public void onCompleted() { subject.onCompleted(); } @Override public void onError(Throwable e) { subject.onError(e); } @Override public void onNext(T args) { subject.onNext(args); } }; final AtomicReference<Subscription> gs = new AtomicReference<Subscription>(); gs.set(Subscriptions.create(new Action0() { @Override public void call() { Subscription s; synchronized (guard) { if ( guardedSubscription == gs.get()) { s = subscription; subscription = null; guardedSubscription = null; connectedSubject.set(null); } else return; } if (s != null) { s.unsubscribe(); } } })); guardedSubscription = gs.get(); for(Subscriber<? super R> s : waitingForConnect) { subject.unsafeSubscribe(s); } waitingForConnect.clear(); connectedSubject.set(subject); } } connection.call(guardedSubscription); Subscriber<T> sub; synchronized (guard) { sub = subscription; } if (sub != null) source.subscribe(sub); } OperatorMulticast(Observable<? extends T> source, final Func0<? extends Subject<? super T, ? extends R>> subjectFactory); private  OperatorMulticast(final Object guard, final AtomicReference<Subject<? super T, ? extends R>> connectedSubject, final List<Subscriber<? super R>> waitingForConnect, Observable<? extends T> source, final Func0<? extends Subject<? super T, ? extends R>> subjectFactory); @Override void connect(Action1<? super Subscription> connection);  }
@Test public void testMulticast() { Subject<String, String> source = PublishSubject.create(); ConnectableObservable<String> multicasted = new OperatorMulticast<String, String>(source, new PublishSubjectFactory()); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); multicasted.subscribe(observer); source.onNext("one"); source.onNext("two"); multicasted.connect(); source.onNext("three"); source.onNext("four"); source.onCompleted(); verify(observer, never()).onNext("one"); verify(observer, never()).onNext("two"); verify(observer, times(1)).onNext("three"); verify(observer, times(1)).onNext("four"); verify(observer, times(1)).onCompleted(); }
@Test public void testMulticastConnectTwice() { Subject<String, String> source = PublishSubject.create(); ConnectableObservable<String> multicasted = new OperatorMulticast<String, String>(source, new PublishSubjectFactory()); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); multicasted.subscribe(observer); source.onNext("one"); Subscription sub = multicasted.connect(); Subscription sub2 = multicasted.connect(); source.onNext("two"); source.onCompleted(); verify(observer, never()).onNext("one"); verify(observer, times(1)).onNext("two"); verify(observer, times(1)).onCompleted(); assertEquals(sub, sub2); }
@Test public void testMulticastDisconnect() { Subject<String, String> source = PublishSubject.create(); ConnectableObservable<String> multicasted = new OperatorMulticast<String, String>(source, new PublishSubjectFactory()); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); multicasted.subscribe(observer); source.onNext("one"); Subscription connection = multicasted.connect(); source.onNext("two"); connection.unsubscribe(); source.onNext("three"); multicasted.subscribe(observer); multicasted.connect(); source.onNext("four"); source.onCompleted(); verify(observer, never()).onNext("one"); verify(observer, times(1)).onNext("two"); verify(observer, never()).onNext("three"); verify(observer, times(1)).onNext("four"); verify(observer, times(1)).onCompleted(); }